From 820840d9d46a7683eeba4970b3067ae9049a7c01 Mon Sep 17 00:00:00 2001 From: Sarah Keating Date: Mon, 2 Feb 2026 15:06:28 +0000 Subject: [PATCH 01/35] first query for flow sheets --- .gitignore | 1 + sql_scripts/flow_sheet_values.sql | 13 +++++++++++++ 2 files changed, 14 insertions(+) create mode 100644 sql_scripts/flow_sheet_values.sql diff --git a/.gitignore b/.gitignore index be3f5ce..5fac6cd 100644 --- a/.gitignore +++ b/.gitignore @@ -11,6 +11,7 @@ wheels/ # IDEs .idea/ +.vscode/ # settings files (should not be in the source tree anyway, but just in case) *.env diff --git a/sql_scripts/flow_sheet_values.sql b/sql_scripts/flow_sheet_values.sql new file mode 100644 index 0000000..77d6cc1 --- /dev/null +++ b/sql_scripts/flow_sheet_values.sql @@ -0,0 +1,13 @@ +select observation_datetime as datetime, +value_as_real, +unit, +value_as_text, +comment, +vo.visit_observation_type_id , +(select display_name +from star.visit_observation_type as vt +where vt.visit_observation_type_id = vo.visit_observation_type_id ) as observation +from star.visit_observation as vo +left join star.visit_observation_type as vt +on vo.visit_observation_type_id = vt.visit_observation_type_id +where hospital_visit_id = 'csn' \ No newline at end of file From fb3d6da7edf40527fdbb5df5a8abf161614e1768 Mon Sep 17 00:00:00 2001 From: Sarah Keating Date: Wed, 4 Feb 2026 10:30:40 +0000 Subject: [PATCH 02/35] building up scripts --- sql_scripts/flow_sheet_values.sql | 2 +- sql_scripts/lab_results.sql | 14 ++++++++++++++ 2 files changed, 15 insertions(+), 1 deletion(-) create mode 100644 sql_scripts/lab_results.sql diff --git a/sql_scripts/flow_sheet_values.sql b/sql_scripts/flow_sheet_values.sql index 77d6cc1..bbfdfb9 100644 --- a/sql_scripts/flow_sheet_values.sql +++ b/sql_scripts/flow_sheet_values.sql @@ -1,7 +1,7 @@ select observation_datetime as datetime, value_as_real, unit, -value_as_text, +value_as_text,£. comment, vo.visit_observation_type_id , (select display_name diff --git a/sql_scripts/lab_results.sql b/sql_scripts/lab_results.sql new file mode 100644 index 0000000..9160ba6 --- /dev/null +++ b/sql_scripts/lab_results.sql @@ -0,0 +1,14 @@ +select +result_last_modified_datetime as date, +value_as_real as value, +units, +abnormal_flag, +comment + + +from star.lab_result as r +join star.lab_order as o +on r.lab_order_id = o.lab_order_id +where o.hospital_visit_id = 'xxx' +and r.result_status like 'FINAL' +and r.lab_test_definition_id in ('1001', '390793054', '390793057', '390793060', '722790196') From 8476351baada271db1b38a1545ce57096dcd5e90 Mon Sep 17 00:00:00 2001 From: Sarah Keating Date: Thu, 19 Feb 2026 14:54:35 +0000 Subject: [PATCH 03/35] adding more files --- sql_scripts/flow_sheet_values.sql | 22 ++++++++++++++++------ sql_scripts/get_hospital_visit_id.sql | 2 ++ sql_scripts/lab_results.sql | 12 +++++++++++- sql_scripts/lab_test_names.sql | 5 +++++ 4 files changed, 34 insertions(+), 7 deletions(-) create mode 100644 sql_scripts/get_hospital_visit_id.sql create mode 100644 sql_scripts/lab_test_names.sql diff --git a/sql_scripts/flow_sheet_values.sql b/sql_scripts/flow_sheet_values.sql index bbfdfb9..2a6c9eb 100644 --- a/sql_scripts/flow_sheet_values.sql +++ b/sql_scripts/flow_sheet_values.sql @@ -1,13 +1,23 @@ +--get the flow sheet values for the particular visit +-- the flow sheet numbers are recorded as id_in_application in the visit_observation_type table +-- Temperature 6 +-- Noradrenalin 3040102622 +-- Metaraminol 12946 +-- Secretion amount 451120 +-- Sputum amount 302600 + + select observation_datetime as datetime, +(select display_name +from star.visit_observation_type as vt +where vt.visit_observation_type_id = vo.visit_observation_type_id ) as observation, value_as_real, unit, -value_as_text,£. +value_as_text, comment, -vo.visit_observation_type_id , -(select display_name -from star.visit_observation_type as vt -where vt.visit_observation_type_id = vo.visit_observation_type_id ) as observation +vo.visit_observation_type_id from star.visit_observation as vo left join star.visit_observation_type as vt on vo.visit_observation_type_id = vt.visit_observation_type_id -where hospital_visit_id = 'csn' \ No newline at end of file +where vt.id_in_application in ('6', '12946', '302600', '451120', '3040102622') +and hospital_visit_id = 'xx' \ No newline at end of file diff --git a/sql_scripts/get_hospital_visit_id.sql b/sql_scripts/get_hospital_visit_id.sql new file mode 100644 index 0000000..5a60221 --- /dev/null +++ b/sql_scripts/get_hospital_visit_id.sql @@ -0,0 +1,2 @@ +select hospital_visit_id from star.hospital_visit as hv +where hv.encounter = 'csn' \ No newline at end of file diff --git a/sql_scripts/lab_results.sql b/sql_scripts/lab_results.sql index 9160ba6..9309722 100644 --- a/sql_scripts/lab_results.sql +++ b/sql_scripts/lab_results.sql @@ -1,5 +1,14 @@ +-- This selects the values are units of lab tests +-- 1011 CRP +-- 722790196 CRP +-- 390793054 WCC +-- 390793057 WCC +-- 390793060 WCC + select result_last_modified_datetime as date, +(select name from star.lab_test_definition as ltd +where ltd.lab_test_definition_id = r.lab_test_definition_id) as name, value_as_real as value, units, abnormal_flag, @@ -9,6 +18,7 @@ comment from star.lab_result as r join star.lab_order as o on r.lab_order_id = o.lab_order_id -where o.hospital_visit_id = 'xxx' +where o.hospital_visit_id = 'xx' and r.result_status like 'FINAL' and r.lab_test_definition_id in ('1001', '390793054', '390793057', '390793060', '722790196') + diff --git a/sql_scripts/lab_test_names.sql b/sql_scripts/lab_test_names.sql new file mode 100644 index 0000000..85ce715 --- /dev/null +++ b/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') \ No newline at end of file From 39611434440923c21c600b6c9b0f83dbe68c393f Mon Sep 17 00:00:00 2001 From: Sarah Keating Date: Sun, 9 Aug 2026 16:54:34 +0100 Subject: [PATCH 04/35] Add and refine SQL scripts for EHR data extraction Introduces new queries for airway and sputum/secretions data, and refactors the flow sheet query to consolidate multiple values into a wider format. A new README documents the overall data extraction goal and the current set of scripts. --- sql_scripts/README.md | 33 ++++++++++++++++++ sql_scripts/airway.sql | 20 +++++++++++ sql_scripts/flow_sheet_values.sql | 50 +++++++++++++++++---------- sql_scripts/get_hospital_visit_id.sql | 7 +++- sql_scripts/lab_results.sql | 10 +++--- sql_scripts/sputum_secretions.sql | 25 ++++++++++++++ 6 files changed, 120 insertions(+), 25 deletions(-) create mode 100644 sql_scripts/README.md create mode 100644 sql_scripts/airway.sql create mode 100644 sql_scripts/sputum_secretions.sql diff --git a/sql_scripts/README.md b/sql_scripts/README.md new file mode 100644 index 0000000..98ae5af --- /dev/null +++ b/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. diff --git a/sql_scripts/airway.sql b/sql_scripts/airway.sql new file mode 100644 index 0000000..3b0889e --- /dev/null +++ b/sql_scripts/airway.sql @@ -0,0 +1,20 @@ + +SELECT +lda._CreationInstant as DateTimeRecorded, +lda.PlacementInstant, +lda.RemovalInstant, +fvf.Value AS TubeSize + +FROM FilteredAccess.LdaFact lda +JOIN FilteredAccess.FlowsheetValueFact fvf ON fvf.LdaKey = lda.LdaKey +JOIN FilteredAccess.FlowsheetRowDim frd ON frd.FlowsheetRowKey = fvf.FlowsheetRowKey +JOIN FilteredAccess.EncounterFact enc ON enc.EncounterKey = lda.InitialEncounterKey + +WHERE +fvf.FlowsheetRowEpicId ='1120100079' +AND enc.Type != 'Anaesthesia' +AND frd.DisplayName like 'Single Lumen Tube Size' +--and enc.PatientDurableKey = '1782941' + +AND lda._CreationInstant BETWEEN %(yesterday)s AND %(today)s +AND enc.EncounterEpicCsn = %(csn) diff --git a/sql_scripts/flow_sheet_values.sql b/sql_scripts/flow_sheet_values.sql index 2a6c9eb..af9b76f 100644 --- a/sql_scripts/flow_sheet_values.sql +++ b/sql_scripts/flow_sheet_values.sql @@ -1,23 +1,35 @@ ---get the flow sheet values for the particular visit +--get the flow sheet values for the particular visit on a particular day -- the flow sheet numbers are recorded as id_in_application in the visit_observation_type table -- Temperature 6 -- Noradrenalin 3040102622 -- Metaraminol 12946 --- Secretion amount 451120 --- Sputum amount 302600 - - -select observation_datetime as datetime, -(select display_name -from star.visit_observation_type as vt -where vt.visit_observation_type_id = vo.visit_observation_type_id ) as observation, -value_as_real, -unit, -value_as_text, -comment, -vo.visit_observation_type_id -from star.visit_observation as vo -left join star.visit_observation_type as vt -on vo.visit_observation_type_id = vt.visit_observation_type_id -where vt.id_in_application in ('6', '12946', '302600', '451120', '3040102622') -and hospital_visit_id = 'xx' \ No newline at end of file + +SELECT + vo.observation_datetime AS DateTimeRecorded, + + (array_agg(vo.value_as_real) FILTER ( + WHERE vt.id_in_application = '6' + ))[1] AS "Temperature", + + (array_agg(vo.value_as_real) FILTER ( + WHERE vt.id_in_application = '3040102622' + ))[1] AS "Noradrenaline", + + (array_agg(vo.value_as_real) FILTER ( + WHERE vt.id_in_application = '12946' + ))[1] AS "Metaraminol", + + vo.unit AS Units, + vo.comment AS Comments + +FROM star.visit_observation AS vo + +LEFT JOIN star.visit_observation_type AS vt + ON vo.visit_observation_type_id = vt.visit_observation_type_id + +WHERE vt.id_in_application IN ('6', '3040102622', '12946') +AND vo.valid_from BETWEEN %(yesterday)s AND %(today)s +AND vo.hospital_visit_id = %(hospital_visit_id)s + +GROUP BY DateTimeRecorded, Units, vo.comment + diff --git a/sql_scripts/get_hospital_visit_id.sql b/sql_scripts/get_hospital_visit_id.sql index 5a60221..408447f 100644 --- a/sql_scripts/get_hospital_visit_id.sql +++ b/sql_scripts/get_hospital_visit_id.sql @@ -1,2 +1,7 @@ + + + select hospital_visit_id from star.hospital_visit as hv -where hv.encounter = 'csn' \ No newline at end of file +where hv.encounter = %(csn)s -- note the CSN must be in quotes + + diff --git a/sql_scripts/lab_results.sql b/sql_scripts/lab_results.sql index 9309722..aa05264 100644 --- a/sql_scripts/lab_results.sql +++ b/sql_scripts/lab_results.sql @@ -6,13 +6,13 @@ -- 390793060 WCC select -result_last_modified_datetime as date, +r.result_last_modified_datetime as date, (select name from star.lab_test_definition as ltd where ltd.lab_test_definition_id = r.lab_test_definition_id) as name, -value_as_real as value, -units, -abnormal_flag, -comment +r.value_as_real as value, +r.units, +r.abnormal_flag, +r.comment from star.lab_result as r diff --git a/sql_scripts/sputum_secretions.sql b/sql_scripts/sputum_secretions.sql new file mode 100644 index 0000000..8f27ab8 --- /dev/null +++ b/sql_scripts/sputum_secretions.sql @@ -0,0 +1,25 @@ +<- Retrieved the information about sputum and secretions > + +SELECT +fv.TakenInstant AS 'DateTimeRecorded', +CASE +WHEN fsd.FlowsheetRowEpicId = '451120' +THEN fv.Value +END +AS 'Secretions' , +CASE +WHEN fsd.FlowsheetRowEpicId = '302600' +THEN fv.Value +END +AS 'Sputum' , +fv.Comment AS Comments + +FROM FilteredAccess.FlowsheetValueFact fv +INNER JOIN FilteredAccess.FlowsheetRowDim fsd ON fv.FlowsheetRowKey = fsd.FlowsheetRowKey +INNER JOIN FilteredAccess.EncounterFact enc ON fv.EncounterKey = enc.EncounterKey + +WHERE +(fsd.FlowsheetRowEpicId = '451120' OR +fsd.FlowsheetRowEpicId ='302600') +AND fv.TakenInstant BETWEEN %(yesterday)s AND %(today)s +AND enc.EncounterEpicCsn = %(csn) From cdec5caed0749c693984742b417bba6d47c2435a Mon Sep 17 00:00:00 2001 From: Stephen Thompson Date: Tue, 11 Aug 2026 15:42:19 +0100 Subject: [PATCH 05/35] Started on functions to wrap snakemake around sql calls --- src/electronic_health_records/__init__.py | 0 src/electronic_health_records/ehr.py | 180 ++++++++++++++++++++++ 2 files changed, 180 insertions(+) create mode 100644 src/electronic_health_records/__init__.py create mode 100644 src/electronic_health_records/ehr.py diff --git a/src/electronic_health_records/__init__.py b/src/electronic_health_records/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/electronic_health_records/ehr.py b/src/electronic_health_records/ehr.py new file mode 100644 index 0000000..ebd514c --- /dev/null +++ b/src/electronic_health_records/ehr.py @@ -0,0 +1,180 @@ +import argparse +import functools +import json +import logging +from decimal import Decimal +from pathlib import Path +from typing import Any + +import pandas as pd +import pyarrow as pa +import pyarrow.parquet as pq +import settings + +from locations import ( + CSV_PATTERN, + ORIGINAL_PARQUET_PATTERN, + PSEUDONYMISED_PARQUET_PATTERN, +) +from .hashing import do_hash + +def ehr_for_csv( + *, + date_str: str, + original_csn: str, + hashed_csn: str, +) -> None: + """Extracts electronic healthcare records for a given csn and writes the + results to a pseudonymised csv file for a single day. + + This is a privacy-sensitive area of code. Unhashed CSNs must not appear in + uploaded files. + :param date_str: the date to look up data for + :param original_csn: the csn to base look up on. + :param hashed_csn: the pseudonymised hash to use for file output. + + """ + # will pick up the logger config defined in the snakemake job (ie. log to file) + logger = logging.getLogger(__name__) + + # it's in the csv_path and original_parquet_path, but at least nowhere else! + del original_csn + + logger.info("Turning CSV %s to parquets", csv_path) + csv_path.parent.mkdir(parents=True, exist_ok=True) + original_parquet_path.parent.mkdir(parents=True, exist_ok=True) + df = pd.read_csv( + str(csv_path), + dtype={ + "csn": str, + "mrn": str, + "source_variable_id": str, + "source_channel_id": str, + "units": str, + "sampling_rate": int, + "timestamp": float, + "location": str, + "values": str, + }, + header=0, # the first line is always the header + ) + + def parse_array(x): + # Not sure if this is the most efficient way. Might be able to do something with DecimalArray? + # return [pa.decimal128(i) for i in x.replace(' ', '').split(',')] + return [Decimal(i) for i in x.strip().strip("[]").replace(" ", "").split(",")] + + df["values"] = df["values"].apply(parse_array) + + # Convert pandas DataFrame to pyarrow Table with proper types + schema = pa.schema( + [ + ("csn", pa.string()), + ("mrn", pa.string()), + ("source_variable_id", pa.string()), + ("source_channel_id", pa.string()), + ("units", pa.string()), + ("sampling_rate", pa.int32()), + ("timestamp", pa.float64()), + ("location", pa.string()), + # As per requirements, compactness is important here. + # decimal32 can have a maximum of 9 significant digits and should + # satisfy our needs, but it only exists in pyarrow >= 19. + # We are currently tied to 18.1 because of PIXL core. + # So for now, use decimal128 instead. + # Not yet tested whether the specified precision + # and scale cause it to be equivalent in size to decimal32. + # See issue #31. + ("values", pa.list_(pa.decimal128(9, 4))), + ] + ) + table = pa.Table.from_pandas(df, schema=schema, preserve_index=True) + + # mark the parquet files themselves as production or not. + our_metadata = {"instance_name": settings.INSTANCE_NAME} + + table = add_waveform_metadata_to_table(table, our_metadata) + + pq.write_table( + table, + str(original_parquet_path), + # valid values: {‘NONE’, ‘SNAPPY’, ‘GZIP’, ‘BROTLI’, ‘LZ4’, ‘ZSTD’} + compression="zstd", + use_dictionary=True, + write_statistics=True, # enable indexes/statistics + flavor="spark", + ) + logger.info( + "Done turning CSV %s to original parquet %s", csv_path, original_parquet_path + ) + + df = pseudonymise_relevant_columns(df) + pseudon_table = pa.Table.from_pandas(df, schema=schema, preserve_index=True) + + # Use same metadata for pseudon, must not contain identifiers! + pseudon_table = add_waveform_metadata_to_table(pseudon_table, our_metadata) + + hashed_path = Path( + str(PSEUDONYMISED_PARQUET_PATTERN).format( + date=date_str, + hashed_csn=hashed_csn, + variable_id=variable_id, + channel_id=channel_id, + units=units, + ) + ) + pq.write_table( + pseudon_table, + str(hashed_path), + compression="zstd", + use_dictionary=True, + write_statistics=True, # enable indexes/statistics + flavor="spark", + ) + logger.info( + "Done turning CSV %s to pseudonymised parquet %s", csv_path, hashed_path + ) + + +def add_waveform_metadata_to_table( + existing_table: pa.Table, metadata: dict[str, Any] +) -> pa.Table: + """Replace our metadata in its entirety, leaving untouched metadata we didn't + set.""" + + # Parquet footer metadata is a series of (byte string) key-value pairs. + # Other users of metadata (eg. pandas) convert their metadata to JSON and store it under + # a single key (a namespace, effectively), so we'll do the same under our own key. + waveform_exporter_metadata_key = b"waveform_exporter" + + existing_metadata = existing_table.schema.metadata or {} + json_byte_string = json.dumps(metadata).encode("utf-8") + existing_table = existing_table.replace_schema_metadata( + {**existing_metadata, waveform_exporter_metadata_key: json_byte_string} + ) + return existing_table + + +SAFE_COLUMNS = [ + "sampling_rate", + "source_variable_id", + "source_channel_id", + "timestamp", + "units", + "values", +] + + +def pseudonymise_relevant_columns(df: pd.DataFrame): + """ "csn", "mrn", "location" are examples of columns that must be pseudonymised. + + However, it's safer to list which columns *don't* need to be pseudonymised. Eg. you + add a column but forget to consider whether it's sensitive, OR you rename one of the + known sensitive columns and forget that this will cause privacy to break. This means + that when you add a new column, you have to add it here if you don't want it to be + hashed. + """ + for col in df.columns: + if col not in SAFE_COLUMNS: + df[col] = df[col].apply(functools.partial(do_hash, col)) + return df From 98972596b5ed6865279777b0b1cb177699f41c4b Mon Sep 17 00:00:00 2001 From: Stephen Thompson Date: Wed, 12 Aug 2026 14:13:13 +0100 Subject: [PATCH 06/35] Started to implement caboodle look up code --- src/db.py | 44 ++++++++ src/electronic_health_records/ehr.py | 153 ++++----------------------- 2 files changed, 64 insertions(+), 133 deletions(-) diff --git a/src/db.py b/src/db.py index ebab871..dff5872 100644 --- a/src/db.py +++ b/src/db.py @@ -53,3 +53,47 @@ def get_row(self, location_string: str, observation_datetime: datetime): ) return rows[0] + + +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.ThreadedConnectionPool + + def connect(self): + """Set up connection to the database.""" + self.connection_pool = pool.SimpleConnectionPool(1, 1, self.connection_string) + + def get_airflow(self, start_datetime: datetime, end_datetime: datetime, csn: str): + """Retrieve airflow data from database.""" + with open("src/sql/airway.sql", "r") as file: + airway_query = sql.SQL(file.read()) + parameters = { + "start_datetime": start_datetime, + "end_datetime": end_datetime, + "csn": csn, + } + return self._get_rows(airway_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 diff --git a/src/electronic_health_records/ehr.py b/src/electronic_health_records/ehr.py index ebd514c..a5d063a 100644 --- a/src/electronic_health_records/ehr.py +++ b/src/electronic_health_records/ehr.py @@ -1,158 +1,45 @@ -import argparse import functools -import json import logging -from decimal import Decimal -from pathlib import Path -from typing import Any import pandas as pd -import pyarrow as pa -import pyarrow.parquet as pq -import settings +from datetime import datetime -from locations import ( - CSV_PATTERN, - ORIGINAL_PARQUET_PATTERN, - PSEUDONYMISED_PARQUET_PATTERN, -) from .hashing import do_hash +from db import caboodleDB + def ehr_for_csv( - *, - date_str: str, + datetime: datetime, original_csn: str, hashed_csn: str, + db_connection: caboodleDB, ) -> None: - """Extracts electronic healthcare records for a given csn and writes the - results to a pseudonymised csv file for a single day. + """Extracts electronic healthcare records for a given csn and writes the results to + a pseudonymised csv file for a single day. - This is a privacy-sensitive area of code. Unhashed CSNs must not appear in - uploaded files. + This is a privacy-sensitive area of code. Unhashed CSNs must not appear in uploaded + files. :param date_str: the date to look up data for :param original_csn: the csn to base look up on. :param hashed_csn: the pseudonymised hash to use for file output. - + :param db_connection: connection to the caboodle database. """ # will pick up the logger config defined in the snakemake job (ie. log to file) logger = logging.getLogger(__name__) - # it's in the csv_path and original_parquet_path, but at least nowhere else! - del original_csn - - logger.info("Turning CSV %s to parquets", csv_path) - csv_path.parent.mkdir(parents=True, exist_ok=True) - original_parquet_path.parent.mkdir(parents=True, exist_ok=True) - df = pd.read_csv( - str(csv_path), - dtype={ - "csn": str, - "mrn": str, - "source_variable_id": str, - "source_channel_id": str, - "units": str, - "sampling_rate": int, - "timestamp": float, - "location": str, - "values": str, - }, - header=0, # the first line is always the header - ) - - def parse_array(x): - # Not sure if this is the most efficient way. Might be able to do something with DecimalArray? - # return [pa.decimal128(i) for i in x.replace(' ', '').split(',')] - return [Decimal(i) for i in x.strip().strip("[]").replace(" ", "").split(",")] - - df["values"] = df["values"].apply(parse_array) - - # Convert pandas DataFrame to pyarrow Table with proper types - schema = pa.schema( - [ - ("csn", pa.string()), - ("mrn", pa.string()), - ("source_variable_id", pa.string()), - ("source_channel_id", pa.string()), - ("units", pa.string()), - ("sampling_rate", pa.int32()), - ("timestamp", pa.float64()), - ("location", pa.string()), - # As per requirements, compactness is important here. - # decimal32 can have a maximum of 9 significant digits and should - # satisfy our needs, but it only exists in pyarrow >= 19. - # We are currently tied to 18.1 because of PIXL core. - # So for now, use decimal128 instead. - # Not yet tested whether the specified precision - # and scale cause it to be equivalent in size to decimal32. - # See issue #31. - ("values", pa.list_(pa.decimal128(9, 4))), - ] - ) - table = pa.Table.from_pandas(df, schema=schema, preserve_index=True) - - # mark the parquet files themselves as production or not. - our_metadata = {"instance_name": settings.INSTANCE_NAME} + caboodle = caboodleDB() + caboodle.connect() - table = add_waveform_metadata_to_table(table, our_metadata) + logger.info("Looking for airway data for %s.", hashed_csn) - pq.write_table( - table, - str(original_parquet_path), - # valid values: {‘NONE’, ‘SNAPPY’, ‘GZIP’, ‘BROTLI’, ‘LZ4’, ‘ZSTD’} - compression="zstd", - use_dictionary=True, - write_statistics=True, # enable indexes/statistics - flavor="spark", - ) - logger.info( - "Done turning CSV %s to original parquet %s", csv_path, original_parquet_path - ) + start_datetime = datetime + end_datetime = datetime + airflow = caboodle.get_airflow(start_datetime, end_datetime, original_csn) + airflow = pseudonymise_relevant_columns(airflow) + logger.info(airflow) - df = pseudonymise_relevant_columns(df) - pseudon_table = pa.Table.from_pandas(df, schema=schema, preserve_index=True) - - # Use same metadata for pseudon, must not contain identifiers! - pseudon_table = add_waveform_metadata_to_table(pseudon_table, our_metadata) - - hashed_path = Path( - str(PSEUDONYMISED_PARQUET_PATTERN).format( - date=date_str, - hashed_csn=hashed_csn, - variable_id=variable_id, - channel_id=channel_id, - units=units, - ) - ) - pq.write_table( - pseudon_table, - str(hashed_path), - compression="zstd", - use_dictionary=True, - write_statistics=True, # enable indexes/statistics - flavor="spark", - ) - logger.info( - "Done turning CSV %s to pseudonymised parquet %s", csv_path, hashed_path - ) - - -def add_waveform_metadata_to_table( - existing_table: pa.Table, metadata: dict[str, Any] -) -> pa.Table: - """Replace our metadata in its entirety, leaving untouched metadata we didn't - set.""" - - # Parquet footer metadata is a series of (byte string) key-value pairs. - # Other users of metadata (eg. pandas) convert their metadata to JSON and store it under - # a single key (a namespace, effectively), so we'll do the same under our own key. - waveform_exporter_metadata_key = b"waveform_exporter" - - existing_metadata = existing_table.schema.metadata or {} - json_byte_string = json.dumps(metadata).encode("utf-8") - existing_table = existing_table.replace_schema_metadata( - {**existing_metadata, waveform_exporter_metadata_key: json_byte_string} - ) - return existing_table + # delete csn once we no longer need it + del original_csn SAFE_COLUMNS = [ From 1f2cfbe47d25f50bd5393cbad8a9841eafe9c82a Mon Sep 17 00:00:00 2001 From: Stephen Thompson Date: Wed, 12 Aug 2026 14:51:19 +0100 Subject: [PATCH 07/35] Moved airway sql to sql directory --- {sql_scripts => src/sql}/airway.sql | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) rename {sql_scripts => src/sql}/airway.sql (81%) diff --git a/sql_scripts/airway.sql b/src/sql/airway.sql similarity index 81% rename from sql_scripts/airway.sql rename to src/sql/airway.sql index 3b0889e..c79c9cb 100644 --- a/sql_scripts/airway.sql +++ b/src/sql/airway.sql @@ -1,4 +1,4 @@ - +-- extract airway data from caboodle for a specific csn and date SELECT lda._CreationInstant as DateTimeRecorded, lda.PlacementInstant, @@ -16,5 +16,5 @@ AND enc.Type != 'Anaesthesia' AND frd.DisplayName like 'Single Lumen Tube Size' --and enc.PatientDurableKey = '1782941' -AND lda._CreationInstant BETWEEN %(yesterday)s AND %(today)s +AND lda._CreationInstant BETWEEN %(start_datetime)s AND %(end_datetime)s AND enc.EncounterEpicCsn = %(csn) From bd72361aadbb9ca867e0d162dd9253fe8ee3d0c7 Mon Sep 17 00:00:00 2001 From: Stephen Thompson Date: Wed, 12 Aug 2026 14:52:10 +0100 Subject: [PATCH 08/35] Setting up config and docker bits --- config.EXAMPLE/ehr_lookup.EXAMPLE | 9 +++++++++ docker-compose.yml | 21 +++++++++++++++++++++ 2 files changed, 30 insertions(+) create mode 100644 config.EXAMPLE/ehr_lookup.EXAMPLE diff --git a/config.EXAMPLE/ehr_lookup.EXAMPLE b/config.EXAMPLE/ehr_lookup.EXAMPLE new file mode 100644 index 0000000..022ce93 --- /dev/null +++ b/config.EXAMPLE/ehr_lookup.EXAMPLE @@ -0,0 +1,9 @@ +# This is an EXAMPLE file, do not put real secrets in here. +# Copy it to ../config/ehr_lookup.env and then DELETE THIS COMMENT. +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 diff --git a/docker-compose.yml b/docker-compose.yml index 6c30ca7..eaaa113 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -52,3 +52,24 @@ services: env_file: - ../config/hasher.env restart: unless-stopped + ehr-lookup: + build: + context: .. + dockerfile: waveform-controller/Dockerfile + target: ehr-lookup + args: + HTTP_PROXY: ${HTTP_PROXY} + http_proxy: ${http_proxy} + HTTPS_PROXY: ${HTTPS_PROXY} + https_proxy: ${https_proxy} + # ideally we'd use docker secrets but it's not enabled currently + env_file: + - ../config/ehr_lookup.env + volumes: + - ../waveform-export:/waveform-export + # because we're launching through cron in the container, which starts + # processes with a clean environment, also mount in the config file so + # it can be read in by snakemake later + - ../config/ehr-lookup.env:/config/ehr-lookup.env:ro + restart: unless-stopped + From 5e303f0edf95ba2fa8e683df5bf8db352984bc69 Mon Sep 17 00:00:00 2001 From: Stephen Thompson Date: Mon, 17 Aug 2026 11:44:28 +0100 Subject: [PATCH 09/35] Add a ehr location to snakefile --- src/pipeline/Snakefile | 1 + 1 file changed, 1 insertion(+) diff --git a/src/pipeline/Snakefile b/src/pipeline/Snakefile index dbce7db..7dc9a3a 100644 --- a/src/pipeline/Snakefile +++ b/src/pipeline/Snakefile @@ -9,6 +9,7 @@ from locations import ( WAVEFORM_ORIGINAL_CSV, WAVEFORM_SNAKEMAKE_LOGS, WAVEFORM_PSEUDONYMISED_PARQUET, + WAVEFORM_PSEUDONYMISED_EHR, WAVEFORM_FTPS_LOGS, HASH_LOOKUP_JSON, HASH_LOOKUP_JSON_REL, From 89e95269c2642699c29d4706cdb262fef3fb685c Mon Sep 17 00:00:00 2001 From: Stephen Thompson Date: Mon, 17 Aug 2026 15:39:19 +0100 Subject: [PATCH 10/35] Tidied up implementation and rewrote Snakefile logic, with help from Claude Sonnet 5.0 --- src/csv_writer.py | 28 ++++++++++- src/electronic_health_records/ehr.py | 73 ++++++++++++++-------------- src/locations.py | 3 ++ src/pipeline/Snakefile | 38 ++++++++++++++- src/pipeline/utils.py | 6 +++ src/pseudon/pseudon.py | 25 +++++----- 6 files changed, 121 insertions(+), 52 deletions(-) diff --git a/src/csv_writer.py b/src/csv_writer.py index 66948b8..42daeb3 100644 --- a/src/csv_writer.py +++ b/src/csv_writer.py @@ -2,8 +2,15 @@ import csv from datetime import datetime +import pandas as pd -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( @@ -77,3 +84,22 @@ def write_frame( ) return True + + +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) + + return True diff --git a/src/electronic_health_records/ehr.py b/src/electronic_health_records/ehr.py index a5d063a..48fc930 100644 --- a/src/electronic_health_records/ehr.py +++ b/src/electronic_health_records/ehr.py @@ -1,15 +1,31 @@ -import functools import logging -import pandas as pd -from datetime import datetime +from datetime import datetime, timedelta -from .hashing import do_hash from db import caboodleDB +from csv_writer import write_ehr +from pseudon.pseudon import pseudonymise_relevant_columns -def ehr_for_csv( - datetime: datetime, +def ehr_for_csv(date_str: str, original_csn: str, hashed_csn: str) -> None: + """Extracts electronic healthcare records for a given csn and writes the results to + a pseudonymised csv file for a single day. + + This is a privacy-sensitive area of code. Unhashed CSNs must not appear in uploaded + files. + :param date_str: the date to look up data for + :param original_csn: the csn to base look up on. + :param hashed_csn: the pseudonymised hash to use for file output. + """ + + db_connection = caboodleDB() + db_connection.connect() + + _ehr_for_csv(date_str, original_csn, hashed_csn, db_connection) + + +def _ehr_for_csv( + date_str: str, original_csn: str, hashed_csn: str, db_connection: caboodleDB, @@ -27,41 +43,24 @@ def ehr_for_csv( # will pick up the logger config defined in the snakemake job (ie. log to file) logger = logging.getLogger(__name__) - caboodle = caboodleDB() - caboodle.connect() - logger.info("Looking for airway data for %s.", hashed_csn) - start_datetime = datetime - end_datetime = datetime - airflow = caboodle.get_airflow(start_datetime, end_datetime, original_csn) - airflow = pseudonymise_relevant_columns(airflow) - logger.info(airflow) + start_datetime = datetime.strptime(date_str, "%Y-%m-%d") + end_datetime = start_datetime + timedelta(days=1) + airflow = db_connection.get_airflow(start_datetime, end_datetime, original_csn) - # delete csn once we no longer need it - del original_csn + safe_columns = [ + "DateTimeRecorded", + "PlacementInstant", + "RemovalInstant", + "TubeSize", + ] + airflow = pseudonymise_relevant_columns(airflow, safe_columns) -SAFE_COLUMNS = [ - "sampling_rate", - "source_variable_id", - "source_channel_id", - "timestamp", - "units", - "values", -] + write_ehr(airflow, date_str, hashed_csn) + logger.info(airflow) -def pseudonymise_relevant_columns(df: pd.DataFrame): - """ "csn", "mrn", "location" are examples of columns that must be pseudonymised. - - However, it's safer to list which columns *don't* need to be pseudonymised. Eg. you - add a column but forget to consider whether it's sensitive, OR you rename one of the - known sensitive columns and forget that this will cause privacy to break. This means - that when you add a new column, you have to add it here if you don't want it to be - hashed. - """ - for col in df.columns: - if col not in SAFE_COLUMNS: - df[col] = df[col].apply(functools.partial(do_hash, col)) - return df + # delete csn once we no longer need it + del original_csn diff --git a/src/locations.py b/src/locations.py index bb15847..f55fa20 100644 --- a/src/locations.py +++ b/src/locations.py @@ -5,6 +5,7 @@ WAVEFORM_ORIGINAL_PARQUET = WAVEFORM_EXPORT_BASE / "original-parquet" WAVEFORM_HASH_LOOKUPS = WAVEFORM_EXPORT_BASE / "hash-lookups" WAVEFORM_PSEUDONYMISED_PARQUET = WAVEFORM_EXPORT_BASE / "pseudonymised" +WAVEFORM_PSEUDONYMISED_EHR = WAVEFORM_EXPORT_BASE / "pseudonymised_ehr" WAVEFORM_SNAKEMAKE_LOGS = WAVEFORM_EXPORT_BASE / "snakemake-logs" WAVEFORM_FTPS_LOGS = WAVEFORM_EXPORT_BASE / "ftps-logs" @@ -14,6 +15,8 @@ FILE_STEM_PATTERN_HASHED = ( "{date}/{date}.{hashed_csn}.{variable_id}.{channel_id}.{units}" ) +# EHR data is per (date, csn), not per variable/channel/units, so it gets its own stem. +EHR_STEM_PATTERN_HASHED = "{date}/{date}.{hashed_csn}" CSV_PATTERN = WAVEFORM_ORIGINAL_CSV / (FILE_STEM_PATTERN + ".csv") ORIGINAL_PARQUET_PATTERN = WAVEFORM_ORIGINAL_PARQUET / (FILE_STEM_PATTERN + ".parquet") PSEUDONYMISED_PARQUET_PATTERN = WAVEFORM_PSEUDONYMISED_PARQUET / ( diff --git a/src/pipeline/Snakefile b/src/pipeline/Snakefile index 7dc9a3a..71042d9 100644 --- a/src/pipeline/Snakefile +++ b/src/pipeline/Snakefile @@ -15,9 +15,11 @@ from locations import ( HASH_LOOKUP_JSON_REL, FILE_STEM_PATTERN, FILE_STEM_PATTERN_HASHED, + EHR_STEM_PATTERN_HASHED, make_file_name, ) from pseudon.pseudon import csv_to_parquets +from electronic_health_records.ehr import ehr_for_csv from utils import config_bool, determine_eventual_outputs @@ -48,6 +50,7 @@ PROCESS_CSV_FROM_DATE = str(config['PROCESS_CSV_FROM_DATE']) all_outputs, hash_to_csn = determine_eventual_outputs(CSV_AGE_THRESHOLD_MINUTES, ONLY_USE_CSV_FROM_YESTERDAY, PROCESS_CSV_FROM_DATE) ALL_FTPS_UPLOADED = [ao.get_ftps_uploaded_file() for ao in all_outputs] ALL_DAILY_HASH_LOOKUPS = sorted({ao.get_daily_hash_lookup() for ao in all_outputs}) +ALL_EHR_LOOKUPS = sorted({ao.get_ehr_lookup() for ao in all_outputs}) def configure_file_logging(log_file): import logging @@ -65,7 +68,8 @@ def configure_file_logging(log_file): rule all: input: ftps_uploaded = ALL_FTPS_UPLOADED, - daily_hash_lookups = ALL_DAILY_HASH_LOOKUPS + daily_hash_lookups = ALL_DAILY_HASH_LOOKUPS, + ehr_lookups = ALL_EHR_LOOKUPS rule all_ftps_uploaded: input: @@ -75,6 +79,10 @@ rule all_daily_hash_lookups: input: ALL_DAILY_HASH_LOOKUPS +rule all_ehr_lookups: + input: + ALL_EHR_LOOKUPS + def input_file_maker(wc): unhashed_csn = hash_to_csn[wc.hashed_csn] # when using input functions, snakemake doesn't do its normal templating, you have to do it, hence the f-string @@ -116,6 +124,34 @@ def pseudonymised_parquet_files_for_date(wc): return [ao.get_pseudonymised_parquet_path() for ao in all_outputs if ao.date == wc.date] +def pseudonymised_parquet_files_for_date_and_hashed_csn(wc): + return [ + ao.get_pseudonymised_parquet_path() + for ao in all_outputs + if ao.date == wc.date and ao.hashed_csn == wc.hashed_csn + ] + + +rule ehr_lookup: + input: + # As with daily_hash_lookup, we lie to Snakemake that the input is the pseudon + # parquets for this csn/day, purely so this rule is tied into the dependency DAG + # and reruns if the underlying data for this csn/day changes. + pseudonymised_parquets = pseudonymised_parquet_files_for_date_and_hashed_csn + output: + WAVEFORM_PSEUDONYMISED_EHR / (EHR_STEM_PATTERN_HASHED + "_ehr.csv") + log: + WAVEFORM_SNAKEMAKE_LOGS / "ehr_lookup" / (EHR_STEM_PATTERN_HASHED + ".log") + run: + logger = configure_file_logging(log[0]) + original_csn = hash_to_csn[wildcards.hashed_csn] + logger.info("Running EHR look up for csn %s. Hash -> %s", original_csn, wildcards.hashed_csn) + ehr_for_csv( + date_str=wildcards.date, + original_csn=original_csn, + hashed_csn=wildcards.hashed_csn) + + rule daily_hash_lookup: input: # Because we don't declare the original parquets in the output of csv_to_parquet, diff --git a/src/pipeline/utils.py b/src/pipeline/utils.py index 6bf58b4..a17cd89 100644 --- a/src/pipeline/utils.py +++ b/src/pipeline/utils.py @@ -8,10 +8,12 @@ from pseudon.hashing import do_hash from locations import ( WAVEFORM_PSEUDONYMISED_PARQUET, + WAVEFORM_PSEUDONYMISED_EHR, WAVEFORM_FTPS_LOGS, HASH_LOOKUP_JSON, ORIGINAL_PARQUET_PATTERN, FILE_STEM_PATTERN_HASHED, + EHR_STEM_PATTERN_HASHED, CSV_PATTERN, make_file_name, ) @@ -73,6 +75,10 @@ def get_ftps_uploaded_file(self) -> Path: def get_daily_hash_lookup(self) -> Path: return Path(make_file_name(str(HASH_LOOKUP_JSON), self._subs_dict)) + def get_ehr_lookup(self) -> Path: + final_stem = make_file_name(EHR_STEM_PATTERN_HASHED, self._subs_dict) + return WAVEFORM_PSEUDONYMISED_EHR / f"{final_stem}_ehr.csv" + def get_file_age(file_path: Path) -> timedelta: # need to use UTC to avoid DST issues diff --git a/src/pseudon/pseudon.py b/src/pseudon/pseudon.py index 01cc498..485d310 100644 --- a/src/pseudon/pseudon.py +++ b/src/pseudon/pseudon.py @@ -142,7 +142,16 @@ def parse_array(x): "Done turning CSV %s to original parquet %s", csv_path, original_parquet_path ) - df = pseudonymise_relevant_columns(df) + safe_columns = [ + "sampling_rate", + "source_variable_id", + "source_channel_id", + "timestamp", + "units", + "values", + ] + + df = pseudonymise_relevant_columns(df, safe_columns) pseudon_table = pa.Table.from_pandas(df, schema=schema, preserve_index=True) # Use same metadata for pseudon, must not contain identifiers! @@ -189,17 +198,7 @@ def add_waveform_metadata_to_table( return existing_table -SAFE_COLUMNS = [ - "sampling_rate", - "source_variable_id", - "source_channel_id", - "timestamp", - "units", - "values", -] - - -def pseudonymise_relevant_columns(df: pd.DataFrame): +def pseudonymise_relevant_columns(df: pd.DataFrame, safe_columns: [str]): """ "csn", "mrn", "location" are examples of columns that must be pseudonymised. However, it's safer to list which columns *don't* need to be pseudonymised. Eg. you @@ -209,6 +208,6 @@ def pseudonymise_relevant_columns(df: pd.DataFrame): hashed. """ for col in df.columns: - if col not in SAFE_COLUMNS: + if col not in safe_columns: df[col] = df[col].apply(functools.partial(do_hash, col)) return df From e0498a20fd354018f072ffd2e828a3b5978d837a Mon Sep 17 00:00:00 2001 From: Stephen Thompson Date: Mon, 17 Aug 2026 15:53:06 +0100 Subject: [PATCH 11/35] We don't need a separate container for ehr lookup --- docker-compose.yml | 21 --------------------- 1 file changed, 21 deletions(-) diff --git a/docker-compose.yml b/docker-compose.yml index eaaa113..6c30ca7 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -52,24 +52,3 @@ services: env_file: - ../config/hasher.env restart: unless-stopped - ehr-lookup: - build: - context: .. - dockerfile: waveform-controller/Dockerfile - target: ehr-lookup - args: - HTTP_PROXY: ${HTTP_PROXY} - http_proxy: ${http_proxy} - HTTPS_PROXY: ${HTTPS_PROXY} - https_proxy: ${https_proxy} - # ideally we'd use docker secrets but it's not enabled currently - env_file: - - ../config/ehr_lookup.env - volumes: - - ../waveform-export:/waveform-export - # because we're launching through cron in the container, which starts - # processes with a clean environment, also mount in the config file so - # it can be read in by snakemake later - - ../config/ehr-lookup.env:/config/ehr-lookup.env:ro - restart: unless-stopped - From f7ac2db9f204c448e710ea71c33327c1cb83132d Mon Sep 17 00:00:00 2001 From: Stephen Thompson Date: Mon, 17 Aug 2026 16:00:46 +0100 Subject: [PATCH 12/35] Added ehr variables to settings and added a testing flag. --- config.EXAMPLE/ehr_lookup.EXAMPLE | 5 +++++ src/settings.py | 9 +++++++++ 2 files changed, 14 insertions(+) diff --git a/config.EXAMPLE/ehr_lookup.EXAMPLE b/config.EXAMPLE/ehr_lookup.EXAMPLE index 022ce93..0c88d12 100644 --- a/config.EXAMPLE/ehr_lookup.EXAMPLE +++ b/config.EXAMPLE/ehr_lookup.EXAMPLE @@ -7,3 +7,8 @@ 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" diff --git a/src/settings.py b/src/settings.py index 3bccd12..0309839 100644 --- a/src/settings.py +++ b/src/settings.py @@ -37,6 +37,15 @@ def get_from_env(env_var, *, default_value=None, setting_name=None, required=Fal get_from_env("HASHER_API_HOSTNAME") get_from_env("HASHER_API_PORT") +get_from_env("CABOODLE_DBNAME") +get_from_env("CABOODLE_USERNAME") +get_from_env("CABOODLE_PASSWORD") +get_from_env("CABOODLE_HOST") +get_from_env("CABOODLE_PORT") +get_from_env("CABOODLE_CONNECT_TIMEOUT") +get_from_env("CABOODLE_QUERY_TIMEOUT") +get_from_env("CABOODLE_TESTING") + get_from_env("LOG_LEVEL", default_value="INFO") get_from_env("INSTANCE_NAME", required=True) From 291be43331de141b05c50f86db05c469d9faa6cf Mon Sep 17 00:00:00 2001 From: Stephen Thompson Date: Mon, 17 Aug 2026 16:18:20 +0100 Subject: [PATCH 13/35] added some caboodle faking logic to enable local running --- src/db.py | 14 +++++++++++++- 1 file changed, 13 insertions(+), 1 deletion(-) diff --git a/src/db.py b/src/db.py index dff5872..e0fa11f 100644 --- a/src/db.py +++ b/src/db.py @@ -1,4 +1,5 @@ from datetime import datetime +import pandas as pd import psycopg2 from psycopg2 import sql, pool import logging @@ -69,13 +70,24 @@ class caboodleDB: settings.CABOODLE_QUERY_TIMEOUT, # type:ignore ) connection_pool: pool.ThreadedConnectionPool + fake_caboodle: bool def connect(self): """Set up connection to the database.""" - self.connection_pool = pool.SimpleConnectionPool(1, 1, self.connection_string) + 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) def get_airflow(self, start_datetime: datetime, end_datetime: datetime, csn: str): """Retrieve airflow data from database.""" + if self.fake_caboodle: + fake_airway = { + "DateTimeRecorded": [0], + "PlacementInstant": [0], + "RemovalInstant": [0], + "TubeSize": [0], + } + return pd.DataFrame(data=fake_airway) with open("src/sql/airway.sql", "r") as file: airway_query = sql.SQL(file.read()) parameters = { From 33e2c08b164aa713c5cdce238dcd0511315aee35 Mon Sep 17 00:00:00 2001 From: Stephen Thompson Date: Tue, 18 Aug 2026 15:21:33 +0100 Subject: [PATCH 14/35] Caboodle setup goes into exporter.env --- config.EXAMPLE/ehr_lookup.EXAMPLE | 14 -------------- config.EXAMPLE/exporter.env.EXAMPLE | 14 ++++++++++++++ 2 files changed, 14 insertions(+), 14 deletions(-) delete mode 100644 config.EXAMPLE/ehr_lookup.EXAMPLE diff --git a/config.EXAMPLE/ehr_lookup.EXAMPLE b/config.EXAMPLE/ehr_lookup.EXAMPLE deleted file mode 100644 index 0c88d12..0000000 --- a/config.EXAMPLE/ehr_lookup.EXAMPLE +++ /dev/null @@ -1,14 +0,0 @@ -# This is an EXAMPLE file, do not put real secrets in here. -# Copy it to ../config/ehr_lookup.env and then DELETE THIS COMMENT. -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" diff --git a/config.EXAMPLE/exporter.env.EXAMPLE b/config.EXAMPLE/exporter.env.EXAMPLE index 39a4119..c49d503 100644 --- a/config.EXAMPLE/exporter.env.EXAMPLE +++ b/config.EXAMPLE/exporter.env.EXAMPLE @@ -32,3 +32,17 @@ ONLY_USE_CSV_FROM_YESTERDAY=TRUE # specify a date to process format YYYY-MM-DD also accepts a regular # 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" From 9b291cb257a86c51802fcf8bdfeb17efe3e53b8d Mon Sep 17 00:00:00 2001 From: Stephen Thompson Date: Tue, 18 Aug 2026 15:51:40 +0100 Subject: [PATCH 15/35] Format fix --- src/db.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/src/db.py b/src/db.py index e0fa11f..04eb37a 100644 --- a/src/db.py +++ b/src/db.py @@ -76,7 +76,9 @@ def connect(self): """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) + self.connection_pool = pool.SimpleConnectionPool( + 1, 1, self.connection_string + ) def get_airflow(self, start_datetime: datetime, end_datetime: datetime, csn: str): """Retrieve airflow data from database.""" From d4fde5905926162f3ab8fdd978556b93c175d4a0 Mon Sep 17 00:00:00 2001 From: Stephen Thompson Date: Tue, 18 Aug 2026 16:06:02 +0100 Subject: [PATCH 16/35] Beginning of test for ehr integration --- src/csv_writer.py | 2 +- tests/helpers.py | 3 +++ tests/test_snakemake_integration.py | 2 ++ 3 files changed, 6 insertions(+), 1 deletion(-) diff --git a/src/csv_writer.py b/src/csv_writer.py index 42daeb3..11588b1 100644 --- a/src/csv_writer.py +++ b/src/csv_writer.py @@ -100,6 +100,6 @@ def write_ehr( filename = WAVEFORM_PSEUDONYMISED_EHR / f"{stem}_ehr.csv" filename.parent.mkdir(exist_ok=True, parents=True) - df.to_csv(filename) + df.to_csv(filename, index = False) return True diff --git a/tests/helpers.py b/tests/helpers.py index 63d82f2..68df5cc 100644 --- a/tests/helpers.py +++ b/tests/helpers.py @@ -51,6 +51,9 @@ def get_orig_parquet(self): def get_pseudon_parquet(self): return f"{self.date}/{self.date}.{self.get_hashed_csn()}.{self.variable_id}.{self.channel_id}.{self.units}.parquet" + def get_pseudon_ehr(self): + return f"{self.date}/{self.date}.{self.get_hashed_csn()}_ehr.csv" + def get_hashes(self): return f"{self.date}/{self.date}.hashes.json" diff --git a/tests/test_snakemake_integration.py b/tests/test_snakemake_integration.py index 50d7b9c..3bf8a4d 100644 --- a/tests/test_snakemake_integration.py +++ b/tests/test_snakemake_integration.py @@ -207,9 +207,11 @@ def test_snakemake_pipeline(tmp_path: Path, background_hasher): tmp_path / "original-parquet" / filename.get_orig_parquet() ) pseudon_path = tmp_path / "pseudonymised" / filename.get_pseudon_parquet() + ehr_path = tmp_path / "pseudonymised_ehr" / filename.get_pseudon_ehr() assert original_parquet_path.exists() assert pseudon_path.exists() + assert ehr_path.exists() _compare_original_parquet_to_expected(original_parquet_path, expected_data) _compare_parquets(original_parquet_path, pseudon_path) From d910831441fe7bb29bfb76706e184f9921bdbe0b Mon Sep 17 00:00:00 2001 From: Stephen Thompson Date: Tue, 18 Aug 2026 16:25:30 +0100 Subject: [PATCH 17/35] SQL formatting --- sql_scripts/flow_sheet_values.sql | 1 - sql_scripts/get_hospital_visit_id.sql | 2 -- sql_scripts/lab_results.sql | 1 - sql_scripts/lab_test_names.sql | 2 +- 4 files changed, 1 insertion(+), 5 deletions(-) diff --git a/sql_scripts/flow_sheet_values.sql b/sql_scripts/flow_sheet_values.sql index af9b76f..3be8624 100644 --- a/sql_scripts/flow_sheet_values.sql +++ b/sql_scripts/flow_sheet_values.sql @@ -32,4 +32,3 @@ AND vo.valid_from BETWEEN %(yesterday)s AND %(today)s AND vo.hospital_visit_id = %(hospital_visit_id)s GROUP BY DateTimeRecorded, Units, vo.comment - diff --git a/sql_scripts/get_hospital_visit_id.sql b/sql_scripts/get_hospital_visit_id.sql index 408447f..862a67f 100644 --- a/sql_scripts/get_hospital_visit_id.sql +++ b/sql_scripts/get_hospital_visit_id.sql @@ -3,5 +3,3 @@ select hospital_visit_id from star.hospital_visit as hv where hv.encounter = %(csn)s -- note the CSN must be in quotes - - diff --git a/sql_scripts/lab_results.sql b/sql_scripts/lab_results.sql index aa05264..81783ce 100644 --- a/sql_scripts/lab_results.sql +++ b/sql_scripts/lab_results.sql @@ -21,4 +21,3 @@ on r.lab_order_id = o.lab_order_id where o.hospital_visit_id = 'xx' and r.result_status like 'FINAL' and r.lab_test_definition_id in ('1001', '390793054', '390793057', '390793060', '722790196') - diff --git a/sql_scripts/lab_test_names.sql b/sql_scripts/lab_test_names.sql index 85ce715..1838b72 100644 --- a/sql_scripts/lab_test_names.sql +++ b/sql_scripts/lab_test_names.sql @@ -2,4 +2,4 @@ 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') \ No newline at end of file +where ltd.lab_test_definition_id in ('1001', '390793054', '390793057', '390793060', '722790196') From e8f29a73f7d15fa9c07578bafc2d8dc5deb98b1a Mon Sep 17 00:00:00 2001 From: Stephen Thompson Date: Tue, 18 Aug 2026 16:44:13 +0100 Subject: [PATCH 18/35] Test with ehr lookup --- tests/test_snakemake_integration.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/test_snakemake_integration.py b/tests/test_snakemake_integration.py index 3bf8a4d..843ca1c 100644 --- a/tests/test_snakemake_integration.py +++ b/tests/test_snakemake_integration.py @@ -268,7 +268,7 @@ def _run_snakemake(tmp_path): tmp_exporter_env_path = tmp_path / "config/exporter.env" tmp_exporter_env_path.parent.mkdir(exist_ok=True) tmp_exporter_env_path.write_text( - "SNAKEMAKE_RULE_UNTIL=all_daily_hash_lookups\n" + "SNAKEMAKE_RULE_UNTIL=all_ehr_lookups\n" "SNAKEMAKE_CORES=1\n" "INSTANCE_NAME=pytest\n" "CSV_AGE_THRESHOLD_MINUTES=5\n" From 86d5dea4f8beb05b22ef1a5089ee8a25d03b67ad Mon Sep 17 00:00:00 2001 From: Stephen Thompson Date: Tue, 18 Aug 2026 16:55:20 +0100 Subject: [PATCH 19/35] use caboodle testing --- tests/test_snakemake_integration.py | 1 + 1 file changed, 1 insertion(+) diff --git a/tests/test_snakemake_integration.py b/tests/test_snakemake_integration.py index 843ca1c..908fed5 100644 --- a/tests/test_snakemake_integration.py +++ b/tests/test_snakemake_integration.py @@ -274,6 +274,7 @@ def _run_snakemake(tmp_path): "CSV_AGE_THRESHOLD_MINUTES=5\n" "ONLY_USE_CSV_FROM_YESTERDAY=False\n" "PROCESS_CSV_FROM_DATE=\n" + "CABOODLE_TESTING=TRUE\n" ) # run system under test (exporter container) in foreground compose_args = [ From 7fbcfc043fc685f0446011f7614c574195570773 Mon Sep 17 00:00:00 2001 From: Stephen Thompson Date: Tue, 18 Aug 2026 17:18:10 +0100 Subject: [PATCH 20/35] Trying to test ehr and hash lookup with ftps --- src/pipeline/Snakefile | 6 ++++++ tests/test_snakemake_integration.py | 2 +- 2 files changed, 7 insertions(+), 1 deletion(-) diff --git a/src/pipeline/Snakefile b/src/pipeline/Snakefile index 71042d9..30eb607 100644 --- a/src/pipeline/Snakefile +++ b/src/pipeline/Snakefile @@ -83,6 +83,12 @@ rule all_ehr_lookups: input: ALL_EHR_LOOKUPS +# a rule combining ehr and hash look ups to enable testing without ftps upload +rule all_ehr_and_hash_lookups: + input: + ALL_EHR_LOOKUPS, + ALL_DAILY_HASH_LOOKUPS + def input_file_maker(wc): unhashed_csn = hash_to_csn[wc.hashed_csn] # when using input functions, snakemake doesn't do its normal templating, you have to do it, hence the f-string diff --git a/tests/test_snakemake_integration.py b/tests/test_snakemake_integration.py index 908fed5..522bb76 100644 --- a/tests/test_snakemake_integration.py +++ b/tests/test_snakemake_integration.py @@ -268,7 +268,7 @@ def _run_snakemake(tmp_path): tmp_exporter_env_path = tmp_path / "config/exporter.env" tmp_exporter_env_path.parent.mkdir(exist_ok=True) tmp_exporter_env_path.write_text( - "SNAKEMAKE_RULE_UNTIL=all_ehr_lookups\n" + "SNAKEMAKE_RULE_UNTIL=all_ehr_and_hash_lookups\n" "SNAKEMAKE_CORES=1\n" "INSTANCE_NAME=pytest\n" "CSV_AGE_THRESHOLD_MINUTES=5\n" From af487e6690af65dbed5089fbd2b6d87934730a31 Mon Sep 17 00:00:00 2001 From: Stephen Thompson Date: Wed, 19 Aug 2026 14:42:36 +0100 Subject: [PATCH 21/35] Moved sql --- {sql_scripts => src/sql}/flow_sheet_values.sql | 0 {sql_scripts => src/sql}/get_hospital_visit_id.sql | 0 {sql_scripts => src/sql}/sputum_secretions.sql | 0 3 files changed, 0 insertions(+), 0 deletions(-) rename {sql_scripts => src/sql}/flow_sheet_values.sql (100%) rename {sql_scripts => src/sql}/get_hospital_visit_id.sql (100%) rename {sql_scripts => src/sql}/sputum_secretions.sql (100%) diff --git a/sql_scripts/flow_sheet_values.sql b/src/sql/flow_sheet_values.sql similarity index 100% rename from sql_scripts/flow_sheet_values.sql rename to src/sql/flow_sheet_values.sql diff --git a/sql_scripts/get_hospital_visit_id.sql b/src/sql/get_hospital_visit_id.sql similarity index 100% rename from sql_scripts/get_hospital_visit_id.sql rename to src/sql/get_hospital_visit_id.sql diff --git a/sql_scripts/sputum_secretions.sql b/src/sql/sputum_secretions.sql similarity index 100% rename from sql_scripts/sputum_secretions.sql rename to src/sql/sputum_secretions.sql From b5351913977f9666270cd2d32b346f10d8dac040 Mon Sep 17 00:00:00 2001 From: Stephen Thompson Date: Wed, 19 Aug 2026 15:43:45 +0100 Subject: [PATCH 22/35] refactored db to make adding ehr lookup implementation clearer --- src/controller.py | 7 ++-- src/db.py | 49 ++++++++++++++--------- src/sql/mrn_based_on_bed_and_datetime.sql | 8 ++-- 3 files changed, 39 insertions(+), 25 deletions(-) diff --git a/src/controller.py b/src/controller.py index 3dc3099..956be16 100644 --- a/src/controller.py +++ b/src/controller.py @@ -43,7 +43,6 @@ def reject_message(ch, delivery_tag, requeue): class WaveformController: def __init__(self): self.emap_db = db.starDB() - self.emap_db.init_query() self.emap_db.connect() def waveform_callback(self, ch, method_frame, _header_frame, body): @@ -76,7 +75,9 @@ def waveform_callback(self, ch, method_frame, _header_frame, body): ) 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( @@ -86,6 +87,7 @@ def waveform_callback(self, ch, method_frame, _header_frame, body): 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) reject_message(ch, method_frame.delivery_tag, True) @@ -96,7 +98,6 @@ def waveform_callback(self, ch, method_frame, _header_frame, body): logger.info("Research opt-out is set for mrn %s, not writing.", mrn) reject_message(ch, method_frame.delivery_tag, False) return - if writer.write_frame( waveform_data, source_variable_id, diff --git a/src/db.py b/src/db.py index 04eb37a..f6d78f6 100644 --- a/src/db.py +++ b/src/db.py @@ -11,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 @@ -21,27 +21,32 @@ class starDB: settings.UDS_CONNECT_TIMEOUT, # type:ignore settings.UDS_QUERY_TIMEOUT, # type:ignore ) - connection_pool: pool.ThreadedConnectionPool + connection_pool: pool.SimpleConnectionPool - def connect(self): + def connect(self) -> None: self.connection_pool = pool.SimpleConnectionPool(1, 1, self.connection_string) - def init_query(self): + def _init_mrn_lookup_query(self) -> None: 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 = sql.SQL(file.read()) # type:ignore + + 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() try: with self.connection_pool.getconn() as db_connection: with db_connection.cursor() as curs: - curs.execute(self.sql_query, parameters) + curs.execute(self.mrn_lookup_query, parameters) rows = curs.fetchall() self.connection_pool.putconn(db_connection) except psycopg2.errors.OperationalError as e: @@ -55,6 +60,9 @@ def get_row(self, location_string: str, observation_datetime: datetime): return rows[0] + def get_hospital_visit_from_csn(self, csn: str) -> str: + return "not implemented yet" + class caboodleDB: """For querying the caboodle database to extract electronic healthcare records per @@ -69,10 +77,10 @@ class caboodleDB: settings.CABOODLE_CONNECT_TIMEOUT, # type:ignore settings.CABOODLE_QUERY_TIMEOUT, # type:ignore ) - connection_pool: pool.ThreadedConnectionPool + connection_pool: pool.SimpleConnectionPool fake_caboodle: bool - def connect(self): + 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: @@ -80,8 +88,19 @@ def connect(self): 1, 1, self.connection_string ) - def get_airflow(self, start_datetime: datetime, end_datetime: datetime, csn: str): + def get_airflow( + self, start_datetime: datetime, end_datetime: datetime, csn: str + ) -> pd.DataFrame: """Retrieve airflow data from database.""" + + with open("src/sql/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], @@ -90,13 +109,7 @@ def get_airflow(self, start_datetime: datetime, end_datetime: datetime, csn: str "TubeSize": [0], } return pd.DataFrame(data=fake_airway) - with open("src/sql/airway.sql", "r") as file: - airway_query = sql.SQL(file.read()) - parameters = { - "start_datetime": start_datetime, - "end_datetime": end_datetime, - "csn": csn, - } + return self._get_rows(airway_query, parameters) def _get_rows(self, sql_query: sql.SQL, parameters: dict): diff --git a/src/sql/mrn_based_on_bed_and_datetime.sql b/src/sql/mrn_based_on_bed_and_datetime.sql index 7eccf5e..4494d21 100644 --- a/src/sql/mrn_based_on_bed_and_datetime.sql +++ b/src/sql/mrn_based_on_bed_and_datetime.sql @@ -1,7 +1,7 @@ -/* Find a medical record number (MRN), NHS number, and contact serial number (CSN) based on location -string and date time. Returns a list of MRN, NHS numbers, and CSN with the -first entry being the most recent. -*/ +-- Find a medical record number (MRN), NHS number, and contact serial number (CSN) based on location +-- string and date time. Returns a list of MRN, NHS numbers, and CSN with the +-- first entry being the most recent. +-- SELECT mn.mrn as mrn, mn.nhs_number as nhs_number, From 4f0ba6f57446119d44f3af146667756b81ad9cd4 Mon Sep 17 00:00:00 2001 From: Stephen Thompson Date: Wed, 19 Aug 2026 16:22:42 +0100 Subject: [PATCH 23/35] Started on logic to get flowsheets, first need hospital visit number --- src/db.py | 33 +++++++++++++++++++-------- src/electronic_health_records/ehr.py | 34 +++++++++++++++------------- src/sql/get_hospital_visit_id.sql | 5 ++-- 3 files changed, 43 insertions(+), 29 deletions(-) diff --git a/src/db.py b/src/db.py index f6d78f6..13d8a3c 100644 --- a/src/db.py +++ b/src/db.py @@ -43,15 +43,8 @@ def get_matched_mrn( } if self.mrn_lookup_query == "": self._init_mrn_lookup_query() - try: - with self.connection_pool.getconn() as db_connection: - with db_connection.cursor() as curs: - curs.execute(self.mrn_lookup_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}") + + rows = self._get_rows(self.mrn_lookup_query, parameters) # type: ignore if len(rows) != 1: raise ValueError( @@ -61,7 +54,27 @@ def get_matched_mrn( return rows[0] def get_hospital_visit_from_csn(self, csn: str) -> str: - return "not implemented yet" + with open("src/sql/get_hospital_visit_id.sql", "r") as file: + hv_query = sql.SQL(file.read()) + + parameters = { + "schema": settings.SCHEMA_NAME, + "csn": csn, + } + + return self._get_rows(hv_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 class caboodleDB: diff --git a/src/electronic_health_records/ehr.py b/src/electronic_health_records/ehr.py index 48fc930..7470282 100644 --- a/src/electronic_health_records/ehr.py +++ b/src/electronic_health_records/ehr.py @@ -2,7 +2,7 @@ from datetime import datetime, timedelta -from db import caboodleDB +from db import caboodleDB, starDB from csv_writer import write_ehr from pseudon.pseudon import pseudonymise_relevant_columns @@ -18,28 +18,24 @@ def ehr_for_csv(date_str: str, original_csn: str, hashed_csn: str) -> None: :param hashed_csn: the pseudonymised hash to use for file output. """ - db_connection = caboodleDB() - db_connection.connect() + caboodle_connection = caboodleDB() + caboodle_connection.connect() - _ehr_for_csv(date_str, original_csn, hashed_csn, db_connection) + star_connection = starDB() + star_connection.connect() + + _ehr_for_csv( + date_str, original_csn, hashed_csn, caboodle_connection, star_connection + ) def _ehr_for_csv( date_str: str, original_csn: str, hashed_csn: str, - db_connection: caboodleDB, + caboodle_connection: caboodleDB, + star_connection: starDB, ) -> None: - """Extracts electronic healthcare records for a given csn and writes the results to - a pseudonymised csv file for a single day. - - This is a privacy-sensitive area of code. Unhashed CSNs must not appear in uploaded - files. - :param date_str: the date to look up data for - :param original_csn: the csn to base look up on. - :param hashed_csn: the pseudonymised hash to use for file output. - :param db_connection: connection to the caboodle database. - """ # will pick up the logger config defined in the snakemake job (ie. log to file) logger = logging.getLogger(__name__) @@ -47,7 +43,13 @@ def _ehr_for_csv( start_datetime = datetime.strptime(date_str, "%Y-%m-%d") end_datetime = start_datetime + timedelta(days=1) - airflow = db_connection.get_airflow(start_datetime, end_datetime, original_csn) + airflow = caboodle_connection.get_airflow( + start_datetime, end_datetime, original_csn + ) + + hospital_visit = star_connection.get_hospital_visit_from_csn(original_csn) + + logger.info(hospital_visit) safe_columns = [ "DateTimeRecorded", diff --git a/src/sql/get_hospital_visit_id.sql b/src/sql/get_hospital_visit_id.sql index 862a67f..15af7a0 100644 --- a/src/sql/get_hospital_visit_id.sql +++ b/src/sql/get_hospital_visit_id.sql @@ -1,5 +1,4 @@ - +-- Retrieve the hospital_visit_id associated with the csn value applied to this function -- - -select hospital_visit_id from star.hospital_visit as hv +select hospital_visit_id from %(schema)s.hospital_visit as hv where hv.encounter = %(csn)s -- note the CSN must be in quotes From 88ba017e6ddfa890aa098758b07636d66b22475b Mon Sep 17 00:00:00 2001 From: Stephen Thompson Date: Wed, 19 Aug 2026 16:47:14 +0100 Subject: [PATCH 24/35] The exporter needs to access uds, do duplicated some of the config from controller --- config.EXAMPLE/exporter.env.EXAMPLE | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/config.EXAMPLE/exporter.env.EXAMPLE b/config.EXAMPLE/exporter.env.EXAMPLE index c49d503..ed8efeb 100644 --- a/config.EXAMPLE/exporter.env.EXAMPLE +++ b/config.EXAMPLE/exporter.env.EXAMPLE @@ -46,3 +46,14 @@ CABOODLE_QUERY_TIMEOUT="3000" # in milliseconds # 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" From 72f4935dfb5f616133550a42f5faa795543e7449 Mon Sep 17 00:00:00 2001 From: Stephen Thompson Date: Thu, 20 Aug 2026 08:22:42 +0100 Subject: [PATCH 25/35] Use enviroment to manage sql locations as it varies between containers --- config.EXAMPLE/controller.env.EXAMPLE | 2 ++ config.EXAMPLE/exporter.env.EXAMPLE | 2 ++ src/db.py | 6 +++--- src/settings.py | 1 + 4 files changed, 8 insertions(+), 3 deletions(-) diff --git a/config.EXAMPLE/controller.env.EXAMPLE b/config.EXAMPLE/controller.env.EXAMPLE index e055216..66d9672 100644 --- a/config.EXAMPLE/controller.env.EXAMPLE +++ b/config.EXAMPLE/controller.env.EXAMPLE @@ -13,3 +13,5 @@ RABBITMQ_PASSWORD="my_pw" RABBITMQ_HOST="localhost" RABBITMQ_PORT=5672 RABBITMQ_QUEUE="waveform" + +SQL_PATH="./src/sql/" diff --git a/config.EXAMPLE/exporter.env.EXAMPLE b/config.EXAMPLE/exporter.env.EXAMPLE index ed8efeb..e8ed0f9 100644 --- a/config.EXAMPLE/exporter.env.EXAMPLE +++ b/config.EXAMPLE/exporter.env.EXAMPLE @@ -57,3 +57,5 @@ UDS_PORT="5433" UDS_CONNECT_TIMEOUT="10" UDS_QUERY_TIMEOUT="3000" SCHEMA_NAME="schemaname" + +SQL_PATH="/app/src/sql/" diff --git a/src/db.py b/src/db.py index 13d8a3c..1d62798 100644 --- a/src/db.py +++ b/src/db.py @@ -27,7 +27,7 @@ def connect(self) -> None: self.connection_pool = pool.SimpleConnectionPool(1, 1, self.connection_string) def _init_mrn_lookup_query(self) -> None: - with open("src/sql/mrn_based_on_bed_and_datetime.sql", "r") as file: + 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 self.mrn_lookup_query = self.mrn_lookup_query.format( @@ -54,7 +54,7 @@ def get_matched_mrn( return rows[0] def get_hospital_visit_from_csn(self, csn: str) -> str: - with open("src/sql/get_hospital_visit_id.sql", "r") as file: + with open(settings.SQL_PATH + "get_hospital_visit_id.sql", "r") as file: hv_query = sql.SQL(file.read()) parameters = { @@ -106,7 +106,7 @@ def get_airflow( ) -> pd.DataFrame: """Retrieve airflow data from database.""" - with open("src/sql/airway.sql", "r") as file: + with open(settings.SQL_PATH + "airway.sql", "r") as file: airway_query = sql.SQL(file.read()) parameters = { "start_datetime": start_datetime, diff --git a/src/settings.py b/src/settings.py index 0309839..df0322c 100644 --- a/src/settings.py +++ b/src/settings.py @@ -49,3 +49,4 @@ def get_from_env(env_var, *, default_value=None, setting_name=None, required=Fal get_from_env("LOG_LEVEL", default_value="INFO") get_from_env("INSTANCE_NAME", required=True) +get_from_env("SQL_PATH") From 29865e01ade17f6c2fa9a4b324016d5f9db1cfa2 Mon Sep 17 00:00:00 2001 From: Stephen Thompson Date: Thu, 20 Aug 2026 08:24:50 +0100 Subject: [PATCH 26/35] path into test --- tests/test_snakemake_integration.py | 1 + 1 file changed, 1 insertion(+) diff --git a/tests/test_snakemake_integration.py b/tests/test_snakemake_integration.py index 522bb76..68a023c 100644 --- a/tests/test_snakemake_integration.py +++ b/tests/test_snakemake_integration.py @@ -275,6 +275,7 @@ def _run_snakemake(tmp_path): "ONLY_USE_CSV_FROM_YESTERDAY=False\n" "PROCESS_CSV_FROM_DATE=\n" "CABOODLE_TESTING=TRUE\n" + "SQL_PATH=/app/src/sql/\n" ) # run system under test (exporter container) in foreground compose_args = [ From de9442254430c1eeb0729b72a339bc6be3b7a253 Mon Sep 17 00:00:00 2001 From: Stephen Thompson Date: Thu, 20 Aug 2026 08:32:59 +0100 Subject: [PATCH 27/35] Use format rather than parameters for schema name --- src/db.py | 3 ++- src/sql/get_hospital_visit_id.sql | 2 +- 2 files changed, 3 insertions(+), 2 deletions(-) diff --git a/src/db.py b/src/db.py index 1d62798..d826231 100644 --- a/src/db.py +++ b/src/db.py @@ -57,8 +57,9 @@ def get_hospital_visit_from_csn(self, csn: str) -> str: 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 = { - "schema": settings.SCHEMA_NAME, "csn": csn, } diff --git a/src/sql/get_hospital_visit_id.sql b/src/sql/get_hospital_visit_id.sql index 15af7a0..8cd2f79 100644 --- a/src/sql/get_hospital_visit_id.sql +++ b/src/sql/get_hospital_visit_id.sql @@ -1,4 +1,4 @@ -- Retrieve the hospital_visit_id associated with the csn value applied to this function -- -select hospital_visit_id from %(schema)s.hospital_visit as hv +select hospital_visit_id from {schema_name}.hospital_visit as hv where hv.encounter = %(csn)s -- note the CSN must be in quotes From 518060b2f5deae0303aef533de4c7d355fcdda1e Mon Sep 17 00:00:00 2001 From: Stephen Thompson Date: Thu, 20 Aug 2026 09:00:32 +0100 Subject: [PATCH 28/35] Default for sql path to simplify testing --- src/settings.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/settings.py b/src/settings.py index df0322c..151306b 100644 --- a/src/settings.py +++ b/src/settings.py @@ -49,4 +49,4 @@ def get_from_env(env_var, *, default_value=None, setting_name=None, required=Fal get_from_env("LOG_LEVEL", default_value="INFO") get_from_env("INSTANCE_NAME", required=True) -get_from_env("SQL_PATH") +get_from_env("SQL_PATH", default_value="./src/sql/") From 03dbb608f1c8d201fcb667a4600e7477b8bb991d Mon Sep 17 00:00:00 2001 From: Stephen Thompson Date: Thu, 20 Aug 2026 09:14:56 +0100 Subject: [PATCH 29/35] Updated mocked function name in test controller --- src/settings.py | 2 +- tests/test_controller.py | 10 ++++++---- 2 files changed, 7 insertions(+), 5 deletions(-) diff --git a/src/settings.py b/src/settings.py index 151306b..df0322c 100644 --- a/src/settings.py +++ b/src/settings.py @@ -49,4 +49,4 @@ def get_from_env(env_var, *, default_value=None, setting_name=None, required=Fal get_from_env("LOG_LEVEL", default_value="INFO") get_from_env("INSTANCE_NAME", required=True) -get_from_env("SQL_PATH", default_value="./src/sql/") +get_from_env("SQL_PATH") diff --git a/tests/test_controller.py b/tests/test_controller.py index d4fd611..3bee1e2 100644 --- a/tests/test_controller.py +++ b/tests/test_controller.py @@ -22,9 +22,11 @@ def test_controller_callback(monkeypatch, opt_out, db_connect_failure, bad_data): emap_db_mock = Mock() if db_connect_failure: - emap_db_mock.get_row.side_effect = ConnectionError("mock database error") + emap_db_mock.get_matched_mrn.side_effect = ConnectionError( + "mock database error" + ) else: - emap_db_mock.get_row.return_value = ("mrn", "nhsno", "csn", opt_out) + emap_db_mock.get_matched_mrn.return_value = ("mrn", "nhsno", "csn", opt_out) monkeypatch.setattr("controller.db.starDB", Mock(return_value=emap_db_mock)) write_frame_mock = Mock(return_value=True) @@ -56,12 +58,12 @@ def test_controller_callback(monkeypatch, opt_out, db_connect_failure, bad_data) if not bad_data: # we at least tried to query the DB - emap_db_mock.get_row.assert_called_once() + emap_db_mock.get_matched_mrn.assert_called_once() if bad_data: write_frame_mock.assert_not_called() # db should not even have been queried if data was bad - emap_db_mock.get_row.assert_not_called() + emap_db_mock.get_matched_mrn.assert_not_called() channel_mock.basic_reject.assert_called_once_with(delivery_tag, False) channel_mock.basic_ack.assert_not_called() elif db_connect_failure: From c08785c6d84b433106eb72be4ec1bd6d4c2ba397 Mon Sep 17 00:00:00 2001 From: Stephen Thompson Date: Mon, 24 Aug 2026 09:07:45 +0100 Subject: [PATCH 30/35] Enable a fake star db shortcut so that snakemake integration test can run --- src/db.py | 12 +++++++++++- src/settings.py | 1 + tests/test_snakemake_integration.py | 1 + 3 files changed, 13 insertions(+), 1 deletion(-) diff --git a/src/db.py b/src/db.py index d826231..1d2459b 100644 --- a/src/db.py +++ b/src/db.py @@ -22,8 +22,16 @@ class starDB: settings.UDS_QUERY_TIMEOUT, # type:ignore ) connection_pool: pool.SimpleConnectionPool + fake_star: bool = False 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 + ) + + self.connection_pool = pool.SimpleConnectionPool(1, 1, self.connection_string) def _init_mrn_lookup_query(self) -> None: @@ -62,6 +70,8 @@ def get_hospital_visit_from_csn(self, csn: str) -> str: parameters = { "csn": csn, } + if self.fake_star: + return '12345678' return self._get_rows(hv_query, parameters) @@ -92,7 +102,7 @@ class caboodleDB: settings.CABOODLE_QUERY_TIMEOUT, # type:ignore ) connection_pool: pool.SimpleConnectionPool - fake_caboodle: bool + fake_caboodle: bool = False def connect(self) -> None: """Set up connection to the database.""" diff --git a/src/settings.py b/src/settings.py index df0322c..c70c05a 100644 --- a/src/settings.py +++ b/src/settings.py @@ -22,6 +22,7 @@ def get_from_env(env_var, *, default_value=None, setting_name=None, required=Fal get_from_env("UDS_PORT") get_from_env("UDS_CONNECT_TIMEOUT") get_from_env("UDS_QUERY_TIMEOUT") +get_from_env("STARDB_TESTING") get_from_env("SCHEMA_NAME") get_from_env("RABBITMQ_USERNAME") get_from_env("RABBITMQ_PASSWORD") diff --git a/tests/test_snakemake_integration.py b/tests/test_snakemake_integration.py index 68a023c..aa00818 100644 --- a/tests/test_snakemake_integration.py +++ b/tests/test_snakemake_integration.py @@ -274,6 +274,7 @@ def _run_snakemake(tmp_path): "CSV_AGE_THRESHOLD_MINUTES=5\n" "ONLY_USE_CSV_FROM_YESTERDAY=False\n" "PROCESS_CSV_FROM_DATE=\n" + "STARDB_TESTING=TRUE\n" "CABOODLE_TESTING=TRUE\n" "SQL_PATH=/app/src/sql/\n" ) From cccf07804689c49753a4b715049928c571845ef0 Mon Sep 17 00:00:00 2001 From: Stephen Thompson Date: Mon, 24 Aug 2026 09:37:10 +0100 Subject: [PATCH 31/35] Don't connect for fake star --- src/db.py | 5 +---- 1 file changed, 1 insertion(+), 4 deletions(-) diff --git a/src/db.py b/src/db.py index 1d2459b..59cdcae 100644 --- a/src/db.py +++ b/src/db.py @@ -31,9 +31,6 @@ def connect(self) -> None: 1, 1, self.connection_string ) - - 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 @@ -71,7 +68,7 @@ def get_hospital_visit_from_csn(self, csn: str) -> str: "csn": csn, } if self.fake_star: - return '12345678' + return "12345678" return self._get_rows(hv_query, parameters) From c84ca84abb6a7687578cbb3d7ff777ed8aeef3f5 Mon Sep 17 00:00:00 2001 From: Sarah Keating Date: Mon, 24 Aug 2026 11:13:57 +0100 Subject: [PATCH 32/35] Refactor lab results query for wide format output Transforms specific lab test results (e.g., CRP, WCC) from a long format into distinct columns using `MAX() FILTER`. This enables direct consumption of lab data in a wider format, simplifying downstream analysis. Additionally, the query is updated to use parameters for `hospital_visit_id` and to include date range filtering, improving its flexibility. --- sql_scripts/lab_results.sql | 41 +++++++++++++++++++++++-------------- 1 file changed, 26 insertions(+), 15 deletions(-) diff --git a/sql_scripts/lab_results.sql b/sql_scripts/lab_results.sql index aa05264..21b8b5d 100644 --- a/sql_scripts/lab_results.sql +++ b/sql_scripts/lab_results.sql @@ -1,24 +1,35 @@ --- This selects the values are units of lab tests +-- This selects the values of lab tests -- 1011 CRP -- 722790196 CRP -- 390793054 WCC -- 390793057 WCC -- 390793060 WCC -select -r.result_last_modified_datetime as date, -(select name from star.lab_test_definition as ltd -where ltd.lab_test_definition_id = r.lab_test_definition_id) as name, -r.value_as_real as value, -r.units, -r.abnormal_flag, -r.comment +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" -from star.lab_result as r -join star.lab_order as o -on r.lab_order_id = o.lab_order_id -where o.hospital_visit_id = 'xx' -and r.result_status like 'FINAL' -and r.lab_test_definition_id in ('1001', '390793054', '390793057', '390793060', '722790196') + 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 From 9ae3715aa4419b08cb79643a42ab439ef5f57c39 Mon Sep 17 00:00:00 2001 From: Stephen Thompson Date: Wed, 26 Aug 2026 08:59:09 +0100 Subject: [PATCH 33/35] Tidied up getting for hospital visit number --- src/db.py | 8 +++++++- src/settings.py | 2 +- 2 files changed, 8 insertions(+), 2 deletions(-) diff --git a/src/db.py b/src/db.py index 59cdcae..25c8e0d 100644 --- a/src/db.py +++ b/src/db.py @@ -70,7 +70,13 @@ def get_hospital_visit_from_csn(self, csn: str) -> str: if self.fake_star: return "12345678" - return self._get_rows(hv_query, parameters) + 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], str): + raise TypeError(f"hospital_visit_id is not string {hospital_visit_id}") + + return hospital_visit_id[0][0] def _get_rows(self, sql_query: sql.SQL, parameters: dict): try: diff --git a/src/settings.py b/src/settings.py index c70c05a..8f6cc91 100644 --- a/src/settings.py +++ b/src/settings.py @@ -50,4 +50,4 @@ def get_from_env(env_var, *, default_value=None, setting_name=None, required=Fal get_from_env("LOG_LEVEL", default_value="INFO") get_from_env("INSTANCE_NAME", required=True) -get_from_env("SQL_PATH") +get_from_env("SQL_PATH", default_value="./src/sql/") From bf90f902edefc9d10cff4bf8f53d550e31d47862 Mon Sep 17 00:00:00 2001 From: Stephen Thompson Date: Wed, 26 Aug 2026 11:12:40 +0100 Subject: [PATCH 34/35] return integer hospital visit id --- src/db.py | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/src/db.py b/src/db.py index 25c8e0d..acbd855 100644 --- a/src/db.py +++ b/src/db.py @@ -58,7 +58,7 @@ def get_matched_mrn( return rows[0] - def get_hospital_visit_from_csn(self, csn: str) -> str: + 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()) @@ -68,13 +68,15 @@ def get_hospital_visit_from_csn(self, csn: str) -> str: "csn": csn, } if self.fake_star: - return "12345678" + 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], str): - raise TypeError(f"hospital_visit_id is not string {hospital_visit_id}") + 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] From aa8f86ab666e32efb14d757845593c8f926d1234 Mon Sep 17 00:00:00 2001 From: Stephen Thompson Date: Wed, 26 Aug 2026 15:02:11 +0100 Subject: [PATCH 35/35] Added flowsheet query to ehr --- src/db.py | 24 ++++++++++++++++++++++++ src/electronic_health_records/ehr.py | 20 +++++++++++++++----- src/sql/flow_sheet_values.sql | 2 +- 3 files changed, 40 insertions(+), 6 deletions(-) diff --git a/src/db.py b/src/db.py index acbd855..abaf3dd 100644 --- a/src/db.py +++ b/src/db.py @@ -141,6 +141,30 @@ def get_airflow( 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: diff --git a/src/electronic_health_records/ehr.py b/src/electronic_health_records/ehr.py index 7470282..eefc830 100644 --- a/src/electronic_health_records/ehr.py +++ b/src/electronic_health_records/ehr.py @@ -1,6 +1,7 @@ import logging from datetime import datetime, timedelta +import pandas as pd from db import caboodleDB, starDB from csv_writer import write_ehr @@ -47,22 +48,31 @@ def _ehr_for_csv( start_datetime, end_datetime, original_csn ) - hospital_visit = star_connection.get_hospital_visit_from_csn(original_csn) + hospital_visit_id = star_connection.get_hospital_visit_from_csn(original_csn) - logger.info(hospital_visit) + logger.info(hospital_visit_id) + + flowsheet_values = caboodle_connection.get_flowsheets( + start_datetime, end_datetime, hospital_visit_id + ) + + ehr_data = pd.concat([airflow, flowsheet_values]) safe_columns = [ "DateTimeRecorded", "PlacementInstant", "RemovalInstant", "TubeSize", + "Temperature", + "Noradrenaline", + "Metaraminol", ] - airflow = pseudonymise_relevant_columns(airflow, safe_columns) + ehr_data = pseudonymise_relevant_columns(ehr_data, safe_columns) - write_ehr(airflow, date_str, hashed_csn) + write_ehr(ehr_data, date_str, hashed_csn) - logger.info(airflow) + logger.info(ehr_data) # delete csn once we no longer need it del original_csn diff --git a/src/sql/flow_sheet_values.sql b/src/sql/flow_sheet_values.sql index 3be8624..b66b626 100644 --- a/src/sql/flow_sheet_values.sql +++ b/src/sql/flow_sheet_values.sql @@ -28,7 +28,7 @@ LEFT JOIN star.visit_observation_type AS vt ON vo.visit_observation_type_id = vt.visit_observation_type_id WHERE vt.id_in_application IN ('6', '3040102622', '12946') -AND vo.valid_from BETWEEN %(yesterday)s AND %(today)s +AND vo.valid_from BETWEEN %(start_datetime)s AND %(end_datetime)s AND vo.hospital_visit_id = %(hospital_visit_id)s GROUP BY DateTimeRecorded, Units, vo.comment