kxepal / python-astm

Python implementation of ASTM E1381/1394 data protocol.
http://python-astm.readthedocs.org/en/latest/
Other
52 stars 47 forks source link

How to simulator an ASTM server? #1

Open HamGuy opened 8 years ago

HamGuy commented 8 years ago

Hi, kepal, Sorry to trouble you. But I am wonder to know how to simulator an ASTM server with your code. I am working with a data reader(a lite vesion LIS ) for a custom's HITCHI 7600 Hitachi 7600 modular chemistry analyzer. Unfortunatly, I can't connect to the machine directly, we are not at the same city. And I wonder if I can silmulator an ASTM server send data as the machine dose to verify my ASTM client(wriitern with c#).

kxepal commented 8 years ago

Hi @HamGuy ! I'm clear remember all the details today, but it should be something like

import logging
from astm.asynclib import loop
from astm.server import Server, BaseRecordsDispatcher

# global astm logging channel configuration
log_root = logging.getLogger('astm')
handler = logging.StreamHandler()
handler.setFormatter(logging.Formatter(
    '[%(asctime)s] [%(name)s] [%(levelname)s] %(message)s'
))
log_root.addHandler(handler)
log_root.setLevel(logging.DEBUG)

# special logging configuration for server connection handler
log_conn = logging.getLogger('astm.server.conn')
handler = logging.StreamHandler()
handler.setFormatter(logging.Formatter(
    '[%(asctime)s] [%(name)s] [%(levelname)s] [%(host)s:%(port)s] %(message)s'
))
log_conn.addHandler(handler)
log_conn.setLevel(logging.DEBUG)
log_conn.propagate = False

class RecordsDispatcher(BaseRecordsDispatcher):

    log = logging.getLogger('astm.server.dispatcher')

    def on_header(self, record):
        self.log.info('Header: %r', record)

    def on_patient(self, record):
        self.log.info('Patient: %r', record)

    def on_order(self, record):
        self.log.info('Order: %r', record)

    def on_result(self, record):
        self.log.info('Result: %r', record)

    def on_comment(self, record):
        self.log.info('Comment: %r', record)

    def on_unknown(self, record):
        self.log.warn('Unknown record: %r', record)

    def on_terminator(self, record):
        self.log.info('Terminator: %r', record)

if __name__ == '__main__':
    server = Server('localhost', 15200, dispatcher=RecordsDispatcher, timeout=10)
    loop(timeout=1)

Just fill message handler methods with your logic about how to process incoming messages.

HamGuy commented 8 years ago

Wow... You must be the angel sent by the God.