ehr-api becomes export-api; losing and gaining some features

.env.sample

@@ -83,11 +83,6 @@ AZ_DICOM_ENDPOINT_CLIENT_ID=
 AZ_DICOM_ENDPOINT_CLIENT_SECRET=
 AZ_DICOM_ENDPOINT_TENANT_ID=
 
-# EHR extraction API
-PIXL_EHR_API_AZ_STORAGE_ACCOUNT_NAME=
-PIXL_EHR_API_AZ_STORAGE_CONTAINER_NAME=
-PIXL_EHR_COGSTACK_REDACT_URL=
-
 # RABBIT MQ queue. UI available at localhost:$RABBITMQ_ADMIN_PORT
 RABBITMQ_HOST=localhost
 RABBITMQ_USERNAME=

.github/workflows/main.yml

@@ -142,7 +142,7 @@ jobs:
         run: |
           pytest
 
-  ehr-api-tests:
+  export-api-tests:
     runs-on: ubuntu-22.04
     timeout-minutes: 30
     steps:
@@ -155,10 +155,10 @@ jobs:
 
       - name: Install Python dependencies
         run: |
-          pip install pytest-pixl/ pixl_core/ pixl_ehr/[test]
+          pip install pytest-pixl/ pixl_core/ pixl_export/[test]
 
       - name: Run tests
-        working-directory: pixl_ehr/tests
+        working-directory: pixl_export/tests
         run: |
           pytest
 
@@ -254,10 +254,10 @@ jobs:
           ./run-system-test.sh
           echo FINISHED SYSTEM TEST SCRIPT
 
-      - name: Dump ehr-api docker logs for debugging
+      - name: Dump export-api docker logs for debugging
         if: ${{ failure() }}
         run: |
-          docker logs -t system-test-ehr-api-1 2>&1
+          docker logs -t system-test-export-api-1 2>&1
 
       - name: Dump orthanc-anon docker logs for debugging
         if: ${{ failure() }}

README.md

@@ -72,7 +72,7 @@ Provides helper functions for de-identifying DICOM data
 
 RDBMS which stores DICOM metadata, application data and anonymised patient record data.
 
-### [Electronic Health Record Extractor](./pixl_ehr/README.md)
+### [Electronic Health Record Extractor](pixl_export/README.md)
 
 HTTP API to process messages from the `ehr` queue and populate raw and anon tables in the PIXL postgres instance.
 
