St/queries to snakemake

.gitignore

@@ -11,6 +11,7 @@ wheels/
 
 # IDEs
 .idea/
+.vscode/
 
 # settings files (should not be in the source tree anyway, but just in case)
 *.env

config.EXAMPLE/controller.env.EXAMPLE

@@ -13,6 +13,7 @@ RABBITMQ_PASSWORD="my_pw"
 RABBITMQ_HOST="localhost"
 RABBITMQ_PORT=5672
 RABBITMQ_QUEUE="waveform"
+SQL_PATH="./src/sql/"
 # OpenTelemetry OTLP/HTTP endpoint of the LGTM collector.
 OTEL_EXPORTER_OTLP_ENDPOINT="http://lgtm:4318"
 OTEL_SERVICE_NAME=waveform-controller

config.EXAMPLE/exporter.env.EXAMPLE

@@ -33,6 +33,33 @@ ONLY_USE_CSV_FROM_YESTERDAY=TRUE
 # expression to match multiple date
 PROCESS_CSV_FROM_DATE=
 
+# We query Caboodle to get electronic healthcare record date per patient per day
+CABOODLE_DBNAME="fakecab"
+CABOODLE_USERNAME="inform_user"
+CABOODLE_PASSWORD="inform"
+CABOODLE_HOST="localhost"
+CABOODLE_PORT="5433"
+CABOODLE_CONNECT_TIMEOUT="10"  # in seconds
+CABOODLE_QUERY_TIMEOUT="3000"  # in milliseconds
+
+# To avoid having to deploy a fake caboodle for testing we have 
+# a testing flag for Caboodle. If set TRUE caboodle connection will
+# fail silently and ehr file will be created with fake data
+CABOODLE_TESTING="FALSE"
+
+# The following is duplicated from controller.env
+# the exporter needs access to uds
+UDS_DBNAME="fakeuds"
+UDS_USERNAME="inform_user"
+UDS_PASSWORD="inform"
+UDS_HOST="172.17.0.1"
+UDS_PORT="5433"
+UDS_CONNECT_TIMEOUT="10"
+UDS_QUERY_TIMEOUT="3000"
+SCHEMA_NAME="schemaname"
+
+SQL_PATH="/app/src/sql/"
+
 # OpenTelemetry OTLP/HTTP endpoint of the LGTM collector.
 OTEL_EXPORTER_OTLP_ENDPOINT="http://lgtm:4318"
 OTEL_SERVICE_NAME=waveform-exporter

sql_scripts/README.md

@@ -0,0 +1,33 @@
+# Notes on putting together the EHR needed
+
+## Goal
+
+The ultimate aim is to have one csv per patient per day which looks roughly like
+
+ | DateTimeRecorded | Temperature | noradrenaline | etc  | Secretions | etc | Placementinstant | RemovalInstant | TubeSize | etc |Units | Comments |
+| --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- |
+| 08/08/2026 00:00:15 | 36.4 |  |  |  |  | | | | | |
+| 08/08/2026 00:00:16 |  | 1 | | | | | | | | mg/L |  |
+| 08/08/2026 00:00:17 |  |  |  | None |  | | | | | |
+| 08/08/2026 00:02:18 |  |  |  |  |  | 07/08/2026 | |8mm | | |
+| 08/08/2026 00:00:15 | 38.5 |  |  |  |  | | | | | | Doctors alerted |
+
+*Note: The insertion date for a tube may well be earlier than the day on which it is recorded as these seem to get populated during the nightly update to caboodle.*
+
+## Current scripts
+
+| script | arguments | record | location of script in repo | database | 
+|- | --- | --- |- | --- |
+| mrn_based_on_bed_and_datetime.sql | location string | csn |waveform-controller/src/sql | star |
+| get_hospital_visit_id.sql| csn | hospital_visit_id | waveform-controller/sql_scripts| star |
+| flow_sheet_values.sql| hospital_visit_id/today/yesterday | part of table above | waveform-controller/sql_scripts| star |
+| airway.sql | csn/today/yesterday | part of the table above | waveform-controller/sql_scripts | caboodle |
+| sputum_secretions.sql | csn/today/yesterday | part of the table above | waveform-controller/sql_scripts | caboodle |
+
+## Unfinished scripts
+
+lab_results.sql need dealing with in the same way as flow_sheet_values
+
+lab_test_names.sql forms part of the above query but is useful for exploring
+
+We need scripts for any of the items in the a tracker that have not yet been covered.

