Merge pull request #18 from senaite/dca-vantage

Add Siemens' DCA Vantage® Analyzer import schema

docs/changelog.rst

@@ -5,6 +5,7 @@ Changelog
 1.0.0 (unreleased)
 ------------------
 
+- #18 Add Siemens' DCA Vantage® Analyzer import schema
 - #17 Add Sysmex XP-100 import schema
 - #16 Add Sysmex XN-550 import schema
 - #15 Add Cobas C311 import schema

src/senaite/astm/instruments/dca_vantage.py

@@ -0,0 +1,180 @@
+# -*- coding: utf-8 -*-
+
+from senaite.astm import records
+from senaite.astm.fields import ComponentField
+from senaite.astm.fields import ConstantField
+from senaite.astm.fields import DateTimeField
+from senaite.astm.fields import NotUsedField
+from senaite.astm.fields import SetField
+from senaite.astm.fields import TextField
+from senaite.astm.mapping import Component
+
+VERSION = "1.0.0"
+
+# Siemens Healthcare Diagnostics, DCA Vantage® Analyzer, Host Computer
+# Communications Link, 17306 Rev. E 2012-06, 2012
+HEADER_RX = r".*(DCA VANTAGE|DCA Vantage)\^"
+
+PROCESSING_IDS = (
+    "P",  # P: Production (the message contains clinical results)
+    "D",  # D: Debugging (the instrument is in service mode)
+)
+
+ACTION_CODES = (
+    "Q",  # Q: when control is run, else unused
+)
+
+REPORT_TYPES = (
+    "F",  # F: Final
+    "C",  # C: Correction of previously transmitted results
+)
+
+RESULT_ABNORMAL_FLAGS = (
+    "<",  # <: Below instrument measurement range
+    ">",  # >: Above instrument measurement range
+    "H",  # H: Above patient reference range or expected range of a control
+    "L",  # L: Below patient reference range or expected range of a control
+)
+
+RESULT_STATUSES = (
+    "F",  # F: Final Result
+    "C",  # C: Correction of previously transmitted results
+)
+
+
+def get_metadata(wrapper):
+    """Additional metadata
+    :param wrapper: The wrapper instance
+    :returns: dictionary of additional metadata
+    """
+    return {
+        "version": VERSION,
+        "header_rx": HEADER_RX,
+    }
+
+
+def get_mapping():
+    """Returns the wrappers for this instrument
+    """
+    return {
+        "H": HeaderRecord,
+        "P": PatientRecord,
+        "O": OrderRecord,
+        "R": ResultRecord,
+        "C": CommentRecord,
+        "Q": RequestInformationRecord,
+        "M": ManufacturerInfoRecord,
+        "L": TerminatorRecord,
+    }
+
+
+class HeaderRecord(records.HeaderRecord):
+    """Message Header Record (H)
+    """
+    sender = ComponentField(
+        Component.build(
+            TextField(name="name", default="DCA VANTAGE"),
+            TextField(name="version"),
+            TextField(name="serial"),
+        ))
+
+    processing_id = SetField(values=PROCESSING_IDS)
+
+
+class PatientRecord(records.PatientRecord):
+    """Patient Information Record (P)
+    """
+    # Practice Assigned Patient ID. (This field is not sent in Service Mode 1.)
+    practice_id = TextField()
+
+    # Component Field: <last name>^<first name> (patient samples only,
+    # only if entered, not in Manufacturing Mode 1)(This field is not sent in
+    # Service Mode 1)
+    name = ComponentField(
+        Component.build(
+            TextField(name="last_name"),
+            TextField(name="first_name"),
+        )
+    )
+
+
+class OrderRecord(records.OrderRecord):
+    """Order Record (O)
+    """
+    sample_id = TextField(default="")
+    instrument = ComponentField(
+        Component.build(
+            # Patient results: 001 through 999 (sample sequence number),
+            # Reagent Lot Number
+            # “<sample sequence number>^<reagent lot number>”
+            TextField("sample_seq_num"),
+            TextField("reagent_lot_num"),
+        )
+    )
+    action_code = SetField(values=ACTION_CODES)
+    report_type = SetField(values=REPORT_TYPES)
+
+
+class CommentRecord(records.CommentRecord):
+    """Comment Record (C)
+    """
+    # I: Clinical instrument system
+    source = ConstantField(default="I")
+
+    # Comment Text
+    # 1-to-many record specific text strings, separated by a component
+    # delimiter. The structure of this record depends on the type of record
+    # processed before:
+    # a) After the Order record for patient tests, transmit GFR data and
+    #    Comment information (if entered)
+    #    e.g. <age>^<gender>^<race>^<creatinine input>^<gfr result>^<c1>^<c2>
+    # b) After HbA 1c results, transmit User Correction Slope, User Correction
+    #    Offset, Primary Reporting Unit, and eAG (when available and enabled).
+    #    The eAG reporting unit is either “mg/dL” or “mmol/L”.
+    #    e.g. 1.000^0.0 <units>^NGSP^<eAG-value><eAG-units>
+    # c) After Microalbumin and Creatinine results, transmit User Correction
+    #    Slope and Offset
+    #    e.g. C|1|I|1.000^0.0 <units>|G<CR>
+    # d) After the order record for controls, transmit Comment Information (if
+    #    entered,) (one for each comment entered) up to 3 comments).
+    #    e.g. C|1|I|<c1>^<c2>^<c3>G<CR>
+    # Therefore, we leave it as a NotUsedField to let the consumer deal with it
+    data = NotUsedField()
+
+    # G: General/Free text comment
+    ctype = ConstantField(default="G")
+
+
+class ResultRecord(records.ResultRecord):
+    """Record to transmit analytical data.
+    """
+    test = ComponentField(
+        Component.build(
+            TextField(name="_"),
+            TextField(name="__"),
+            TextField(name="___"),
+            TextField(name="parameter"),
+        )
+    )
+    value = TextField(default="")
+    units = TextField(default="")
+    references = TextField(default="")
+    abnormal_flag = SetField(values=RESULT_ABNORMAL_FLAGS)
+    status = SetField(values=RESULT_STATUSES)
+    operator = TextField()
+    started_at = DateTimeField()
+
+
+class RequestInformationRecord(records.RequestInformationRecord):
+    """Request information Record (Q)
+    """
+
+
+class ManufacturerInfoRecord(records.ManufacturerInfoRecord):
+    """Manufacturer Specific Records (M)
+    """
+
+
+class TerminatorRecord(records.TerminatorRecord):
+    """Message Termination Record (L)
+    """

