EDIFACT in healthcare tutorial - Receive lab results and send letters

# -*- coding: utf-8 -*-

# Zato
# Importing a dialect package registers its message classes -
# here, the Dutch healthcare dialect shipped as an example.
import zato.edifact.nl

from zato.edifact import parse_edifact
from zato.server.service import Service

# #####################################################################
# #####################################################################

class LabResults(Service):
    """ Receives laboratory results sent as EDIFACT interchanges.
    """

    def handle(self) -> 'None':

        # The wire text, as the channel delivered it ..
        raw = self.request.input

        # .. one call parses the envelope and the message in it ..
        interchange = parse_edifact(raw)
        msg = interchange.message

        # .. the UNB header names the sending laboratory ..
        sender = interchange.header.sender.identification
        message_type = msg.unh.identifier.message_type

        self.logger.info(f'Received {message_type} from {sender}')

        # .. and each specimen carries its determinations.
        for material in msg.materials:
            for determination in material.determinations:
                value = f'{determination.result} {determination.unit}'
                self.logger.info(f'{determination.determination}: {value}')

msg = interchange.message

# Composites have typed components
family_name = msg.pid.patient_name.married_name   # 'Dijk'
born_year   = msg.pid.date_of_birth.year          # '1975'

# The sending laboratory and its department
lab        = msg.hospital.institution_name        # 'Streeklab Rijnmond'
department = msg.department.department            # 'Klinische chemie'

for material in msg.materials:
    for determination in material.determinations:

        # The laboratory flagged this value as above range
        if determination.normality_indicator == 'H':
            name  = determination.determination
            value = f'{determination.result} {determination.unit}'
            limit = determination.upper_limit
            self.logger.info(f'Above range: {name} {value}, upper limit {limit}')

class Router(Service):
    """ Passes each message of an interchange to its handler.
    """

    def handle(self) -> 'None':

        # Message types and the services that handle them ..
        handlers = {
            'MEDLAB': 'edifact-api.lab-results',
            'MEDVRI': 'edifact-api.letters',
        }

        # .. one interchange may carry several messages ..
        interchange = parse_edifact(self.request.input)

        for msg in interchange.messages:

            # .. the UNH identity decides where each one goes ..
            message_type = msg.unh.identifier.message_type

            if service_name := handlers.get(message_type):
                _ = self.invoke(service_name, msg=msg)

            # .. and unknown types are still parsed into fully navigable messages.
            else:
                self.logger.info(f'Skipping {message_type}')

interchange = parse_edifact(raw)

# The wire text, reproduced byte for byte
wire_text = interchange.serialize()

# One assignment, one changed segment - the rest stays as received
msg = interchange.message
msg.pid.sex = 'M'
modified = interchange.serialize()

# Zato
from zato.edifact.nl import MEDVRI
from zato.edifact.nl.segments import TXT

# #####################################################################
# #####################################################################

class SendLetter(Service):
    """ Builds a letter to a GP and sends it to the mailbox network.
    """

    def handle(self) -> 'None':

        # A new, empty letter ..
        letter = MEDVRI()

        # .. its UNH header carries the reference and the message identity ..
        letter.unh.reference_number = '8001'
        letter.unh.identifier.message_type = 'MEDVRI'
        letter.unh.identifier.version = '1'

        # .. the sending practitioner and institution ..
        letter.sender.person_name      = 'E. Vermeer'
        letter.sender.institution_name = 'GGZ De Linde'

        # .. the date the letter was written ..
        letter.det.date.year  = '26'
        letter.det.date.month = '08'
        letter.det.date.day   = '31'

        # .. the patient it concerns ..
        letter.pid.date_of_birth.year  = '1975'
        letter.pid.date_of_birth.month = '03'
        letter.pid.date_of_birth.day   = '18'
        letter.pid.sex = 'V'
        letter.pid.patient_name.married_name        = 'Dijk'
        letter.pid.patient_name.married_name_prefix = 'van der'

        # .. the letter's text, one TXT segment per line ..
        line_1 = TXT()
        line_1.text = 'Dear colleague'

        line_2 = TXT()
        line_2.text = 'Your patient has been in our care for six months now'

        letter.text = [line_1, line_2]

        # .. the UNT trailer closes the message ..
        letter.unt.segment_count    = '7'
        letter.unt.reference_number = '8001'

        # .. serialization produces the wire text ..
        wire_text = letter.serialize()

        # .. the connection created a moment ago is looked up by name ..
        conn = self.rest['EDI Mailbox']

        # .. and one POST delivers the letter to the mailbox.
        _ = conn.post(self.cid, wire_text)

        self.logger.info(f'Sent letter {letter.unh.reference_number}')

interchange = parse_edifact(raw)

data      = interchange.to_dict()
json_text = interchange.to_json(indent=2)

# Skip empty elements for compact output
compact = interchange.to_json(include_empty=False)