Vetera finalization

orthanc_tools/hl7Lib/__init__.py

@@ -10,3 +10,5 @@
 from .hl7_oru_report_msg_handler import Hl7OruReportMsgHandler
 from .hl7_report_series_builder import ReportSeriesBuilder
 from .hl7_report_parser import Hl7ReportParser
+from .hl7_worklist_parser_assistovet import Hl7WorklistParserAssistovet
+from .hl7_worklist_parser_vetera import Hl7WorklistParserVetera

orthanc_tools/hl7Lib/hl7_dicom_worklist_builder.py

@@ -73,7 +73,8 @@ def generate(self, values: typing.Dict[str, str], file_name: str = None) -> str:
                                         ('PatientSpeciesDescription', DicomElementType.OPTIONAL),
                                         ('PatientBreedDescription', DicomElementType.OPTIONAL),
                                         ('ResponsiblePerson', DicomElementType.OPTIONAL),
-                                        ('PatientSexNeutered', DicomElementType.OPTIONAL)
+                                        ('PatientSexNeutered', DicomElementType.OPTIONAL),
+                                        ('BreedRegistrationNumber', DicomElementType.OPTIONAL)
                           ]:
             self._add_field(ds, values, field_name, element_type)
 

orthanc_tools/hl7Lib/hl7_worklist_parser.py

@@ -133,13 +133,5 @@ def parse(self, hl7_message: str) -> typing.Dict:
         elif values.get('_consultingDoctorPV1') is not None:
             values['RequestingPhysician'] = values.get('_consultingDoctorPV1')
 
-        # specific to AssistoVet, this is the first Vet HL7 messages provider integrated, so this could elvove in the future:
-        if values.get('PatientSpeciesDescription') is not None and values.get('PatientBreedDescription') is not None:
-            # this should be an animal, so let's fill the responsible name
-            last_name = self._get('PID.F5.R1.C1', default_value="")
-            first_name = values.get('PatientMotherBirthName')
-            if first_name is None: first_name = ""
-            values['ResponsiblePerson'] = '^'.join([last_name, first_name])
-
         return values
 

orthanc_tools/hl7Lib/hl7_worklist_parser_assistovet.py

@@ -0,0 +1,21 @@
+from .hl7_worklist_parser import Hl7WorklistParser
+import typing
+
+
+class Hl7WorklistParserAssistovet(Hl7WorklistParser):
+
+    def __init__(self, specific_fields: dict = None, patient_name_components_count: int = 5):
+        super(Hl7WorklistParserAssistovet, self).__init__(specific_fields, patient_name_components_count)
+
+    def parse(self, hl7_message: str) -> typing.Dict:
+        # let's parse the default fields
+        values = Hl7WorklistParser.parse(self, hl7_message=hl7_message)
+
+        # let's add the field specific to Assistovet
+        if values.get('PatientSpeciesDescription') is not None and values.get('PatientBreedDescription') is not None:
+            # this should be an animal, so let's fill the responsible name
+            last_name = self._get('PID.F5.R1.C1', default_value="")
+            first_name = values.get('PatientMotherBirthName')
+            if first_name is None: first_name = ""
+            values['ResponsiblePerson'] = '^'.join([last_name, first_name])
+        return values

orthanc_tools/hl7Lib/hl7_worklist_parser_vetera.py

@@ -0,0 +1,47 @@
+from .hl7_worklist_parser import Hl7WorklistParser
+import typing
+
+
+class Hl7WorklistParserVetera(Hl7WorklistParser):
+
+    def __init__(self, specific_fields: dict = None, patient_name_components_count: int = 5):
+
+        vetera_dict = {
+            'PatientID': 'PID.F2',
+            'PatientSpeciesDescription': 'PID.F35',
+            'PatientBreedDescription': 'PID.F36',
+            'PatientSexNeutered': 'PID.F37',
+            'BreedRegistrationNumber': 'PID.F38', # TO BE CONFIRMED (waiting for Vetera's feeback)
+            '_scheduledProcedureStepStartDateTime': 'OBR.F6',
+            'Modality': 'OBR.F21',
+            'RequestingPhysician': 'OBR.F32'
+        }
+        if specific_fields is not None:
+            vetera_dict.update(specific_fields)
+
+        super(Hl7WorklistParserVetera, self).__init__(vetera_dict, patient_name_components_count)
+
+
+
+    def parse(self, hl7_message: str) -> typing.Dict:
+        # let's parse the default fields
+        values = Hl7WorklistParser.parse(self, hl7_message=hl7_message)
+
+        # let's parse according to rules specific to Vetera
+        owner = self._get('PID.F5.R1.C1', default_value="")
+        name = self._get('PID.F5.R1.C2', default_value="")
+
+        # TODO confirm with the customer that this is what they want (because the owner name won't appear in Orthanc UI)
+        values['ResponsiblePerson'] = owner
+        values['PatientName'] = name
+
+        if values.get('_scheduledProcedureStepStartDateTime') is not None:
+            datetimeString = values.get('_scheduledProcedureStepStartDateTime')
+            values['ScheduledProcedureStepStartDate'] = datetimeString[:8]  # date is made of the 8 first chars of the string
+            if len(datetimeString) == 12:
+                values['ScheduledProcedureStepStartTime'] = datetimeString[8:12] + "00"
+            elif len(datetimeString) == 14:
+                values['ScheduledProcedureStepStartTime'] = datetimeString[8:14]
+
+        return values
+