src/senaite/astm/tests/data/dca_vantage.txt

@@ -0,0 +1 @@
+1H|\^&|||DCA VANTAGE^04.04.00.00^S067337|||||||P||20240820151746P|1|BU24R554O|1||660^0090||||||||||||||||||||||FR|1|^^^Alb|63.7|mg/L||||F|||20240820151030C|1|I|1.000^0.0 mg/L|GR|2|^^^Crt|230.8|mg/dL||||F|||20240820151030C|1|I|1.000^0.0 mg/dL|GR|3|^^^Ratio|27.6|mg/g||||F|||20240820151030L|1|N43

src/senaite/astm/tests/test_dca_vantage.py

@@ -0,0 +1,140 @@
+# -*- coding: utf-8 -*-
+
+from unittest.mock import MagicMock
+from unittest.mock import Mock
+
+from senaite.astm import codec
+from senaite.astm.constants import ACK
+from senaite.astm.constants import ENQ
+from senaite.astm.instruments import dca_vantage
+from senaite.astm.protocol import ASTMProtocol
+from senaite.astm.tests.base import ASTMTestBase
+from senaite.astm.wrapper import Wrapper
+
+
+class DCAVantage(ASTMTestBase):
+    """Test ASTM communication protocol for the SIEMENS DCA Vantage® Analyzer,
+    a multi-parameter, point-of-care analyzer for monitoring glycemic control
+    in people with diabetes and detecting early kidney disease.
+    """
+
+    async def asyncSetUp(self):
+        self.protocol = ASTMProtocol()
+
+        # read instrument file
+        path = self.get_instrument_file_path("dca_vantage.txt")
+        self.lines = self.read_file_lines(path)
+
+        # Mock transport and protocol objects
+        self.transport = self.get_mock_transport()
+        self.protocol.transport = self.transport
+        self.mapping = dca_vantage.get_mapping()
+
+    def get_mock_transport(self, ip="127.0.0.1", port=12345):
+        transport = MagicMock()
+        transport.get_extra_info = Mock(return_value=(ip, port))
+        transport.write = MagicMock()
+        return transport
+
+    def test_communication(self):
+        """Test common instrument communication """
+
+        # Establish the connection to build setup the environment
+        self.protocol.connection_made(self.transport)
+
+        # Send ENQ
+        self.protocol.data_received(ENQ)
+
+        for line in self.lines:
+            self.protocol.data_received(line)
+            # We expect an ACK as response
+            self.transport.write.assert_called_with(ACK)
+
+    def test_decode_messages(self):
+        self.test_communication()
+
+        data = {}
+        keys = []
+
+        for line in self.protocol.messages:
+            records = codec.decode(line)
+
+            self.assertTrue(isinstance(records, list), True)
+            self.assertTrue(len(records) > 0, True)
+
+            record = records[0]
+            rtype = record[0]
+            wrapper = self.mapping[rtype](*record)
+            data[rtype] = wrapper.to_dict()
+            keys.append(rtype)
+
+        for key in keys:
+            self.assertTrue(key in data)
+
+    def test_dca_vantage_header_record(self):
+        """Test the Header Record wrapper
+        """
+        wrapper = Wrapper(self.lines)
+        data = wrapper.to_dict()
+        record = data["H"][0]
+
+        # test model
+        self.assertEqual(record["sender"]["name"], "DCA VANTAGE")
+
+        # test software version
+        self.assertEqual(record["sender"]["version"], "04.04.00.00")
+
+        # test serial number
+        self.assertEqual(record["sender"]["serial"], "S067337")
+
+        # test processing id
+        self.assertEqual(record["processing_id"], "P")
+
+    def test_dca_vantage_order_record(self):
+        """Test the Order Record wrapper
+        """
+        wrapper = Wrapper(self.lines)
+        data = wrapper.to_dict()
+        record = data["O"][0]
+
+        # test sample sequence number
+        self.assertEqual(record["instrument"]["sample_seq_num"], "660"),
+
+        # test reagent lot number
+        self.assertEqual(record["instrument"]["reagent_lot_num"], "0090"),
+
+        # test action code
+        self.assertEqual(record["action_code"], None)
+
+        # test report type
+        self.assertEqual(record["report_type"], "F")
+
+    def test_dca_vantage_result_records(self):
+        """Test the Result Record wrapper
+        """
+        wrapper = Wrapper(self.lines)
+        data = wrapper.to_dict()
+        records = data["R"]
+
+        # we should have 3 results
+        self.assertEqual(len(records), 3)
+
+        # test parameter name
+        tests = ["Alb", "Crt", "Ratio"]
+        for idx, record in enumerate(records):
+            self.assertEqual(record["test"]["parameter"], tests[idx])
+
+        # test results
+        results = ["63.7", "230.8", "27.6"]
+        for idx, record in enumerate(records):
+            self.assertEqual(record.get("value"), results[idx])
+
+        # test units
+        units = ["mg/L", "mg/dL", "mg/g"]
+        for idx, record in enumerate(records):
+            self.assertEqual(record.get("units"), units[idx])
+
+        # test status
+        statuses = ["F", "F", "F"]
+        for idx, record in enumerate(records):
+            self.assertEqual(record.get("status"), statuses[idx])