sql_scripts/lab_results.sql

@@ -0,0 +1,35 @@
+-- This selects the values of lab tests
+-- 1011      CRP
+-- 722790196 CRP
+-- 390793054 WCC
+-- 390793057 WCC
+-- 390793060 WCC
+
+SELECT
+    r.result_last_modified_datetime AS DateTimeRecorded,
+
+    MAX(r.value_as_real) FILTER (WHERE r.lab_test_definition_id = '1001') AS "C-reactive protein",
+    MAX(r.value_as_real) FILTER (WHERE r.lab_test_definition_id = '390793054') AS "CSF WCC TUBE 1",
+    MAX(r.value_as_real) FILTER (WHERE r.lab_test_definition_id = '390793057') AS "CSF WCC TUBE 2",
+    MAX(r.value_as_real) FILTER (WHERE r.lab_test_definition_id = '390793060') AS "CSF WCC TUBE 3",
+    MAX(r.value_as_real) FILTER (WHERE r.lab_test_definition_id = '722790196') AS "C-reactive protein"
+
+    r.units AS Units,
+    r.abnormal_flag AS Abnormal_result,
+    r.comment AS Comments
+
+FROM star.lab_result AS r
+LEFT JOIN star.lab_order AS o
+    ON r.lab_order_id = o.lab_order_id
+
+WHERE r.result_status  like 'FINAL'
+AND 
+r.lab_test_definition_id IN ('1001', 
+                             '390793054', 
+                             '390793057', 
+                             '390793060', 
+                             '722790196')
+AND vo.valid_from BETWEEN %(yesterday)s AND %(today)s
+AND o.hospital_visit_id = %(hospital_visit_id)s
+
+GROUP BY DateTimeRecorded, Units, Abnormal_result, Comments

sql_scripts/lab_test_names.sql

@@ -0,0 +1,5 @@
+select lab_test_definition_id as id,
+       name,
+       standardised_vocabulary as vocab
+from star.lab_test_definition as ltd
+where ltd.lab_test_definition_id in ('1001', '390793054', '390793057', '390793060', '722790196')

src/controller.py

@@ -108,7 +108,6 @@ def finalise_message(outcome: MessageOutcome):
 class WaveformController:
     def __init__(self):
         self.emap_db = db.starDB()