orthanc_tools/hl7Lib/tests/test_hl7_orm_worklist_msg_handler.py

@@ -2,9 +2,10 @@
 import hl7  # https://python-hl7.readthedocs.org/en/latest/
 import pydicom
 from pydicom import dcmread
-from orthanc_tools import MLLPClient, DicomWorklistBuilder, Hl7WorklistParser, Hl7MessageValidator, MLLPServer, Hl7OrmWorklistMsgHandler
+from orthanc_tools import MLLPClient, DicomWorklistBuilder, Hl7WorklistParser, Hl7MessageValidator, MLLPServer, Hl7OrmWorklistMsgHandler, Hl7WorklistParserAssistovet, Hl7WorklistParserVetera
 import tempfile
 import logging
+import typing
 
 
 class TestHl7OrmWorklistMsgHandler(unittest.TestCase):
@@ -271,9 +272,10 @@ def test_isosl_worklists(self):
 
     def test_assistovet_worklists(self):
         port_number = 2007  # there are currently some issues when trying to reuse the same port in 2 tests (it's probably not freed soon enough -> let's use another port for each test)
+
         with tempfile.TemporaryDirectory() as temporary_dir:
 
-            parser = Hl7WorklistParser()
+            parser = Hl7WorklistParserAssistovet()
             builder = DicomWorklistBuilder(folder = temporary_dir)
             orm_handler = Hl7OrmWorklistMsgHandler(parser=parser, builder=builder)
 
@@ -330,4 +332,60 @@ def test_assistovet_worklists(self):
                     self.assertEqual("114936",  wl.ScheduledProcedureStepSequence[0].ScheduledProcedureStepStartTime)
 
 
+    def test_vetera_worklists(self):
+        port_number = 2008  # there are currently some issues when trying to reuse the same port in 2 tests (it's probably not freed soon enough -> let's use another port for each test)
+
+        with tempfile.TemporaryDirectory() as temporary_dir:
+
+            parser = Hl7WorklistParserVetera()
+            builder = DicomWorklistBuilder(folder = temporary_dir)
+            orm_handler = Hl7OrmWorklistMsgHandler(parser=parser, builder=builder)
+
+            mllp_server = MLLPServer(
+                    host='localhost',
+                    port=port_number,
+                    handlers={
+                    'ORM^O01': (orm_handler.handle_orm_message,)
+                    }
+            )
+
+            with mllp_server as server:
+                # validate that ADT messages do create worklist files
+                with MLLPClient('localhost', port_number) as client:
+                    hl7_request = hl7.parse(
+                        "\x0bMSH|^~\&|VETERA|VETERA|conquest|conquest|20170731081517||ORM^O01|1000000001|P|2.5.0|||||\r"
+                        "PID|1|999888777||123456789012345|GP.Software^Vetera||20070501|F|||||||||||||||||||||||||||Katze|Balinese|ALTERED|ZH-123|\r"
+                        "ORC|NW||||||||20170731081517||||||||||\r"
+                        "OBR|||1000000001|HD||20170731081517|||||||||||||||DX|||ZUG||||||||Dr. P. Muster||||\r"
+                        "\x1c\x0d"
+                    )
+                    response = client.send(hl7_request)
+                    hl7Response = hl7.parse(response)
+
+                # make sure a file has been created
+                files = glob.glob('{path}/*.wl'.format(path = temporary_dir))
+                self.assertEqual(1, len(files))
+                worklist_file_path = files[0]
+
+                #TODO:
+                # check the content of the file
+                with dcmread(worklist_file_path) as wl:
+                    self.assertEqual("Vetera", wl.PatientName)
+                    self.assertEqual("GP.Software", wl.ResponsiblePerson)
+                    self.assertEqual("20070501", wl.PatientBirthDate)
+                    self.assertEqual("999888777", wl.PatientID)
+                    self.assertEqual("ISO_IR 100", wl.SpecificCharacterSet)  # default char set if not specified in HL7 message
+
+                    # make sure all 'mandatory' fields are there
+                    self.assertEqual("123456789012345", wl.OtherPatientIDs)
+                    self.assertEqual("Katze", wl.PatientSpeciesDescription)
+                    self.assertEqual("Balinese", wl.PatientBreedDescription)
+                    self.assertEqual("ALTERED", wl.PatientSexNeutered)
+                    self.assertEqual("ZH-123", wl.BreedRegistrationNumber)
+                    self.assertEqual("20170731",  wl.ScheduledProcedureStepSequence[0].ScheduledProcedureStepStartDate)
+                    self.assertEqual("081517",  wl.ScheduledProcedureStepSequence[0].ScheduledProcedureStepStartTime)
+                    self.assertEqual("1000000001", wl.AccessionNumber)
+                    self.assertEqual("F", wl.PatientSex)
+                    self.assertEqual("DX", wl.ScheduledProcedureStepSequence[0].Modality)
+                    self.assertEqual("Dr. P. Muster", wl.RequestingPhysician)