EHR queries

.dockerignore

@@ -4,3 +4,4 @@
 .git
 **/.mypy_cache
 **/.pytest_tmp
+**/build

.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

.gitignore

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

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/

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:
 ```

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

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

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.

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)

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,

src/csv_writer.py

@@ -2,10 +2,18 @@
 
 import csv
 import json
+import pandas as pd
+
 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,
+    WAVEFORM_PSEUDONYMISED_EHR,
+    make_file_name,
+    FILE_STEM_PATTERN,
+    EHR_STEM_PATTERN_HASHED,
+)
 
 
 def create_file_name(
@@ -76,7 +84,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 = [
@@ -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