diff --git a/.gitignore b/.gitignore index f810280..0e1ed95 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/config.EXAMPLE/controller.env.EXAMPLE b/config.EXAMPLE/controller.env.EXAMPLE index 43b3af5..8c4bc21 100644 --- a/config.EXAMPLE/controller.env.EXAMPLE +++ b/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 diff --git a/config.EXAMPLE/exporter.env.EXAMPLE b/config.EXAMPLE/exporter.env.EXAMPLE index cd963ac..f48233f 100644 --- a/config.EXAMPLE/exporter.env.EXAMPLE +++ b/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 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/lab_results.sql b/sql_scripts/lab_results.sql new file mode 100644 index 0000000..21b8b5d --- /dev/null +++ b/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 diff --git a/sql_scripts/lab_test_names.sql b/sql_scripts/lab_test_names.sql new file mode 100644 index 0000000..1838b72 --- /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') diff --git a/src/controller.py b/src/controller.py index c5c84fc..32fd79f 100644 --- a/src/controller.py +++ b/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) diff --git a/src/csv_writer.py b/src/csv_writer.py index a762273..aa780d3 100644 --- a/src/csv_writer.py +++ b/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 diff --git a/src/db.py b/src/db.py index ebab871..abaf3dd 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 @@ -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 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..eefc830 --- /dev/null +++ b/src/electronic_health_records/ehr.py @@ -0,0 +1,78 @@ +import logging + +from datetime import datetime, timedelta +import pandas as pd + +from db import caboodleDB, starDB +from csv_writer import write_ehr +from pseudon.pseudon import pseudonymise_relevant_columns + + +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. + """ + + caboodle_connection = caboodleDB() + caboodle_connection.connect() + + 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, + caboodle_connection: caboodleDB, + star_connection: starDB, +) -> None: + # will pick up the logger config defined in the snakemake job (ie. log to file) + logger = logging.getLogger(__name__) + + logger.info("Looking for airway data for %s.", hashed_csn) + + start_datetime = datetime.strptime(date_str, "%Y-%m-%d") + end_datetime = start_datetime + timedelta(days=1) + airflow = caboodle_connection.get_airflow( + start_datetime, end_datetime, original_csn + ) + + hospital_visit_id = star_connection.get_hospital_visit_from_csn(original_csn) + + 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", + ] + + ehr_data = pseudonymise_relevant_columns(ehr_data, safe_columns) + + write_ehr(ehr_data, date_str, hashed_csn) + + logger.info(ehr_data) + + # 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 97965a4..6cf0de6 100644 --- a/src/pipeline/Snakefile +++ b/src/pipeline/Snakefile @@ -9,14 +9,17 @@ 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, 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, report_ftp_upload @@ -47,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 @@ -64,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: @@ -74,6 +79,16 @@ rule all_daily_hash_lookups: input: ALL_DAILY_HASH_LOOKUPS +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 @@ -115,6 +130,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 7560b82..7615bc5 100644 --- a/src/pipeline/utils.py +++ b/src/pipeline/utils.py @@ -9,10 +9,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, ) @@ -74,6 +76,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 8200a5e..9707a76 100644 --- a/src/pseudon/pseudon.py +++ b/src/pseudon/pseudon.py @@ -165,7 +165,17 @@ def csv_to_parquets( "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", + "numeric_values", + "string_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! @@ -213,18 +223,7 @@ def add_waveform_metadata_to_table( return existing_table -SAFE_COLUMNS = [ - "sampling_rate", - "source_variable_id", - "source_channel_id", - "timestamp", - "units", - "numeric_values", - "string_values", -] - - -def pseudonymise_relevant_columns(df: pd.DataFrame): +def pseudonymise_relevant_columns(df: pd.DataFrame, safe_columns: list[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 @@ -234,6 +233,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 diff --git a/src/settings.py b/src/settings.py index 75f6908..08a7bfd 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") @@ -37,9 +38,19 @@ 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) +get_from_env("SQL_PATH", default_value="./src/sql/") # OpenTelemetry: OTLP/HTTP base URL of the LGTM collector, e.g. http://lgtm:4318 get_from_env("OTEL_EXPORTER_OTLP_ENDPOINT") diff --git a/src/sql/airway.sql b/src/sql/airway.sql new file mode 100644 index 0000000..c79c9cb --- /dev/null +++ b/src/sql/airway.sql @@ -0,0 +1,20 @@ +-- extract airway data from caboodle for a specific csn and date +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 %(start_datetime)s AND %(end_datetime)s +AND enc.EncounterEpicCsn = %(csn) diff --git a/src/sql/flow_sheet_values.sql b/src/sql/flow_sheet_values.sql new file mode 100644 index 0000000..b66b626 --- /dev/null +++ b/src/sql/flow_sheet_values.sql @@ -0,0 +1,34 @@ +--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 + +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 %(start_datetime)s AND %(end_datetime)s +AND vo.hospital_visit_id = %(hospital_visit_id)s + +GROUP BY DateTimeRecorded, Units, vo.comment diff --git a/src/sql/get_hospital_visit_id.sql b/src/sql/get_hospital_visit_id.sql new file mode 100644 index 0000000..8cd2f79 --- /dev/null +++ b/src/sql/get_hospital_visit_id.sql @@ -0,0 +1,4 @@ +-- Retrieve the hospital_visit_id associated with the csn value applied to this function -- + +select hospital_visit_id from {schema_name}.hospital_visit as hv +where hv.encounter = %(csn)s -- note the CSN must be in quotes 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, diff --git a/src/sql/sputum_secretions.sql b/src/sql/sputum_secretions.sql new file mode 100644 index 0000000..8f27ab8 --- /dev/null +++ b/src/sql/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) diff --git a/tests/helpers.py b/tests/helpers.py index 0ceaff0..18b720a 100644 --- a/tests/helpers.py +++ b/tests/helpers.py @@ -69,6 +69,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_controller.py b/tests/test_controller.py index 092b356..9845288 100644 --- a/tests/test_controller.py +++ b/tests/test_controller.py @@ -171,9 +171,11 @@ def test_controller_callback( 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() @@ -207,12 +209,12 @@ def test_controller_callback( was_bad_data = bad_data_type or lf_value_type == "both" if not was_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 was_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: diff --git a/tests/test_snakemake_integration.py b/tests/test_snakemake_integration.py index 6fbc866..d28eacb 100644 --- a/tests/test_snakemake_integration.py +++ b/tests/test_snakemake_integration.py @@ -283,9 +283,11 @@ def test_snakemake_pipeline(tmp_path: Path, background_hasher, monkeypatch): 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) @@ -367,12 +369,15 @@ 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_and_hash_lookups\n" "SNAKEMAKE_CORES=1\n" "INSTANCE_NAME=pytest\n" "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" ) # Collect coverage from Python processes inside the exporter container