diff --git a/.dockerignore b/.dockerignore index 2a0c6b6..e740dab 100644 --- a/.dockerignore +++ b/.dockerignore @@ -4,3 +4,4 @@ .git **/.mypy_cache **/.pytest_tmp +**/build diff --git a/.github/workflows/pytest.yml b/.github/workflows/pytest.yml index 670b270..be4479b 100644 --- a/.github/workflows/pytest.yml +++ b/.github/workflows/pytest.yml @@ -14,6 +14,13 @@ jobs: PROJECT_ROOT: ${{ github.workspace }}/waveform-controller environment: hasher steps: + - name: Generate GitHub App Token + id: generate-token + uses: actions/create-github-app-token@v1 + with: + app-id: ${{ secrets.SAFEHR_READ_ONLY_WORKER_ID }} + private-key: ${{ secrets.SAFEHR_READ_ONLY_WORKER_KEY }} + repositories: waveform-private-queries - uses: actions/checkout@v5 with: path: waveform-controller @@ -23,6 +30,13 @@ jobs: repository: SAFEHR-data/PIXL ref: e29f4b15c3b9d21c9a6e08c272aca7773311b32c # pragma: allowlist secret path: PIXL + - name: Checkout private SQL scripts + uses: actions/checkout@v5 + with: + repository: SAFEHR-data/waveform-private-queries + token: ${{ steps.generate-token.outputs.token }} + ref: 0345bb6e3ed51b05e46982434911c9138c84a035 # pragma: allowlist secret + path: waveform-private-queries - name: Install uv uses: astral-sh/setup-uv@v7 @@ -65,6 +79,8 @@ jobs: echo "AZURE_KEY_VAULT_SECRET_NAME=${AZURE_KEY_VAULT_SECRET_NAME}" } >> ../config/hasher.env + # install private SQL scripts + cp ../waveform-private-queries/src/sql/*.sql src/sql/private/ # exporter config can't be done here because test_snakemake_integration.py wires in its own config file - name: Run the tests 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/Dockerfile b/Dockerfile index dab631a..a3d80df 100644 --- a/Dockerfile +++ b/Dockerfile @@ -4,7 +4,11 @@ LABEL authors="Stephen Thompson, Jeremy Stein" # put it on both images even though we only need it on exporter. RUN export DEBIAN_FRONTEND=noninteractive && \ apt-get update && \ - apt-get install --yes --no-install-recommends cron && \ + apt-get install --yes --no-install-recommends \ + cron \ + libgssapi-krb5-2 \ + libkrb5-3 \ + libltdl7 && \ apt-get autoremove --yes && apt-get clean --yes && rm -rf /var/lib/apt/lists/* # uv image label "0.12.5" COPY --from=ghcr.io/astral-sh/uv@sha256:e85be844203885286c60ffad8a858d48afb6c5a5c237ca0e67f12e74b8f174b1 /uv /uvx /bin/ diff --git a/README.md b/README.md index 833acb7..26badb4 100644 --- a/README.md +++ b/README.md @@ -64,10 +64,14 @@ separate to the Emap project root. ##### Clone repos -Clone this repo (`waveform-controller`) and [PIXL](https://github.com/SAFEHR-data/PIXL), +Clone +* this repo (`waveform-controller`) +* [PIXL](https://github.com/SAFEHR-data/PIXL), +* [Private query repo](https://github.com/SAFEHR-data/waveform-private-queries) both inside your root directory. -Inside the PIXL repo, checkout the commit that we have pinned in [workflow file](.github/workflows/pytest.yml). +Inside the PIXL and Private query repos, checkout the respective commits that +we have pinned in [the workflow file](.github/workflows/pytest.yml). If on a system that has access to sensitive data, disable push remotes on all cloned repos as follows: ``` diff --git a/config.EXAMPLE/controller.env.EXAMPLE b/config.EXAMPLE/controller.env.EXAMPLE index 43b3af5..123a3b0 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" + # 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..8a1404d 100644 --- a/config.EXAMPLE/exporter.env.EXAMPLE +++ b/config.EXAMPLE/exporter.env.EXAMPLE @@ -33,6 +33,31 @@ 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="1433" +CABOODLE_CONNECT_TIMEOUT="10" # in seconds +CABOODLE_QUERY_TIMEOUT="10" # in seconds + +# 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" + # OpenTelemetry OTLP/HTTP endpoint of the LGTM collector. OTEL_EXPORTER_OTLP_ENDPOINT="http://lgtm:4318" OTEL_SERVICE_NAME=waveform-exporter diff --git a/docs/deployment.md b/docs/deployment.md index 2a7f8ba..8773460 100644 --- a/docs/deployment.md +++ b/docs/deployment.md @@ -35,6 +35,7 @@ sledgehammer approach which is rather similar to * Delete all Emap tables in `star_dev` as per Emap deployment instructions. * Waveform: `docker compose down` to bring everything down * git pull and rebuild containers for the two repos. +* **REMEMBER: that the files from waveform-private-queries repository need to be copied to the waveform-controller/src/sql/private directory.** * Change config if necessary * Bring it all up again @@ -183,15 +184,14 @@ Here is the Filtering by variable is not currently possible. -# Run de-id on ad adhoc basis +# Run the Snakemake workflow on an ad hoc basis (ie. de-id, EHR lookup, upload) > [!NOTE] > Due to the way scheduled-script.sh pulls in its config from the config file, the contents of -> that file will override any env vars you specify on the command line below. +> that file will override any env vars you specify via docker below. So, temporarily +> changing the exporter.env config file is the only way to pass in a certain config. -You need to temporarily change the exporter.env config file to run this command. - -You are likely to want to set the following values (example date shown): +Variables you may wish to modify: ``` ONLY_USE_CSV_FROM_YESTERDAY=FALSE # something shorter than the standard 180 may be needed if you only just processed the data @@ -202,4 +202,4 @@ PROCESS_CSV_FROM_DATE=1234-12-12 docker compose run --entrypoint /app/exporter-scripts/scheduled-script.sh waveform-exporter ``` -Remember to put the config back afterwards. +Remember to revert the config changes you made if applicable. diff --git a/pyproject.toml b/pyproject.toml index 6c15d11..cc17338 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -10,6 +10,7 @@ dependencies = [ "pika>=1.3.2", "pre-commit>=4.5.0", "snakemake==9.14.5", + "mssql-python>=1.13.0", "opentelemetry-distro==0.63b1", "opentelemetry-exporter-otlp-proto-http==1.42.1", # need to be compatible with PIXL, which currently pins 2.9.11 (arguably it shouldn't) @@ -36,6 +37,10 @@ coverage = [ [project.scripts] emap-extract-waveform = "controller:receiver" +# Include .sql files as part of the package (otherwise only editable installs would work) +[tool.setuptools.package-data] +sql = ["*.sql", "private/*.sql"] + [tool.pytest.ini_options] # Force temp dirs under the repo so Docker can mount them on macOS. # The default under /private/var/folders seems to silently fail (gives you an empty directory) diff --git a/src/controller.py b/src/controller.py index c5c84fc..8f9fd1a 100644 --- a/src/controller.py +++ b/src/controller.py @@ -12,7 +12,7 @@ from pika import spec from pika.adapters.blocking_connection import BlockingChannel -import db as db # type:ignore +import db_pg import settings as settings # type:ignore import csv_writer as writer # type:ignore import telemetry as telemetry # type:ignore @@ -107,8 +107,7 @@ def finalise_message(outcome: MessageOutcome): class WaveformController: def __init__(self): - self.emap_db = db.starDB() - self.emap_db.init_query() + self.emap_db = db_pg.starDB() 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) @@ -228,7 +230,6 @@ def outcome( if opt_out: logger.info("Research opt-out is set for mrn %s, not writing.", mrn) return outcome("reject", reason="opt_out", requeue=False) - writer.write_frame( source_variable_id=source_variable_id, source_channel_id=source_channel_id, diff --git a/src/csv_writer.py b/src/csv_writer.py index a762273..b904ca7 100644 --- a/src/csv_writer.py +++ b/src/csv_writer.py @@ -2,10 +2,15 @@ import csv import json + from datetime import datetime from typing import Optional -from locations import WAVEFORM_ORIGINAL_CSV, make_file_name, FILE_STEM_PATTERN +from locations import ( + WAVEFORM_ORIGINAL_CSV, + make_file_name, + FILE_STEM_PATTERN, +) def create_file_name( @@ -76,7 +81,6 @@ def write_frame( wv_writer = csv.writer( fileout, delimiter=",", quoting=csv.QUOTE_ALL, lineterminator="\n" ) - # Encode value lists as JSON so parquet conversion can use json.loads # (Python list repr breaks on commas / quotes in string values). row_array = [ diff --git a/src/db.py b/src/db.py deleted file mode 100644 index ebab871..0000000 --- a/src/db.py +++ /dev/null @@ -1,55 +0,0 @@ -from datetime import datetime -import psycopg2 -from psycopg2 import sql, pool -import logging - -import settings as settings # type:ignore - -logging.basicConfig(format="%(levelname)s:%(asctime)s: %(message)s") -logger = logging.getLogger(__name__) - - -class starDB: - sql_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 - settings.UDS_PASSWORD, # type:ignore - settings.UDS_HOST, # type:ignore - settings.UDS_PORT, # type:ignore - settings.UDS_CONNECT_TIMEOUT, # type:ignore - settings.UDS_QUERY_TIMEOUT, # type:ignore - ) - connection_pool: pool.ThreadedConnectionPool - - def connect(self): - self.connection_pool = pool.SimpleConnectionPool(1, 1, self.connection_string) - - 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( - schema_name=sql.Identifier(settings.SCHEMA_NAME) - ) - - def get_row(self, location_string: str, observation_datetime: datetime): - parameters = { - "location_string": location_string, - "observation_datetime": observation_datetime, - } - try: - with self.connection_pool.getconn() as db_connection: - with db_connection.cursor() as curs: - curs.execute(self.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}") - - 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] diff --git a/src/db_mssql.py b/src/db_mssql.py new file mode 100644 index 0000000..497bfa7 --- /dev/null +++ b/src/db_mssql.py @@ -0,0 +1,131 @@ +from datetime import datetime +from typing import Any + +import mssql_python +import pandas as pd + +import settings as settings # type:ignore +from db_utils import ( + get_sql_query_text, + utc_to_naive_local, + naive_local_to_utc, + validate_args_must_be_utc, +) + + +def _odbc_escape(value: object) -> str: + """Quote a value for use in an ODBC connection string.""" + return "{" + str(value).replace("}", "}}") + "}" + + +def _get_connection_string() -> str: + return "Server={server};Database={database};UID={username};PWD={password};".format( + server=_odbc_escape( + f"{settings.CABOODLE_HOST},{settings.CABOODLE_PORT}" # type:ignore + ), + database=_odbc_escape(settings.CABOODLE_DBNAME), # type:ignore + username=_odbc_escape(settings.CABOODLE_USERNAME), # type:ignore + password=_odbc_escape(settings.CABOODLE_PASSWORD), # type:ignore + ) + + +def tz_adjust(value: Any) -> Any: + """Only relevant to data returned from MS SQL Server, convert naive local times to + UTC.""" + if isinstance(value, datetime): + return naive_local_to_utc(value) + else: + # non-datetimes are returned unchanged + return value + + +class caboodleDB: + """For querying the caboodle database to extract electronic healthcare records per + patient. + + Caboodle stores its timestamps as timezone-naive SQL types + (`datetime` in SQL Server) that are in the local time of the hospital. + + Note: hospital time is not necessarily system time! + + The behaviour of the DB driver is also relevant here: + https://learn.microsoft.com/en-us/sql/connect/python/mssql-python/datetime-handling?view=sql-server-ver17 + + The waveform pipeline uses UTC wherever possible, so appropriate conversions to local time and back + again are the responsibility of these methods. + """ + + connection_string: str + db_connection: mssql_python.Connection + fake_caboodle: bool = False + + def connect(self) -> None: + """Set up connection to the database.""" + self.fake_caboodle = settings.CABOODLE_TESTING == "TRUE" + if not self.fake_caboodle: + self.connection_string = _get_connection_string() + self.db_connection = mssql_python.connect( + self.connection_string, + timeout=int(settings.CABOODLE_QUERY_TIMEOUT), + attrs_before={ + mssql_python.SQL_ATTR_LOGIN_TIMEOUT: int( + settings.CABOODLE_CONNECT_TIMEOUT # type:ignore + ) + }, + ) + + def get_airway( + self, utc_start_datetime: datetime, utc_end_datetime: datetime, csn: str + ) -> pd.DataFrame: + """Retrieve airflow data from database.""" + validate_args_must_be_utc(utc_start_datetime, utc_end_datetime) + + local_start_datetime = utc_to_naive_local(utc_start_datetime) + local_end_datetime = utc_to_naive_local(utc_end_datetime) + + airway_query = get_sql_query_text("private/airway.sql") + parameters = { + "start_datetime": local_start_datetime, + "end_datetime": local_end_datetime, + "csn": csn, + } + + if self.fake_caboodle: + fake_airway = { + "DateTimeRecorded": [0], + "PlacementInstant": [0], + "RemovalInstant": [0], + "TubeSize": [0], + } + return pd.DataFrame(data=fake_airway) + + rows, columns = self._get_rows(airway_query, parameters) + rows_adjusted = [tuple(tz_adjust(v) for v in r) for r in rows] + return pd.DataFrame(rows_adjusted, columns=columns) + + def get_sputum_secretions( + self, utc_start_datetime: datetime, utc_end_datetime: datetime, csn: str + ) -> pd.DataFrame: + validate_args_must_be_utc(utc_start_datetime, utc_end_datetime) + local_start_datetime = utc_to_naive_local(utc_start_datetime) + local_end_datetime = utc_to_naive_local(utc_end_datetime) + secr_query = get_sql_query_text("private/sputum_secretions.sql") + parameters = { + "start_datetime": local_start_datetime, + "end_datetime": local_end_datetime, + "csn": csn, + } + rows, columns = self._get_rows(secr_query, parameters) + rows_adjusted = [tuple(tz_adjust(v) for v in r) for r in rows] + return pd.DataFrame(rows_adjusted, columns=columns) + + def _get_rows(self, sql_query: str, parameters: dict) -> tuple[list, list[str]]: + try: + with self.db_connection.cursor() as curs: + curs.execute(sql_query, parameters) + rows = curs.fetchall() + col_names = [col[0] for col in curs.description] + except mssql_python.OperationalError as e: + raise ConnectionError(f"Database error: {e}") from e + + return rows, col_names diff --git a/src/db_pg.py b/src/db_pg.py new file mode 100644 index 0000000..11cbbfe --- /dev/null +++ b/src/db_pg.py @@ -0,0 +1,162 @@ +from datetime import datetime +from typing import Optional + +import pandas as pd +import psycopg2 +from psycopg2 import sql, pool +import logging + +import settings as settings # type:ignore +from db_utils import get_sql_query_text, validate_args_must_be_utc + +logging.basicConfig(format="%(levelname)s:%(asctime)s: %(message)s") +logger = logging.getLogger(__name__) + + +def get_sql_query_with_schema( + query_rel_path: str, schema_name: Optional[str] = None +) -> sql.Composable: + query_text_tmpl = sql.SQL(get_sql_query_text(query_rel_path)) + if schema_name is None: + return query_text_tmpl + else: + return query_text_tmpl.format(schema_name=sql.Identifier(schema_name)) + + +class starDB: + 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 + settings.UDS_PASSWORD, # type:ignore + settings.UDS_HOST, # type:ignore + settings.UDS_PORT, # type:ignore + settings.UDS_CONNECT_TIMEOUT, # type:ignore + 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 + ) + + def get_matched_mrn( + self, location_string: str, observation_datetime: datetime + ) -> tuple: + validate_args_must_be_utc(observation_datetime) + + parameters = { + "location_string": location_string, + "observation_datetime": observation_datetime, + } + mrn_lookup_query = get_sql_query_with_schema( + "mrn_based_on_bed_and_datetime.sql", settings.SCHEMA_NAME + ) + + rows, col_names = self._get_rows(mrn_lookup_query, parameters) + + num_rows = len(rows) + if num_rows != 1: + raise ValueError( + f"Wrong number of rows returned from database. {num_rows} != 1, for {location_string}:{observation_datetime}" + ) + + return rows[0] + + def get_hospital_visit_from_csn(self, csn: str) -> int: + hv_query = get_sql_query_with_schema( + "get_hospital_visit_id.sql", settings.SCHEMA_NAME + ) + + parameters = { + "csn": csn, + } + if self.fake_star: + return 12345678 + + hospital_visit_rows, col_names = self._get_rows(hv_query, parameters) + return int(hospital_visit_rows[0][0]) + + def get_flowsheets( + self, + utc_start_datetime: datetime, + utc_end_datetime: datetime, + hospital_visit_id: int, + ) -> pd.DataFrame: + """Retrieve airflow data from database.""" + validate_args_must_be_utc(utc_start_datetime, utc_end_datetime) + + flowsheet_query = get_sql_query_with_schema( + "flow_sheet_values.sql", settings.SCHEMA_NAME + ) + + parameters = { + "start_datetime": utc_start_datetime, + "end_datetime": utc_end_datetime, + "hospital_visit_id": hospital_visit_id, + } + + if self.fake_star: + fake_flowsheet = { + "DateTimeRecorded": [0], + "Temperature": [0], + "Noradrenaline": [0], + "Metaraminol": [0], + } + return pd.DataFrame(data=fake_flowsheet) + + rows, col_names = self._get_rows(flowsheet_query, parameters) + return pd.DataFrame(rows, columns=col_names) + + def get_lab_results( + self, + utc_start_datetime: datetime, + utc_end_datetime: datetime, + hospital_visit_id: int, + ) -> pd.DataFrame: + """Retrieve lab result data from caboodle.""" + validate_args_must_be_utc(utc_start_datetime, utc_end_datetime) + + lab_result_query = get_sql_query_with_schema( + "lab_results.sql", settings.SCHEMA_NAME + ) + parameters = { + "start_datetime": utc_start_datetime, + "end_datetime": utc_end_datetime, + "hospital_visit_id": hospital_visit_id, + } + + if self.fake_star: + fake_lab_result = { + "DateTimeRecorded": [0], + "Units": ["None"], + "Abnormal_result": ["No"], + "Comments": ["None"], + "C-reactive protein 1": ["-"], + "CSF WCC TUBE 1": ["-"], + "CSF WCC TUBE 2": ["-"], + "CSF WCC TUBE 3": ["-"], + "C-reactive protein 2": ["-"], + } + return pd.DataFrame(data=fake_lab_result) + + rows, col_names = self._get_rows(lab_result_query, parameters) + return pd.DataFrame(rows, columns=col_names) + + def _get_rows( + self, sql_query: sql.Composable, parameters: dict + ) -> tuple[list, list[str]]: + try: + with self.connection_pool.getconn() as db_connection: + with db_connection.cursor() as curs: + curs.execute(sql_query, parameters) + rows = curs.fetchall() + col_names = [col.name for col in curs.description] + 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, col_names diff --git a/src/db_utils.py b/src/db_utils.py new file mode 100644 index 0000000..71e229c --- /dev/null +++ b/src/db_utils.py @@ -0,0 +1,29 @@ +from datetime import datetime, timezone +from importlib import resources +from zoneinfo import ZoneInfo + + +def get_sql_query_text(query_rel_path: str) -> str: + return (resources.files("sql") / query_rel_path).read_text() + + +# Timezone that the hospital sits in (and thus Caboodle records data using) +HOSPITAL_TZ = ZoneInfo("Europe/London") + + +def utc_to_naive_local(utc_dt: datetime) -> datetime: + return utc_dt.astimezone(HOSPITAL_TZ).replace(tzinfo=None) + + +def naive_local_to_utc(naive_dt: datetime) -> datetime: + if naive_dt.tzinfo is not None: + raise ValueError(f"datetime {naive_dt} is not a tz-naive datetime") + return naive_dt.replace(tzinfo=HOSPITAL_TZ).astimezone(timezone.utc) + + +def validate_args_must_be_utc(*args): + for a in args: + if not isinstance(a, datetime): + raise TypeError(f"Argument {a} must be a datetime") + if a.tzinfo is None or a.tzinfo != timezone.utc: + raise TypeError(f"Argument {a} must be have a UTC timezone") 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..ae0906d --- /dev/null +++ b/src/electronic_health_records/ehr.py @@ -0,0 +1,123 @@ +import logging + +from datetime import datetime, timedelta, timezone +import pandas as pd + +from db_mssql import caboodleDB +from db_pg import starDB +from locations import ( + make_file_name, + WAVEFORM_PSEUDONYMISED_EHR, + EHR_STEM_PATTERN_HASHED, +) +from pseudon.pseudon import pseudonymise_relevant_columns, write_ehr_parquet + + +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) + + # When waveform data is grouped into days, it's always in UTC, so calculate the day + # boundaries as UTC. + utc_start_datetime = datetime.strptime(date_str, "%Y-%m-%d").replace( + tzinfo=timezone.utc + ) + utc_end_datetime = utc_start_datetime + timedelta(days=1) + + # we need hospital visit id for flowsheet and lab_result queries + hospital_visit_id = star_connection.get_hospital_visit_from_csn(original_csn) + + # fetch data from caboodle + airways = caboodle_connection.get_airway( + utc_start_datetime, utc_end_datetime, original_csn + ) + + secretions = caboodle_connection.get_sputum_secretions( + utc_start_datetime, utc_end_datetime, original_csn + ) + + # delete csn once we no longer need it + del original_csn + + # fetch data from Emap + flowsheet_values = star_connection.get_flowsheets( + utc_start_datetime, utc_end_datetime, hospital_visit_id + ) + + lab_results = star_connection.get_lab_results( + utc_start_datetime, utc_end_datetime, hospital_visit_id + ) + + ehr_data = pd.concat([airways, secretions, flowsheet_values, lab_results]) + + # we can pseudonymise to safe, although at the moment all columns + # are considered safe + safe_columns = [ + "TubeEventId", + "TubeDateTimeRecorded", + "TubePlacementInstant", + "TubeRemovalInstant", + "TubeSize", + "SecrDateTimeRecorded", + "SecrSecretions", + "SecrSputum", + # Free text comments could in principle contain sensitive information but + # we have assessed this particular column to be low risk + "SecrComments", + "Repositioned", + "Position frequency", + "FlowsheetDateTimeRecorded", + "FlowsheetTemperature", + "FlowsheetNoradrenaline", + "FlowsheetMetaraminol", + "FlowsheetPaO2", + "FlowsheetPaCO2", + "FlowsheetUnits", + "LabDateTimeRecorded", + "LabUnits", + "LabCRP", + "LabWCC", + ] + + ehr_data = pseudonymise_relevant_columns(ehr_data, safe_columns) + print(ehr_data.columns) + print(ehr_data) + t = ehr_data["FlowsheetTemperature"].iloc[0] + print(f"({type(t)}) {t}") + + 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) + + write_ehr_parquet(ehr_data, filename) diff --git a/src/exporter/ftps.py b/src/exporter/ftps.py index 54ac620..71bc946 100644 --- a/src/exporter/ftps.py +++ b/src/exporter/ftps.py @@ -6,7 +6,11 @@ from time import perf_counter from typing import Any -from core.uploader._ftps import _connect_to_ftp, _create_and_set_as_cwd_multi_path +from core.uploader._ftps import ( + _connect_to_ftp, + _create_and_set_as_cwd_multi_path, + ImplicitFtpTls, +) import settings import telemetry @@ -103,6 +107,7 @@ def do_upload_multiple( settings.FTPS_PORT, settings.FTPS_USERNAME, settings.FTPS_PASSWORD, + ImplicitFtpTls, ) _create_and_set_as_cwd_multi_path(ftp, remote_project_dir) command = f"STOR {remote_tar_filename}" diff --git a/src/locations.py b/src/locations.py index a3c867f..53c4e94 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 3876b86..97d6b40 100644 --- a/src/pipeline/Snakefile +++ b/src/pipeline/Snakefile @@ -8,16 +8,19 @@ from locations import ( WAVEFORM_ORIGINAL_CSV, WAVEFORM_SNAKEMAKE_LOGS, WAVEFORM_PSEUDONYMISED_PARQUET, + WAVEFORM_PSEUDONYMISED_EHR, HASH_LOOKUP_JSON, HASH_LOOKUP_JSON_REL, FILE_STEM_PATTERN, FILE_STEM_PATTERN_HASHED, + EHR_STEM_PATTERN_HASHED, make_file_name, ALL_UPLOADED_JSON, ALL_FTPS_LOG, ) from pipeline.utils import config_bool, determine_eventual_outputs, timestamp_for_paths from pseudon.pseudon import csv_to_parquets +from electronic_health_records.ehr import ehr_for_csv # How long before we assume that no more data will be written to the file, and @@ -46,6 +49,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 = sorted({ao.get_ftps_uploaded_all_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 @@ -63,7 +67,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: @@ -73,6 +78,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 @@ -114,6 +129,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 c16f0f1..b1175e9 100644 --- a/src/pipeline/utils.py +++ b/src/pipeline/utils.py @@ -9,9 +9,11 @@ from pseudon.hashing import do_hash from locations import ( WAVEFORM_PSEUDONYMISED_PARQUET, + WAVEFORM_PSEUDONYMISED_EHR, HASH_LOOKUP_JSON, ORIGINAL_PARQUET_PATTERN, FILE_STEM_PATTERN_HASHED, + EHR_STEM_PATTERN_HASHED, CSV_PATTERN, make_file_name, ALL_UPLOADED_JSON, @@ -73,6 +75,10 @@ def get_ftps_uploaded_all_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 27d653c..af1f626 100644 --- a/src/pseudon/pseudon.py +++ b/src/pseudon/pseudon.py @@ -151,13 +151,22 @@ def csv_to_parquets( use_dictionary=True, write_statistics=True, write_page_index=True, - flavor="spark", ) logger.info( "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! @@ -179,13 +188,54 @@ def csv_to_parquets( use_dictionary=True, write_statistics=True, write_page_index=True, - flavor="spark", ) logger.info( "Done turning CSV %s to pseudonymised parquet %s", csv_path, hashed_path ) +def write_ehr_parquet(df: pd.DataFrame, ehr_parquet_path: Path): + schema = pa.schema( + [ + # for numeric types we match what's in the database, for better or worse + ("SecrDateTimeRecorded", pa.timestamp("us", tz="UTC")), + ("SecrSecretions", pa.string()), + ("SecrSputum", pa.string()), + ("SecrComments", pa.string()), + ("TubeEventId", pa.int64()), + ("TubeDateTimeRecorded", pa.timestamp("us", tz="UTC")), + ("TubePlacementInstant", pa.timestamp("us", tz="UTC")), + ("TubeRemovalInstant", pa.timestamp("us", tz="UTC")), + ("TubeSize", pa.string()), + # ("Repositioned", ), + # ("Position frequency", ), + ("FlowsheetDateTimeRecorded", pa.timestamp("us", tz="UTC")), + ("FlowsheetTemperature", pa.float64()), + ("FlowsheetNoradrenaline", pa.float64()), + ("FlowsheetMetaraminol", pa.float64()), + ("FlowsheetPaO2", pa.float64()), + ("FlowsheetPaCO2", pa.float64()), + ("FlowsheetUnits", pa.string()), + ("LabDateTimeRecorded", pa.timestamp("us", tz="UTC")), + ("LabUnits", pa.string()), + ("LabCRP", pa.float64()), + ("LabWCC", pa.float64()), + ] + ) + ehr_table = pa.Table.from_pandas(df, schema=schema, preserve_index=True) + pq.write_table( + ehr_table, + str(ehr_parquet_path), + # valid values: {‘NONE’, ‘SNAPPY’, ‘GZIP’, ‘BROTLI’, ‘LZ4’, ‘ZSTD’} + compression="zstd", + use_dictionary=True, + write_statistics=True, + write_page_index=True, + # we do not use flavor="spark" here because that would + # use legacy INT96 timestamps which are timezone-naive. + ) + + def add_waveform_metadata_to_table( existing_table: pa.Table, metadata: dict[str, Any] ) -> pa.Table: @@ -205,18 +255,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 @@ -226,6 +265,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..b93b367 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,6 +38,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) diff --git a/src/sql/README.md b/src/sql/README.md new file mode 100644 index 0000000..799fab3 --- /dev/null +++ b/src/sql/README.md @@ -0,0 +1,45 @@ +# Notes on putting together the EHR needed + +## Private scripts + +This is a public repository and so we cannot include any scripts that are proprietary from the hospital system EPIC. +These are included in a separate private repository named waveform-private-queries. This has a directory structure + +[top-level]/src/sql/private + +so that it can be copied directly onto the directory structure of this repository and thus all scripts will be contained in the same place upon deployment. + + + +## 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/src/sql| star | +| flow_sheet_values.sql| hospital_visit_id/today/yesterday | part of table above | waveform-controller/src/sql| star | +| lab_results.sql | csn/today/yesterday | part of the table above | waveform-controller/src/sql | star | +| sputum_secretions.sql | csn/today/yesterday | part of the table above | waveform-private-queries/src/sql | caboodle | +| reposition.sql | csn/today/yesterday | part of the table above | waveform-private-queries/src/sql | caboodle | +--- + +## Unfinished scripts + +| script | arguments | record | location of script in repo | database | +|- | --- | --- |- | --- | +| airway.sql | csn/today/yesterday | part of the table above | waveform-private-queries/src/sql | caboodle | +--- diff --git a/src/sql/__init__.py b/src/sql/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/sql/flow_sheet_values.sql b/src/sql/flow_sheet_values.sql new file mode 100644 index 0000000..0adac9b --- /dev/null +++ b/src/sql/flow_sheet_values.sql @@ -0,0 +1,46 @@ +-- 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 +-- Noradrenaline 3040102622 +-- Metaraminol 12946 +-- PaO2 40191 +-- PaCO2 39947 + + +SELECT + vo.observation_datetime AS "FlowsheetDateTimeRecorded", + + MAX(vo.value_as_real) FILTER ( + WHERE vt.id_in_application = '6' + ) AS "FlowsheetTemperature", + + MAX(vo.value_as_real) FILTER ( + WHERE vt.id_in_application = '3040102622' + ) AS "FlowsheetNoradrenaline", + + MAX(vo.value_as_real) FILTER ( + WHERE vt.id_in_application = '12946' + ) AS "FlowsheetMetaraminol", + + MAX(vo.value_as_real) FILTER ( + WHERE vt.id_in_application = '40191' + ) AS "FlowsheetPaO2", + + MAX(vo.value_as_real) FILTER ( + WHERE vt.id_in_application = '39947' + ) AS "FlowsheetPaCO2", + + vo.unit AS "FlowsheetUnits" + +FROM {schema_name}.visit_observation AS vo + +LEFT JOIN {schema_name}.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', '40191', '39947') + AND vo.valid_from >= %(start_datetime)s AND vo.valid_from < %(end_datetime)s + AND vo.hospital_visit_id = %(hospital_visit_id)s + +GROUP BY "FlowsheetDateTimeRecorded", "FlowsheetUnits" 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/lab_results.sql b/src/sql/lab_results.sql new file mode 100644 index 0000000..6d5efe0 --- /dev/null +++ b/src/sql/lab_results.sql @@ -0,0 +1,27 @@ +-- This selects the values of lab tests +-- 1011 C REACTIVE PROTEIN +-- 686 WHITE CELL COUNT + +SELECT + r.result_last_modified_datetime AS "LabDateTimeRecorded", + + MAX(r.value_as_real) FILTER + (WHERE r.lab_test_definition_id = '1001') AS "LabCRP", + + MAX(r.value_as_real) FILTER + (WHERE r.lab_test_definition_id = '686') AS "LabWCC", + + r.units AS "LabUnits" + +FROM {schema_name}.lab_result AS r +LEFT JOIN {schema_name}.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', '686') + AND r.result_last_modified_datetime >= %(start_datetime)s + AND r.result_last_modified_datetime < %(end_datetime)s + AND o.hospital_visit_id = %(hospital_visit_id)s + +GROUP BY "DateTimeRecorded", "Units" 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/private/.gitignore b/src/sql/private/.gitignore new file mode 100644 index 0000000..083a98e --- /dev/null +++ b/src/sql/private/.gitignore @@ -0,0 +1,2 @@ +# for queries that come from another source that are not to be included in this repo +* 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..bcf6eb6 100644 --- a/tests/test_controller.py +++ b/tests/test_controller.py @@ -171,10 +171,12 @@ 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) - monkeypatch.setattr("controller.db.starDB", Mock(return_value=emap_db_mock)) + emap_db_mock.get_matched_mrn.return_value = ("mrn", "nhsno", "csn", opt_out) + monkeypatch.setattr("controller.db_pg.starDB", Mock(return_value=emap_db_mock)) write_frame_mock = Mock() monkeypatch.setattr("controller.writer.write_frame", write_frame_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_ehr.py b/tests/test_ehr.py new file mode 100644 index 0000000..427f3f9 --- /dev/null +++ b/tests/test_ehr.py @@ -0,0 +1,233 @@ +from unittest.mock import Mock + +import pandas as pd +import pytest + +from db_mssql import caboodleDB +from db_pg import starDB, get_sql_query_with_schema +from datetime import datetime, timedelta, timezone + +import pyarrow.parquet as pq +from db_utils import get_sql_query_text +from electronic_health_records.ehr import ehr_for_csv + +import settings + + +@pytest.fixture(scope="function", autouse=True) +def patch_mock_get_rows(monkeypatch): + """Replace the DB fetchers with simplified versions that just look at the query (but + not the params) and returns some data with the right types/shape (as it comes from + cursor.fetchall(), so it's already converted to Python types). + + This allows testing of some of the TZ conversion and EHR output writing files but + doesn't test the DB behaviour (esp re TZ) + """ + + def mock_get_rows_mssql(self, query, params): + if query == get_sql_query_text("private/airway.sql"): + col_names = [ + "TubeEventId", + "TubeDateTimeRecorded", + "TubePlacementInstant", + "TubeRemovalInstant", + "TubeSize", + ] + # this is based on the assumed data, not a real query + # Assumptions: + # * all return timezone-naive Python datetimes because that's what it is in the DB + # * placement and removal usually appear in different rows + rows = [ + ( + 10, + datetime(2026, 9, 14, 3, 30), + datetime(2026, 9, 14, 3, 20), + # use a mix of summer and winter dates + datetime(2026, 11, 14, 5, 10), + "7 mm", + ), + ( + 20, + datetime(2026, 9, 14, 2, 31), + datetime(2026, 9, 14, 3, 21), + None, + "7.5 mm", + ), + ] + return rows, col_names + elif query == get_sql_query_text("private/sputum_secretions.sql"): + col_names = [ + "SecrDateTimeRecorded", + "SecrSecretions", + "SecrSputum", + "SecrComments", + ] + rows = [ + ( + datetime(2026, 9, 14, 3, 30), + "Small", + None, + "", + ), + ( + datetime(2026, 9, 14, 6, 30), + None, + "None", # "None" as in no Sputum! + "", + ), + ] + return rows, col_names + else: + raise ValueError(f"Caboodle query not recognised: {query}") + + def mock_get_rows_pg(self, query, params): + if query == get_sql_query_with_schema("lab_results.sql", settings.SCHEMA_NAME): + # this is based on real queries + rows = [ + ( + datetime( + 2026, 9, 14, 6, 30, tzinfo=timezone(timedelta(seconds=3600)) + ), + 30.1, + None, + "mg/L", + ), + ( + datetime( + 2026, 9, 14, 6, 39, tzinfo=timezone(timedelta(seconds=3600)) + ), + None, + 30.21, + "x10^9/L", + ), + ] + col_names = ["LabDateTimeRecorded", "LabCRP", "LabWCC", "LabUnits"] + return rows, col_names + elif query == get_sql_query_with_schema( + "flow_sheet_values.sql", settings.SCHEMA_NAME + ): + col_names = [ + "FlowsheetDateTimeRecorded", + "FlowsheetTemperature", + "FlowsheetNoradrenaline", + "FlowsheetMetaraminol", + "FlowsheetPaO2", + "FlowsheetPaCO2", + "FlowsheetUnits", + ] + # example flowsheets, based on real queries + rows = [ + # Noradrenaline (shouldn't there be a concentration or a time component to the unit?) + ( + datetime( + 2026, 9, 14, 1, 2, tzinfo=timezone(timedelta(seconds=3600)) + ), + None, + 1.0, + None, + None, + None, + "mL", + ), + # Temperature (why no units?) + ( + datetime( + 2026, 9, 14, 2, 5, tzinfo=timezone(timedelta(seconds=3600)) + ), + 97.5, + None, + None, + None, + None, + None, + ), + ] + return rows, col_names + elif query == get_sql_query_with_schema( + "get_hospital_visit_id.sql", settings.SCHEMA_NAME + ): + rows = [ + (123456,), + ] + col_names = ["hospital_visit_id"] + return rows, col_names + else: + raise ValueError(f"Star query not recognised: {query}") + + monkeypatch.setattr(caboodleDB, "connect", Mock()) + monkeypatch.setattr(caboodleDB, "_get_rows", mock_get_rows_mssql) + monkeypatch.setattr(starDB, "connect", Mock()) + monkeypatch.setattr(starDB, "_get_rows", mock_get_rows_pg) + + +def test_ehr(monkeypatch, tmp_path): + fake_abs_root = tmp_path.absolute() + fake_waveform_pseudonymised_ehr = fake_abs_root / "pseudonymised_ehr" + monkeypatch.setattr( + "electronic_health_records.ehr.WAVEFORM_PSEUDONYMISED_EHR", + fake_waveform_pseudonymised_ehr, + ) + + ehr_for_csv(date_str="2026-09-14", original_csn="SECRET1234", hashed_csn="fakehash") + + # just check the file contains something for now (it will be changing to parquet) + expected_file = ( + fake_waveform_pseudonymised_ehr / "2026-09-14" / "2026-09-14.fakehash_ehr.csv" + ) + assert expected_file.exists() + + ehr_data = pq.read_table(expected_file) + assert ehr_data.num_rows == 8 + df = ehr_data.to_pandas( + # in particular, stop ints from being loaded as floats if there are null values in the column + types_mapper=pd.ArrowDtype + ) + + def non_null_vals(df, flowsheet_col) -> pd.DataFrame: + return df[df[flowsheet_col].notna()][ + ["FlowsheetDateTimeRecorded", flowsheet_col, "FlowsheetUnits"] + ] + + # always check against UTC + non_null_temps = non_null_vals(df, "FlowsheetTemperature") + assert non_null_temps.shape[0] == 1 + row0 = non_null_temps.iloc[0] + assert row0[0] == datetime(2026, 9, 14, 1, 5, tzinfo=timezone.utc) + assert row0[1] == 97.5 + assert pd.isna(row0[2]) + + non_null_norad = non_null_vals(df, "FlowsheetNoradrenaline") + assert non_null_norad.shape[0] == 1 + assert tuple(non_null_norad.iloc[0]) == ( + datetime(2026, 9, 14, 0, 2, tzinfo=timezone.utc), + 1.0, + "mL", + ) + + non_null_pa02 = non_null_vals(df, "FlowsheetPaO2") + assert non_null_pa02.shape[0] == 0 + + tube_events = df[df["TubeEventId"].notna()][ + [ + "TubeEventId", + "TubeDateTimeRecorded", + "TubePlacementInstant", + "TubeRemovalInstant", + "TubeSize", + ] + ] + assert tube_events.shape[0] == 2 + assert tuple(tube_events.iloc[0]) == ( + 10, + datetime(2026, 9, 14, 2, 30, tzinfo=timezone.utc), + datetime(2026, 9, 14, 2, 20, tzinfo=timezone.utc), + datetime(2026, 11, 14, 5, 10, tzinfo=timezone.utc), + "7 mm", + ) + assert tuple(tube_events.iloc[1].iloc[[0, 1, 2, 4]]) == ( + 20, + datetime(2026, 9, 14, 1, 31, tzinfo=timezone.utc), + datetime(2026, 9, 14, 2, 21, tzinfo=timezone.utc), + "7.5 mm", + ) + assert pd.isna(tube_events.iloc[1].iloc[3]) diff --git a/tests/test_snakemake_integration.py b/tests/test_snakemake_integration.py index 6fbc866..3c53b7b 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" + "SCHEMA_NAME=\n" # in testing mode, value doesn't matter but it has to exist ) # Collect coverage from Python processes inside the exporter container @@ -406,14 +411,14 @@ def _run_snakemake(tmp_path): compose_args, cwd=REPO_ROOT, ) - # for convenience print the snakemake log files if they exist (on success or error) + # for debugging convenience print all the log files if they exist (on success or error) outer_logs_dir = tmp_path / "snakemake-logs" - outer_logs = sorted(outer_logs_dir.glob("snakemake-outer-log*.log")) - if not outer_logs: - print("No outer logs found") - for ol in outer_logs: - print(f"Log file {ol}:") - print(ol.read_text()) + all_logs = sorted(outer_logs_dir.rglob("*.log")) + if not all_logs: + print("No log files found") + for lf in all_logs: + print(f"Log file {lf}:") + print(lf.read_text()) # print all output then raise if there was an error print(f"stdout:\n{result.stdout}\n" f"stderr:\n{result.stderr}") result.check_returncode() diff --git a/uv.lock b/uv.lock index b063abf..92a3903 100644 --- a/uv.lock +++ b/uv.lock @@ -935,6 +935,45 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/5e/75/bd9b7bb966668920f06b200e84454c8f3566b102183bc55c5473d96cb2b9/msal_extensions-1.3.1-py3-none-any.whl", hash = "sha256:96d3de4d034504e969ac5e85bae8106c8373b5c6568e4c8fa7af2eca9dbe6bca", size = 20583, upload-time = "2025-03-14T23:51:03.016Z" }, ] +[[package]] +name = "mssql-python" +version = "1.15.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "azure-identity" }, + { name = "mssql-python-odbc" }, +] +wheels = [ + { url = "https://files.pythonhosted.org/packages/9d/0e/1c879dcab6f73e521b734f9224ac68b394958f267e6a278b76f6fae640d0/mssql_python-1.15.0-cp313-cp313-macosx_15_0_universal2.whl", hash = "sha256:2e9d78861d4f97d779ad770b9b4af26c1aa332e2d877b9df3c8896fe766968c6", size = 9960882, upload-time = "2026-09-11T15:22:13.403Z" }, + { url = "https://files.pythonhosted.org/packages/85/26/40b7cd3a6cb5ae13b3278192f31df3fa2ef3e153255d56662d71da2ae772/mssql_python-1.15.0-cp313-cp313-manylinux_2_28_aarch64.whl", hash = "sha256:1614c5d32fe5ddf30bb5a576412280eeb588a7bf371a6a81de3588c741059c13", size = 7130021, upload-time = "2026-09-11T15:22:15.609Z" }, + { url = "https://files.pythonhosted.org/packages/75/c7/591359e2ce07bf8024d372484be8fbb58d292425381763487d16f2fab626/mssql_python-1.15.0-cp313-cp313-manylinux_2_28_x86_64.whl", hash = "sha256:384d5729eed457047cfb3b506f489950700f10d285370d388312807b3bcffa17", size = 7913266, upload-time = "2026-09-11T15:22:17.198Z" }, + { url = "https://files.pythonhosted.org/packages/fd/aa/275d459975bc06b92826226e1f631e8a6e59e27325927bc487a8d16abf49/mssql_python-1.15.0-cp313-cp313-musllinux_1_2_aarch64.whl", hash = "sha256:f92146ce5ba3f52838041df2cc699be508a2632fc50c25599844f7b50a016fb4", size = 6872555, upload-time = "2026-09-11T15:22:19.274Z" }, + { url = "https://files.pythonhosted.org/packages/59/e6/8f9825141b9875728cba0a5f402ad408e8fbab741412c6747e98b021e639/mssql_python-1.15.0-cp313-cp313-musllinux_1_2_x86_64.whl", hash = "sha256:e16bcad380466cc6232852d372f48ef4c23c981f4d5e12698957b398c4bd04e2", size = 7585129, upload-time = "2026-09-11T15:22:21.266Z" }, + { url = "https://files.pythonhosted.org/packages/5e/2f/79ee4ec452e02af4fa3146818b976c1b00f0c8d8362faf031d292c4bca9a/mssql_python-1.15.0-cp313-cp313-win_amd64.whl", hash = "sha256:1f88f4497d2e40c75bfeb37b6fe0a790b661ac9049dd85f4a5b85143c2cd1f68", size = 5183190, upload-time = "2026-09-11T15:22:23.028Z" }, + { url = "https://files.pythonhosted.org/packages/e1/8b/a51b4485b303aa536a9fec1a84ac5927b81932a9764e4cabb0645f7933ab/mssql_python-1.15.0-cp313-cp313-win_arm64.whl", hash = "sha256:7b08b2906ff7339f9fb1752b21a6f25a8a00c36fc311a1371b9ca4c2bd66a0e3", size = 4930532, upload-time = "2026-09-11T15:22:24.855Z" }, + { url = "https://files.pythonhosted.org/packages/cc/fc/0942797afac9b2cf74b59ce3255364040ba3f50ba551c701df6e6262ffdb/mssql_python-1.15.0-cp314-cp314-macosx_15_0_universal2.whl", hash = "sha256:6b4f7826e22e7c0ae8613cf78509a81ce4f5d95f7a16b10493bc073653bb0176", size = 9971965, upload-time = "2026-09-11T15:22:27.039Z" }, + { url = "https://files.pythonhosted.org/packages/ce/6a/361e2e7fd7e1fccfcc3e7e56ab6793b0018b4b61597e6afd21171a350c47/mssql_python-1.15.0-cp314-cp314-manylinux_2_28_aarch64.whl", hash = "sha256:061670881c8b6f395aa1dad606fe2ff68af579367f91effde436e9f10042152d", size = 7679594, upload-time = "2026-09-11T15:22:29.259Z" }, + { url = "https://files.pythonhosted.org/packages/c7/bf/50d2e0911f383038e59cda0730c52a82a1fcd0763d084ebefd6b3f5387cd/mssql_python-1.15.0-cp314-cp314-manylinux_2_28_x86_64.whl", hash = "sha256:69640f8ea7726e9c1d7b673098baec4b7b637bb5b144764e2e688cade664064d", size = 8642553, upload-time = "2026-09-11T15:22:31.258Z" }, + { url = "https://files.pythonhosted.org/packages/0c/67/5fb9ec913454cddd4a27ce9e142b1b5edc01d96dec25e8ec2c15ef12311f/mssql_python-1.15.0-cp314-cp314-musllinux_1_2_aarch64.whl", hash = "sha256:dcb141191daf6214f614907e40de2559aba9ff97f570ab2d4e3c1f02958c7e76", size = 7359238, upload-time = "2026-09-11T15:22:35.151Z" }, + { url = "https://files.pythonhosted.org/packages/96/c9/f3bdb4bf3fcac6102abdc4194a4b43b77a8da155e69c4c8aa1c18dbce265/mssql_python-1.15.0-cp314-cp314-musllinux_1_2_x86_64.whl", hash = "sha256:c7305c55ba30bb9135e7273ffcb6624a1d25546c8b09033e12d9631d5f1e8e65", size = 8240128, upload-time = "2026-09-11T15:22:36.92Z" }, + { url = "https://files.pythonhosted.org/packages/af/e8/0793bdcd7016adf52c8f39ac8e1abd1f0eeef658c75d25e90ecc36a3a733/mssql_python-1.15.0-cp314-cp314-win_amd64.whl", hash = "sha256:cf265057da5337aa134358b97f80b0e4a118661848ccaff52db61d912c36d6f5", size = 5407508, upload-time = "2026-09-11T15:22:38.51Z" }, + { url = "https://files.pythonhosted.org/packages/6f/f3/175854dc9ca8ad74cd1823c94d82cfbd58b8f9602401f056f834245125d6/mssql_python-1.15.0-cp314-cp314-win_arm64.whl", hash = "sha256:f91c5d9027ed044d0d954525a05ce87fbcc5ac5a9c02072f0e50e391f37ed3dc", size = 5167181, upload-time = "2026-09-11T15:43:51.102Z" }, +] + +[[package]] +name = "mssql-python-odbc" +version = "18.6.2.1" +source = { registry = "https://pypi.org/simple" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/e2/21/5d8e34820d986426dc65f88702ccecd1a981184e6bffab2968011461eaef/mssql_python_odbc-18.6.2.1-py3-none-macosx_15_0_universal2.whl", hash = "sha256:c0c4d2446f37656be1ed32c8e3229233554c1fbfc9f23c73c2ab0d07a2bfdcf2", size = 2035381, upload-time = "2026-08-05T15:51:46.762Z" }, + { url = "https://files.pythonhosted.org/packages/85/98/e07617483dd395156fba003cda6ff85ca9ad64de1347b823951f2afb3d60/mssql_python_odbc-18.6.2.1-py3-none-manylinux_2_28_aarch64.whl", hash = "sha256:d665529c195829970dcb72b7b3200e3c0ee1b053bee88e9a6eabaa385874dd6e", size = 2730032, upload-time = "2026-08-05T15:51:48.6Z" }, + { url = "https://files.pythonhosted.org/packages/35/1f/6cdb575549fbbe6f49bcadb505768649c43226bd83ec57a53a934d48e459/mssql_python_odbc-18.6.2.1-py3-none-manylinux_2_28_x86_64.whl", hash = "sha256:b8aad821f3b8637b0e6d4637f4b730ed9897c4665fa99d47e44e97fd1c79c77a", size = 3897879, upload-time = "2026-08-05T15:51:50.134Z" }, + { url = "https://files.pythonhosted.org/packages/89/ed/fa18a5f48eb4f0f003010af808b032806082c5441a076d698835ed5151b6/mssql_python_odbc-18.6.2.1-py3-none-musllinux_1_2_aarch64.whl", hash = "sha256:bcfca2a994b17499dc24a8b40228b851ccd0cd5807def855a6c41a8cedc4f8ac", size = 2730032, upload-time = "2026-08-05T15:51:51.672Z" }, + { url = "https://files.pythonhosted.org/packages/6c/66/7c96d719dfe7e9dcbbbf7acb345c72dfaef392fa6c5042e2718dfb1779b4/mssql_python_odbc-18.6.2.1-py3-none-musllinux_1_2_x86_64.whl", hash = "sha256:3db256103b09156f0980bee2dedc47995f528f76d01388ab5c5f1758b7e8fc53", size = 3897877, upload-time = "2026-08-05T15:51:53.147Z" }, + { url = "https://files.pythonhosted.org/packages/bb/f6/e7725d5dc791f5535fa7ec750c2ab9a1ff731f923d01b9c0d5cf1779b676/mssql_python_odbc-18.6.2.1-py3-none-win_amd64.whl", hash = "sha256:07a987221f99b368161db912065ef9e64c85f6b0901a2b05e9c1e6b41da08d55", size = 3712266, upload-time = "2026-08-05T15:51:54.589Z" }, + { url = "https://files.pythonhosted.org/packages/ea/d8/cda7ae46dd3c723196043109204fcc7749d688194de834b3f8aa4cd94415/mssql_python_odbc-18.6.2.1-py3-none-win_arm64.whl", hash = "sha256:00e1eae3c44a157ca60f992c6305be45856f18276ccff5258c1e9619847f3a5a", size = 7017266, upload-time = "2026-08-05T15:51:56.958Z" }, +] + [[package]] name = "multidict" version = "6.7.0" @@ -2282,6 +2321,7 @@ version = "0.1.0" source = { virtual = "." } dependencies = [ { name = "core" }, + { name = "mssql-python" }, { name = "opentelemetry-distro" }, { name = "opentelemetry-exporter-otlp-proto-http" }, { name = "pandas" }, @@ -2308,12 +2348,13 @@ dev = [ requires-dist = [ { name = "core", directory = "../PIXL/pixl_core" }, { name = "coverage", marker = "extra == 'coverage'", specifier = ">=7.0" }, + { name = "mssql-python", specifier = ">=1.13.0" }, { name = "opentelemetry-distro", specifier = "==0.63b1" }, { name = "opentelemetry-exporter-otlp-proto-http", specifier = "==1.42.1" }, { name = "pandas", specifier = "==2.3.2" }, { name = "pika", specifier = ">=1.3.2" }, { name = "pre-commit", specifier = ">=4.5.0" }, - { name = "psycopg2-binary", specifier = ">=2.9.10" }, + { name = "psycopg2-binary", specifier = ">=2.9.11" }, { name = "pytest", marker = "extra == 'dev'", specifier = ">=9.0.2" }, { name = "pytest-cov", marker = "extra == 'dev'", specifier = ">=7.1.0" }, { name = "requests", specifier = "==2.33.0" },