-        self.emap_db.init_query()
         self.emap_db.connect()
 
     def waveform_callback(
@@ -210,7 +209,9 @@ def outcome(
         )
         lookup_success = True
         try:
-            matched_mrn = self.emap_db.get_row(location_string, observation_time)
+            matched_mrn = self.emap_db.get_matched_mrn(
+                location_string, observation_time
+            )
         except ValueError:
             lookup_success = False
             logger.error(
@@ -220,6 +221,7 @@ def outcome(
                 exc_info=True,
             )
             matched_mrn = ("unmatched_mrn", "unmatched_nhs", "unmatched_csn", False)
+            # matched_mrn = ("1234568", "12345678", "12345678", False)
         except ConnectionError:
             logger.error("Database error, will try again", exc_info=True)
             return outcome("reject", reason="db_conn_err", requeue=True)

src/csv_writer.py

@@ -3,9 +3,16 @@
 import csv
 import json
 from datetime import datetime
+import pandas as pd
 from typing import Optional
 
-from locations import WAVEFORM_ORIGINAL_CSV, make_file_name, FILE_STEM_PATTERN
+from locations import (
+    WAVEFORM_ORIGINAL_CSV,
+    WAVEFORM_PSEUDONYMISED_EHR,
+    make_file_name,
+    FILE_STEM_PATTERN,
+    EHR_STEM_PATTERN_HASHED,
+)
 
 
 def create_file_name(
@@ -93,3 +100,22 @@ def write_frame(
         ]
 
         wv_writer.writerow(row_array)
+
+
+def write_ehr(
+    df: pd.DataFrame,
+    date_str: str,
+    hashed_csn: str,
+) -> bool:
+    """Writes a frame of electronic healthcare data to a csv file.
+
+    :return: True if write was successful.
+    """
+    subs_dict = dict(date=date_str, hashed_csn=hashed_csn)
+    stem = make_file_name(EHR_STEM_PATTERN_HASHED, subs_dict)
+    filename = WAVEFORM_PSEUDONYMISED_EHR / f"{stem}_ehr.csv"
+    filename.parent.mkdir(exist_ok=True, parents=True)
+
+    df.to_csv(filename, index=False)
+
+    return True

src/db.py

@@ -1,4 +1,5 @@
 from datetime import datetime
+import pandas as pd
 import psycopg2
 from psycopg2 import sql, pool
 import logging
@@ -10,7 +11,7 @@
 
 
 class starDB:
-    sql_query: str = ""
+    mrn_lookup_query: str = ""
     connection_string: str = "dbname={} user={} password={} host={} port={} connect_timeout={} options='-c statement_timeout={}'".format(
         settings.UDS_DBNAME,  # type:ignore
         settings.UDS_USERNAME,  # type:ignore
@@ -20,36 +21,159 @@ class starDB:
         settings.UDS_CONNECT_TIMEOUT,  # type:ignore
         settings.UDS_QUERY_TIMEOUT,  # type:ignore
     )
-    connection_pool: pool.ThreadedConnectionPool
+    connection_pool: pool.SimpleConnectionPool
+    fake_star: bool = False
 
-    def connect(self):
-        self.connection_pool = pool.SimpleConnectionPool(1, 1, self.connection_string)
+    def connect(self) -> None:
+        self.fake_star = True if settings.STARDB_TESTING == "TRUE" else False
+        if not self.fake_star:
+            self.connection_pool = pool.SimpleConnectionPool(
+                1, 1, self.connection_string
+            )
+
+    def _init_mrn_lookup_query(self) -> None:
+        with open(settings.SQL_PATH + "mrn_based_on_bed_and_datetime.sql", "r") as file:
+            self.mrn_lookup_query = sql.SQL(file.read())  # type:ignore
 
-    def init_query(self):
-        with open("src/sql/mrn_based_on_bed_and_datetime.sql", "r") as file:
-            self.sql_query = sql.SQL(file.read())
-        self.sql_query = self.sql_query.format(
+        self.mrn_lookup_query = self.mrn_lookup_query.format(
             schema_name=sql.Identifier(settings.SCHEMA_NAME)
         )
 
-    def get_row(self, location_string: str, observation_datetime: datetime):
+    def get_matched_mrn(
+        self, location_string: str, observation_datetime: datetime
+    ) -> pd.DataFrame:
         parameters = {
             "location_string": location_string,
             "observation_datetime": observation_datetime,
         }
+        if self.mrn_lookup_query == "":
+            self._init_mrn_lookup_query()
+
+        rows = self._get_rows(self.mrn_lookup_query, parameters)  # type: ignore
+
+        if len(rows) != 1:
+            raise ValueError(
+                f"Wrong number of rows returned from database. {len(rows)} != 1, for {location_string}:{observation_datetime}"
+            )
+
+        return rows[0]
+
+    def get_hospital_visit_from_csn(self, csn: str) -> int:
+        with open(settings.SQL_PATH + "get_hospital_visit_id.sql", "r") as file:
+            hv_query = sql.SQL(file.read())
+
+        hv_query = hv_query.format(schema_name=sql.Identifier(settings.SCHEMA_NAME))  # type: ignore
+
+        parameters = {
+            "csn": csn,
+        }
+        if self.fake_star:
+            return 12345678
+
+        hospital_visit_id = self._get_rows(hv_query, parameters)
+
+        # fetchall returns a list of tuples. We want the first element of the first tuple
+        if not isinstance(hospital_visit_id[0][0], int):
+            logger.warning(
+                f"hospital_visit_id[0][0] is not integer {hospital_visit_id}"
+            )
+
+        return hospital_visit_id[0][0]
+
+    def _get_rows(self, sql_query: sql.SQL, parameters: dict):
         try:
             with self.connection_pool.getconn() as db_connection:
                 with db_connection.cursor() as curs:
-                    curs.execute(self.sql_query, parameters)
+                    curs.execute(sql_query, parameters)
                     rows = curs.fetchall()
                 self.connection_pool.putconn(db_connection)
         except psycopg2.errors.OperationalError as e:
             self.connection_pool.putconn(db_connection)
             raise ConnectionError(f"Data base error: {e}")
+        return rows
 
-        if len(rows) != 1:
-            raise ValueError(
-                f"Wrong number of rows returned from database. {len(rows)} != 1, for {location_string}:{observation_datetime}"
+
+class caboodleDB:
+    """For querying the caboodle database to extract electronic healthcare records per
+    patient."""
+
+    connection_string: str = "dbname={} user={} password={} host={} port={} connect_timeout={} options='-c statement_timeout={}'".format(
+        settings.CABOODLE_DBNAME,  # type:ignore
+        settings.CABOODLE_USERNAME,  # type:ignore
+        settings.CABOODLE_PASSWORD,  # type:ignore
+        settings.CABOODLE_HOST,  # type:ignore
+        settings.CABOODLE_PORT,  # type:ignore
+        settings.CABOODLE_CONNECT_TIMEOUT,  # type:ignore
+        settings.CABOODLE_QUERY_TIMEOUT,  # type:ignore
+    )
+    connection_pool: pool.SimpleConnectionPool
+    fake_caboodle: bool = False
+
+    def connect(self) -> None:
+        """Set up connection to the database."""
+        self.fake_caboodle = True if settings.CABOODLE_TESTING == "TRUE" else False
+        if not self.fake_caboodle:
+            self.connection_pool = pool.SimpleConnectionPool(
+                1, 1, self.connection_string
             )
 
-        return rows[0]
+    def get_airflow(
+        self, start_datetime: datetime, end_datetime: datetime, csn: str
+    ) -> pd.DataFrame:
+        """Retrieve airflow data from database."""
+
+        with open(settings.SQL_PATH + "airway.sql", "r") as file:
+            airway_query = sql.SQL(file.read())
+        parameters = {
+            "start_datetime": start_datetime,
+            "end_datetime": end_datetime,
+            "csn": csn,
+        }
+
+        if self.fake_caboodle:
+            fake_airway = {
+                "DateTimeRecorded": [0],
+                "PlacementInstant": [0],
+                "RemovalInstant": [0],
+                "TubeSize": [0],
+            }
+            return pd.DataFrame(data=fake_airway)
+
+        return self._get_rows(airway_query, parameters)
+
+    def get_flowsheets(
+        self, start_datetime: datetime, end_datetime: datetime, hospital_visit_id: int
+    ) -> pd.DataFrame:
+        """Retrieve airflow data from database."""
+
+        with open(settings.SQL_PATH + "flow_sheet_values.sql", "r") as file:
+            flowsheet_query = sql.SQL(file.read())
+        parameters = {
+            "start_datetime": start_datetime,
+            "end_datetime": end_datetime,
+            "hospital_visit_id": hospital_visit_id,
+        }
+
+        if self.fake_caboodle:
+            fake_flowsheet = {
+                "DateTimeRecorded": [0],
+                "Temperature": [0],
+                "Noradrenaline": [0],
+                "Metaraminol": [0],
+            }
+            return pd.DataFrame(data=fake_flowsheet)
+
+        return self._get_rows(flowsheet_query, parameters)
+
+    def _get_rows(self, sql_query: sql.SQL, parameters: dict):
+        try:
+            with self.connection_pool.getconn() as db_connection:
+                with db_connection.cursor() as curs:
+                    curs.execute(sql_query, parameters)
+                    rows = curs.fetchall()
+                self.connection_pool.putconn(db_connection)
+        except psycopg2.errors.OperationalError as e:
+            self.connection_pool.putconn(db_connection)
+            raise ConnectionError(f"Data base error: {e}")
+
+        return rows