Adding more utils for FOBTRegressionTests

Author: adrianoaru-nhs

classes/analyser.py

@@ -0,0 +1,50 @@
+from dataclasses import dataclass
+from typing import Optional
+import pandas as pd
+
+
+@dataclass
+class Analyser:
+    """
+    Data class representing an analyser.
+    """
+
+    analyser_id: Optional[int] = None
+    analyser_code: Optional[str] = None
+    hub_id: Optional[int] = None
+    analyser_type_id: Optional[int] = None
+    spoil_result_code: Optional[int] = None
+    tech_fail_result_code: Optional[int] = None
+    below_range_result_code: Optional[int] = None
+    above_range_result_code: Optional[int] = None
+
+    def __str__(self) -> str:
+        return (
+            f"Analyser [analyser_id={self.analyser_id}, analyser_code={self.analyser_code}, "
+            f"hub_id={self.hub_id}, spoil_result_code={self.spoil_result_code}, "
+            f"tech_fail_result_code={self.tech_fail_result_code}, "
+            f"below_range_result_code={self.below_range_result_code}, "
+            f"above_range_result_code={self.above_range_result_code}]"
+        )
+
+    @staticmethod
+    def from_dataframe_row(row: pd.Series) -> "Analyser":
+        """
+        Creates an Analyser object from a pandas DataFrame row containing analyser query results.
+
+        Args:
+            row (pd.Series): A row from a pandas DataFrame with columns:
+                - tk_analyser_id
+                - analyser_code
+                - hub_id
+                - tk_analyser_type_id
+
+        Returns:
+            Analyser: The constructed Analyser object.
+        """
+        return Analyser(
+            analyser_id=row.get("tk_analyser_id"),
+            analyser_code=row.get("analyser_code"),
+            hub_id=row.get("hub_id"),
+            analyser_type_id=row.get("tk_analyser_type_id"),
+        )

classes/analyser_result_code_type.py

@@ -0,0 +1,66 @@
+from enum import Enum
+from typing import Optional, Dict
+
+
+class AnalyserResultCodeType(Enum):
+    """
+    Enum for analyser result code types, mapped to valid value IDs and descriptions.
+    """
+
+    ABOVE_DILUTION_RANGE = (309030, "Above dilution range")
+    BELOW_MEASURING_RANGE = (309028, "Below measuring range")
+    SPOILT = (309026, "Spoilt")
+    TECHNICAL_FAIL = (309027, "Technical Fail")
+
+    def __init__(self, valid_value_id: int, description: str) -> None:
+        self._valid_value_id = valid_value_id
+        self._description = description
+
+    @property
+    def valid_value_id(self) -> int:
+        """
+        Returns the valid value ID for the analyser result code type.
+        """
+        return self._valid_value_id
+
+    @property
+    def description(self) -> str:
+        """
+        Returns the description for the analyser result code type.
+        """
+        return self._description
+
+    @classmethod
+    def by_description(cls, description: str) -> Optional["AnalyserResultCodeType"]:
+        """
+        Returns the enum member matching the given description (case-sensitive).
+        """
+        for member in cls:
+            if member.description == description:
+                return member
+        return None
+
+    @classmethod
+    def by_description_case_insensitive(
+        cls, description: str
+    ) -> Optional["AnalyserResultCodeType"]:
+        """
+        Returns the enum member matching the given description (case-insensitive).
+        """
+        desc_lower = description.lower()
+        for member in cls:
+            if member.description.lower() == desc_lower:
+                return member
+        return None
+
+    @classmethod
+    def by_valid_value_id(
+        cls, valid_value_id: int
+    ) -> Optional["AnalyserResultCodeType"]:
+        """
+        Returns the enum member matching the given valid value ID.
+        """
+        for member in cls:
+            if member.valid_value_id == valid_value_id:
+                return member
+        return None

classes/entities/kit_service_management_entity.py