@@ -122,11 +122,6 @@ The maximum storage size of the `orthanc-raw` instance can be configured through
 the specified value (in MB). When the storage is full [Orthanc will automatically recycle older
 studies in favour of new ones](https://orthanc.uclouvain.be/book/faq/features.html#id8).
 
-#### CogStack URL
-
-For the deidentification of the EHR extracts, we rely on an instance running the
-[CogStack API](https://cogstack.org/) with a `/redact` endpoint. The URL of this instance should be
-set in `.env` as `COGSTACK_REDACT_URL`.
 
 ### 3. Configure a new project
 
@@ -284,7 +279,8 @@ EXTRACT_DIR/public /*.parquet
 
 ### PIXL Export dir (PIXL intermediate)
 
-The directory where PIXL will copy the public OMOP extract files and radiology reports to.
+The directory where PIXL will copy the public OMOP extract files (which now contain
+the radiology reports) to.
 These files will subsequently be uploaded to the `parquet` destination specified in the
 [project config](#3-configure-a-new-project).
 

cli/README.md

@@ -25,7 +25,7 @@ pip install -e ../pixl_core/ -e .
 
 ## Usage
 
-> **Note** The `rabbitmq`, `ehr-api` and `imaging-api` services must be started prior to using the CLI
+> **Note** The `rabbitmq`, `export-api` and `imaging-api` services must be started prior to using the CLI
 > This is typically done by spinning up the necessary Docker containers through `docker compose`.
 > For convenience, we provide the [`bin/pixldc`](../bin/pixldc) script to spin up the relevant
 > services in production.
@@ -69,7 +69,7 @@ where the `*_RATE` variables set the default querying rate for the message queue
 
 ### Running the pipeline
 
-Populate queue for Imaging and EHR extraction
+Populate queue for Imaging
 
 ```bash
 pixl populate </path/to/parquet_dir>
@@ -92,15 +92,13 @@ customisation of the rate per queue is required or a queue should not be started
 then supply the argument `--no-start` and use `pixl start...` to launch
 processing.
 
-Once the messages have been processed, the radiology reports can be extracted and exported to a
-`parquet file` using
+Once the messages have been processed, the OMOP extracts (including radiology reports) can be
+exported to a `parquet file` using
 
 ```sh
-pixl extract-radiology-reports </path/to/parquet_dir>
+pixl export-patient-data </path/to/parquet_dir>
 ```
 
-This will also upload the radiology reports to the destination specified in the project config.
-
 Stop Imaging and EHR database extraction
 
 ```bash
@@ -123,7 +121,6 @@ Options:
   --help                Show this message and exit.
 
 Commands:
-  az-copy-ehr                Copy the EHR data to azure
   extract-radiology-reports  Export processed radiology reports to...
   kill                       Stop all the PIXL services
   populate                   Populate a (set of) queue(s) from a parquet...

cli/src/pixl_cli/_config.py

@@ -54,7 +54,7 @@ def base_url(self) -> str:
 
 
 API_CONFIGS = {
-    "ehr_api": APIConfig(
+    "export_api": APIConfig(
         host=config("PIXL_EHR_API_HOST"),
         port=int(config("PIXL_EHR_API_PORT")),
         default_rate=float(config("PIXL_EHR_API_RATE", default=1)),

cli/src/pixl_cli/_database.py

@@ -14,6 +14,8 @@
 
 """Interaction with the PIXL database."""
 
+from typing import cast
+
 from core.db.models import Extract, Image
 from core.patient_queue.message import Message
 from sqlalchemy import URL, create_engine
@@ -110,3 +112,13 @@ def _add_new_image_to_session(extract: Extract, message: Message, session: Sessi
     )
     session.add(new_image)
     return new_image
+
+
+def images_for_project(project_slug: str) -> list[Image]:
+    """Given a project, get all images in the DB for that project."""
+    PixlSession = sessionmaker(engine)
+    with PixlSession() as session:
+        return cast(
+            list[Image],
+            session.query(Image).join(Extract).filter(Extract.slug == project_slug).all(),
+        )

cli/src/pixl_cli/_io.py

@@ -18,14 +18,18 @@
 import json
 from datetime import UTC, datetime
 from pathlib import Path
+from typing import TYPE_CHECKING
 
 import pandas as pd
 from core.exports import ParquetExport
 from core.patient_queue.message import Message
 from loguru import logger
 
+if TYPE_CHECKING:
+    from core.db.models import Image
+
 # The export root dir from the point of view of the docker host (which is where the CLI runs)
-# For the view from inside, see pixl_ehr/main.py: EHR_EXPORT_ROOT_DIR
+# For the view from inside, see pixl_export/main.py: EHR_EXPORT_ROOT_DIR
 HOST_EXPORT_ROOT_DIR = Path(__file__).parents[3] / "projects" / "exports"
 
 
@@ -43,8 +47,8 @@ def project_info(resources_path: Path) -> tuple[str, datetime]:
 
 def copy_parquet_return_logfile_fields(resources_path: Path) -> tuple[str, datetime]:
     """Copy public parquet file to extracts directory, and return fields from logfile"""
-    project_name, extract_generated_timestamp = project_info(resources_path)
-    extract = ParquetExport(project_name, extract_generated_timestamp, HOST_EXPORT_ROOT_DIR)
+    project_name_raw, extract_generated_timestamp = project_info(resources_path)
+    extract = ParquetExport(project_name_raw, extract_generated_timestamp, HOST_EXPORT_ROOT_DIR)
     project_name_slug = extract.copy_to_exports(resources_path)
     return project_name_slug, extract_generated_timestamp
 
@@ -117,26 +121,14 @@ def messages_from_parquet(
     private_dir = dir_path / "private"
 
     cohort_data = _check_and_parse_parquet(private_dir, public_dir)
-
-    map_column_to_message_params = {
-        "mrn": "PrimaryMrn",
-        "accession_number": "AccessionNumber",
-        "study_date": "procedure_date",
-        "procedure_occurrence_id": "procedure_occurrence_id",
-    }
-
-    _raise_if_column_names_not_found(cohort_data, list(map_column_to_message_params.values()))
+    cohort_data_mapped = _map_columns(cohort_data)
 
     messages = []
-
-    for _, row in cohort_data.iterrows():
-        message_params = {
-            param: row[column] for param, column in map_column_to_message_params.items()
-        }
+    for _, row in cohort_data_mapped.iterrows():
         message = Message(
             project_name=project_name,
             extract_generated_timestamp=extract_generated_timestamp,
-            **message_params,
+            **{column: row[column] for column in MAP_PARQUET_TO_MESSAGE_KEYS.values()},
         )
         messages.append(message)
 
@@ -148,6 +140,19 @@ def messages_from_parquet(
     return messages
 
 
+MAP_PARQUET_TO_MESSAGE_KEYS = {
+    "PrimaryMrn": "mrn",
+    "AccessionNumber": "accession_number",
+    "procedure_date": "study_date",
+    "procedure_occurrence_id": "procedure_occurrence_id",
+}
+
+
+def _map_columns(input_df: pd.DataFrame) -> pd.DataFrame:
+    _raise_if_column_names_not_found(input_df, list(MAP_PARQUET_TO_MESSAGE_KEYS.keys()))
+    return input_df.rename(MAP_PARQUET_TO_MESSAGE_KEYS, axis=1)
+
+
 def _check_and_parse_parquet(private_dir: Path, public_dir: Path) -> pd.DataFrame:
     for d in [public_dir, private_dir]:
         if not d.is_dir():
@@ -167,6 +172,23 @@ def _check_and_parse_parquet(private_dir: Path, public_dir: Path) -> pd.DataFram
     return people_procedures_accessions[~people_procedures_accessions["AccessionNumber"].isna()]
 
 
+def make_radiology_linker_table(parquet_dir: Path, images: list[Image]) -> pd.DataFrame:
+    """
+    Make a table linking the OMOP procedure_occurrence_id to the hashed image/study ID.
+    :param parquet_dir: location of OMOP extract
+                        (this gives us: procedure_occurrence_id <-> mrn+accession mapping)
+    :param images: the images already processed by PIXL, from the DB
+                        (this gives us: mrn+accession <-> hashed ID)
+    """
+    public_dir = parquet_dir / "public"
+    private_dir = parquet_dir / "private"
+    people_procedures_accessions = _map_columns(_check_and_parse_parquet(private_dir, public_dir))
+
+    images_df = pd.DataFrame.from_records([vars(im) for im in images])
+    merged = people_procedures_accessions.merge(images_df, on=("mrn", "accession_number"))
+    return merged[["procedure_occurrence_id", "hashed_identifier"]]
+
+
 def _raise_if_column_names_not_found(
     cohort_data: pd.DataFrame, expected_col_names: list[str]
 ) -> None:

cli/src/pixl_cli/main.py

@@ -24,14 +24,17 @@
 
 import click
 import requests
+from core.exports import ParquetExport
 from core.patient_queue.producer import PixlProducer
 from decouple import RepositoryEnv, UndefinedValueError, config
 from loguru import logger
 
 from pixl_cli._config import SERVICE_SETTINGS, api_config_for_queue
-from pixl_cli._database import filter_exported_or_add_to_db
+from pixl_cli._database import filter_exported_or_add_to_db, images_for_project
 from pixl_cli._io import (
+    HOST_EXPORT_ROOT_DIR,
     copy_parquet_return_logfile_fields,
+    make_radiology_linker_table,
     messages_from_csv,
     messages_from_parquet,
     project_info,
@@ -82,7 +85,7 @@ def check_env(*, error: bool, sample_env_file: click.Path) -> None:
 @cli.command()
 @click.option(
     "--queues",
-    default="imaging,ehr",
+    default="imaging",
     show_default=True,
     help="Comma seperated list of queues to populate with messages generated from the "
     "input file(s)",
@@ -146,20 +149,25 @@ def populate(
 @click.argument(
     "parquet-dir", required=True, type=click.Path(path_type=Path, exists=True, file_okay=False)
 )
-def extract_radiology_reports(parquet_dir: Path) -> None:
+def export_patient_data(parquet_dir: Path) -> None:
     """
     Export processed radiology reports to parquet file.
 
-    PARQUET_DIR: Directory containing the extract_summary.json
-    log file defining which extract to export radiology reports for.
+    PARQUET_DIR: Directory containing the extract_summary.json log file
+                 defining which extract to export patient data for. (not a CSV)
     """
-    project_name, omop_es_datetime = project_info(parquet_dir)
+    project_name_raw, omop_es_datetime = project_info(parquet_dir)
+    export = ParquetExport(project_name_raw, omop_es_datetime, HOST_EXPORT_ROOT_DIR)
 
-    # Call the EHR API
-    api_config = api_config_for_queue("ehr")
+    images = images_for_project(export.project_slug)
+    linker_data = make_radiology_linker_table(parquet_dir, images)
+    export.export_radiology_linker(linker_data)
+
+    # Call the Export API
+    api_config = api_config_for_queue("export")
     response = requests.post(
         url=f"{api_config.base_url}/export-patient-data",
-        json={"project_name": project_name, "extract_datetime": omop_es_datetime.isoformat()},
+        json={"project_name": project_name_raw, "extract_datetime": omop_es_datetime.isoformat()},
         timeout=300,
     )
 
@@ -175,7 +183,7 @@ def extract_radiology_reports(parquet_dir: Path) -> None:
 @cli.command()
 @click.option(
     "--queues",
-    default="ehr,imaging",
+    default="imaging",
     show_default=True,
     help="Comma seperated list of queues to start consuming from",
 )
@@ -197,7 +205,7 @@ def start(queues: str, rate: Optional[float]) -> None:
 @cli.command()
 @click.option(
     "--queues",
-    default="ehr,imaging",
+    default="imaging",
     show_default=True,
     help="Comma seperated list of queues to update the consume rate of",
 )
@@ -250,7 +258,7 @@ def _update_extract_rate(queue_name: str, rate: Optional[float]) -> None:
 @cli.command()
 @click.option(
     "--queues",
-    default="ehr,imaging",
+    default="imaging",
     show_default=True,
     help="Comma seperated list of queues to consume messages from",
 )
@@ -286,7 +294,7 @@ def kill() -> None:
 @cli.command()
 @click.option(
     "--queues",
-    default="ehr,imaging",
+    default="imaging",
     show_default=True,
     help="Comma seperated list of queues to consume messages from",
 )
@@ -296,18 +304,6 @@ def status(queues: str) -> None:
         logger.info(f"[{queue:^10s}] refresh rate = ", _get_extract_rate(queue))
 
 
-@cli.command()
-def az_copy_ehr() -> None:
-    """Copy the EHR data to azure"""
-    api_config = api_config_for_queue("ehr")
-    response = requests.get(url=f"{api_config.base_url}/az-copy-current", timeout=10)
-
-    success_code = 200
-    if response.status_code != success_code:
-        msg = f"Failed to run az copy due to: {response.text}"
-        raise RuntimeError(msg)
-
-
 def _get_extract_rate(queue_name: str) -> str:
     """
     Get the extraction rate in items per second from a queue