@@ -0,0 +1,83 @@
+from dataclasses import dataclass
+from typing import Optional
+from decimal import Decimal
+from datetime import datetime
+from classes.kit_service_management_record import KitServiceManagementRecord
+
+
+@dataclass
+class KitServiceManagementEntity:
+    """
+    Data class representing a kit service management entity (KIT_QUEUE table).
+    """
+
+    device_id: Optional[str] = None
+    test_kit_type: Optional[str] = None
+    test_kit_name: Optional[str] = None
+    test_kit_status: Optional[str] = None
+    logged_by_hub: Optional[str] = None
+    date_time_logged: Optional[datetime] = None
+    test_result: Optional[Decimal] = None
+    calculated_result: Optional[str] = None
+    error_code: Optional[int] = None
+    analyser_code: Optional[str] = None
+    date_time_authorised: Optional[datetime] = None
+    authoriser_user_code: Optional[str] = None
+    datestamp: Optional[datetime] = None
+    bcss_error_id: Optional[int] = None
+    post_response: Optional[int] = None
+    post_attempts: Optional[int] = None
+    put_response: Optional[int] = None
+    put_attempts: Optional[int] = None
+    date_time_error_archived: Optional[datetime] = None
+    error_archived_user_code: Optional[str] = None
+    issue_date: Optional[datetime] = None
+    issued_by_hub: Optional[str] = None
+    nhs_number: Optional[str] = None
+    analyser_error_description: Optional[str] = None
+    error_type: Optional[str] = None
+    screening_test_result: Optional[str] = None
+
+    @staticmethod
+    def from_record(
+        record: "KitServiceManagementRecord",
+    ) -> "KitServiceManagementEntity":
+        """
+        Converts a KitServiceManagementRecord object into a KitServiceManagementEntity object.
+
+        Args:
+            record (KitServiceManagementRecord): The record to convert.
+
+        Returns:
+            KitServiceManagementEntity: The converted entity.
+        """
+        return KitServiceManagementEntity(
+            device_id=record.device_id,
+            test_kit_type=(
+                str(record.test_kit_type) if record.test_kit_type is not None else None
+            ),
+            test_kit_name=record.test_kit_name,
+            test_kit_status=record.test_kit_status,
+            logged_by_hub=record.logged_by_hub,
+            date_time_logged=record.date_time_logged,
+            test_result=record.test_result,
+            calculated_result=record.calculated_result,
+            error_code=record.error_code,
+            analyser_code=record.analyser_code,
+            date_time_authorised=record.date_time_authorised,
+            authoriser_user_code=record.authoriser_user_code,
+            datestamp=record.datestamp,
+            bcss_error_id=record.bcss_error_id,
+            post_response=record.post_response,
+            post_attempts=record.post_attempts,
+            put_response=record.put_response,
+            put_attempts=record.put_attempts,
+            date_time_error_archived=record.date_time_error_archived,
+            error_archived_user_code=record.error_archived_user_code,
+            issue_date=record.date_time_issued,
+            issued_by_hub=record.issued_by_hub,
+            nhs_number=record.nhs_number,
+            analyser_error_description=record.analyser_error_description,
+            error_type=record.error_type,
+            screening_test_result=record.screening_test_result,
+        )

classes/hub_type.py

@@ -0,0 +1,14 @@
+from enum import Enum
+
+
+class HubType(Enum):
+    """
+    Enum representing hub types.
+    """
+
+    BCS01 = "BCS01"
+    BCS02 = "BCS02"
+    BCS03 = "BCS03"
+    BCS04 = "BCS04"
+    BCS05 = "BCS05"
+    BCS099 = "BCS099"

classes/kit_service_management_record.py

@@ -0,0 +1,119 @@
+from dataclasses import dataclass
+from typing import Optional
+from decimal import Decimal
+from datetime import datetime
+from utils.date_time_utils import DateTimeUtils
+
+
+@dataclass
+class KitServiceManagementRecord:
+    """
+    Data class representing a kit service management record.
+    """
+
+    # Kit queue data items
+    device_id: Optional[str] = None
+    test_kit_type: Optional[int] = None
+    test_kit_name: Optional[str] = None
+    test_kit_status: Optional[str] = None
+    logged_by_hub: Optional[str] = None
+    date_time_logged: Optional[datetime] = None
+    test_result: Optional[Decimal] = None
+    calculated_result: Optional[str] = None
+    error_code: Optional[int] = None
+    analyser_code: Optional[str] = None
+    date_time_authorised: Optional[datetime] = None
+    authoriser_user_code: Optional[str] = None
+    datestamp: Optional[datetime] = None
+    bcss_error_id: Optional[int] = None
+    post_response: Optional[int] = None
+    post_attempts: Optional[int] = None
+    put_response: Optional[int] = None
+    put_attempts: Optional[int] = None
+    date_time_error_archived: Optional[datetime] = None
+    error_archived_user_code: Optional[str] = None
+
+    # Kit data items
+    date_time_issued: Optional[datetime] = None
+    issued_by_hub: Optional[str] = None
+    nhs_number: Optional[str] = None
+    analyser_error_description: Optional[str] = None
+    error_type: Optional[str] = None
+    screening_test_result: Optional[str] = None
+
+    def __str__(self) -> str:
+        return (
+            f"KitServiceManagementRecord [device_id={self.device_id}, test_kit_type={self.test_kit_type}, "
+            f"test_kit_name={self.test_kit_name}, test_kit_status={self.test_kit_status}, "
+            f"logged_by_hub={self.logged_by_hub}, date_time_logged={self.date_time_logged}, "
+            f"test_result={self.test_result}, calculated_result={self.calculated_result}, "
+            f"error_code={self.error_code}, analyser_code={self.analyser_code}, "
+            f"date_time_authorised={self.date_time_authorised}, authoriser_user_code={self.authoriser_user_code}, "
+            f"datestamp={self.datestamp}, bcss_error_id={self.bcss_error_id}, post_response={self.post_response}, "
+            f"post_attempts={self.post_attempts}, put_response={self.put_response}, put_attempts={self.put_attempts}, "
+            f"date_time_error_archived={self.date_time_error_archived}, error_archived_user_code={self.error_archived_user_code}, "
+            f"date_time_issued={self.date_time_issued}, issued_by_hub={self.issued_by_hub}, nhs_number={self.nhs_number}, "
+            f"analyser_error_description={self.analyser_error_description}, error_type={self.error_type}, "
+            f"screening_test_result={self.screening_test_result}]"
+        )
+
+    @staticmethod
+    def from_dataframe_row(row) -> "KitServiceManagementRecord":
+        """
+        Creates a KitServiceManagementRecord object from a pandas DataFrame row containing kit service management query results.
+
+        Args:
+            row (pd.Series): A row from a pandas DataFrame with columns matching the query.
+
+        Returns:
+            KitServiceManagementRecord: The constructed object.
+        """
+
+        def parse_enum(enum_cls, value):
+            if value is None:
+                return None
+            try:
+                return enum_cls(value)
+            except ValueError:
+                return None
+
+        def parse_decimal(value):
+            if value is None or value == "":
+                return None
+            try:
+                return Decimal(str(value))
+            except Exception:
+                return None
+
+        return KitServiceManagementRecord(
+            device_id=row.get("device_id"),
+            test_kit_type=row.get("test_kit_type"),
+            test_kit_name=row.get("test_kit_name"),
+            test_kit_status=row.get("test_kit_status"),
+            logged_by_hub=row.get("logged_by_hub"),
+            date_time_logged=DateTimeUtils.parse_datetime(row.get("date_time_logged")),
+            test_result=parse_decimal(row.get("test_result")),
+            calculated_result=row.get("calculated_result"),
+            error_code=row.get("error_code"),
+            analyser_code=row.get("analyser_code"),
+            date_time_authorised=DateTimeUtils.parse_datetime(
+                row.get("date_time_authorised")
+            ),
+            authoriser_user_code=row.get("authoriser_user_code"),
+            datestamp=DateTimeUtils.parse_datetime(row.get("datestamp")),
+            bcss_error_id=row.get("bcss_error_id"),
+            post_response=row.get("post_response"),
+            post_attempts=row.get("post_attempts"),
+            put_response=row.get("put_response"),
+            put_attempts=row.get("put_attempts"),
+            date_time_error_archived=DateTimeUtils.parse_datetime(
+                row.get("date_time_error_archived")
+            ),
+            error_archived_user_code=row.get("error_archived_user_code"),
+            date_time_issued=DateTimeUtils.parse_datetime(row.get("issue_date")),
+            issued_by_hub=row.get("issued_by_hub"),
+            nhs_number=row.get("subject_nhs_number"),
+            analyser_error_description=row.get("analyser_error_description"),
+            error_type=row.get("error_type"),
+            screening_test_result=row.get("test_results"),
+        )

classes/kit_status.py

@@ -0,0 +1,17 @@
+from enum import Enum
+
+
+class KitStatus(Enum):
+    """
+    Enum representing kit statuses.
+    """
+
+    NEW = "NEW"
+    LOGGED = "LOGGED"
+    POSTED = "POSTED"
+    AUTHORISED = "AUTHORISED"
+    BCSS_READY = "BCSS_READY"
+    BCSS_PROCESSING = "BCSS_PROCESSING"
+    BCSS_PROCESSED = "BCSS_PROCESSED"
+    COMPLETE = "COMPLETE"
+    ERROR = "ERROR"

classes/kit_type.py

@@ -0,0 +1,19 @@
+from enum import Enum
+
+
+class KitType(Enum):
+    """
+    Enum representing kit types.
+    """
+
+    ANY = 0
+    GFOBT = 1
+    FIT = 2
+    QC = 999
+
+    @property
+    def id(self) -> int:
+        """
+        Returns the kit type ID.
+        """
+        return self.value