Merge pull request #24 from SAFEHR-data/16-back_to_no_thread

Fixing errors whilst sticking to a single threaded application.

Author: thompson318

.github/workflows/pytest.yml

@@ -0,0 +1,25 @@
+name: Run pytest
+
+on: 
+  push:
+    branches: [main, dev]
+    pull_request:
+      branches: [main, dev]
+      types: ["opened", "reopened", "synchronize", "ready_for_review", "draft"]
+  workflow_dispatch:
+
+jobs:
+  build:
+    name: Run pytest
+    runs-on: ubuntu-latest
+
+    steps:
+      - uses: actions/checkout@v5
+      - name: Install uv
+        uses: astral-sh/setup-uv@v7
+
+      - name: Install dependencies
+        run: uv sync --locked --all-extras --dev
+
+      - name: Run the tests
+        run: uv run pytest tests

.pre-commit-config.yaml

@@ -10,7 +10,7 @@ repos:
       - id: ruff
         args:
           - --fix
-          - waveform_controller
+          - ./ 
       - id: ruff-format
   # Type-checking python code.
   - repo: https://github.com/pre-commit/mirrors-mypy
@@ -23,7 +23,7 @@ repos:
             "types-psycopg2",
             "types-pika"
           ]
-        files: waveform_controller/
+        files: src/
   # ----------
   # Formats docstrings to comply with PEP257
   - repo: https://github.com/PyCQA/docformatter

pyproject.toml

@@ -10,5 +10,8 @@ dependencies = [
     "psycopg2-binary>=2.9.11",
 ]
 
+[project.optional-dependencies]
+dev = ["pytest>=9.0.2"]
+
 [project.scripts]
-emap-extract-waveform = "waveform_controller.controller:receiver"
+emap-extract-waveform = "controller:receiver"

settings.env.EXAMPLE

@@ -5,6 +5,8 @@ UDS_USERNAME="inform_user"
 UDS_PASSWORD="inform"
 UDS_HOST="localhost"
 UDS_PORT="5433"
+UDS_CONNECT_TIMEOUT="10"  # in seconds
+UDS_QUERY_TIMEOUT="3000"  # in milliseconds
 SCHEMA_NAME="star_dev"
 RABBITMQ_USERNAME="my_name"
 RABBITMQ_PASSWORD="my_pw"

waveform_controller/__init__.py src/__init__.pywaveform_controller/__init__.py renamed to src/__init__.py

@@ -0,0 +1,116 @@
+"""
+A script to receive messages in the waveform queue and write them to stdout,
+based on https://www.rabbitmq.com/tutorials/tutorial-one-python
+"""
+
+import json
+from datetime import datetime, timezone
+import logging
+import pika
+import db as db  # type:ignore
+import settings as settings  # type:ignore
+import csv_writer as writer  # type:ignore
+
+logging.basicConfig(format="%(levelname)s:%(asctime)s: %(message)s")
+logger = logging.getLogger(__name__)
+
+
+emap_db = db.starDB()
+emap_db.init_query()
+emap_db.connect()
+
+
+class waveform_message:
+    def __init__(self, ch, delivery_tag, body):
+        self.ch = ch
+        self.delivery_tag = delivery_tag
+        self.body = body
+
+
+def ack_message(ch, delivery_tag):
+    """Note that `ch` must be the same pika channel instance via which the
+    message being ACKed was retrieved (AMQP protocol constraint)."""
+    if ch.is_open:
+        ch.basic_ack(delivery_tag)
+    else:
+        logger.warning("Attempting to acknowledge a message on a closed channel.")
+
+
+def reject_message(ch, delivery_tag, requeue):
+    if ch.is_open:
+        ch.basic_reject(delivery_tag, requeue)
+    else:
+        logger.warning("Attempting to not acknowledge a message on a closed channel.")
+
+
+def waveform_callback(ch, method_frame, _header_frame, body):
+    data = json.loads(body)
+    try:
+        location_string = data["mappedLocationString"]
+        observation_timestamp = data["observationTime"]
+        source_stream_id = data["sourceStreamId"]
+        sampling_rate = data["samplingRate"]
+        units = data["unit"]
+        waveform_data = data["numericValues"]
+        mapped_location_string = data["mappedLocationString"]
+    except IndexError as e:
+        reject_message(ch, method_frame.delivery_tag, False)
+        logger.error(
+            f"Waveform message {method_frame.delivery_tag} is missing required data {e}."
+        )
+        return
+
+    observation_time = datetime.fromtimestamp(observation_timestamp, tz=timezone.utc)
+    lookup_success = True
+    try:
+        matched_mrn = emap_db.get_row(location_string, observation_time)
+    except ValueError as e:
+        lookup_success = False
+        logger.error(f"Ambiguous or non existent match: {e}")
+        matched_mrn = ("unmatched_mrn", "unmatched_nhs", "unmatched_csn")
+    except ConnectionError as e:
+        logger.error(f"Database error, will try again: {e}")
+        reject_message(ch, method_frame.delivery_tag, True)
+        return
+
+    if writer.write_frame(
+        waveform_data,
+        source_stream_id,
+        observation_timestamp,
+        units,
+        sampling_rate,
+        mapped_location_string,
+        matched_mrn[2],
+        matched_mrn[0],
+    ):
+        if lookup_success:
+            ack_message(ch, method_frame.delivery_tag)
+        else:
+            reject_message(ch, method_frame.delivery_tag, False)
+
+
+def receiver():
+    # set up database connection
+    rabbitmq_credentials = pika.PlainCredentials(
+        username=settings.RABBITMQ_USERNAME, password=settings.RABBITMQ_PASSWORD
+    )
+    connection_parameters = pika.ConnectionParameters(
+        credentials=rabbitmq_credentials,
+        host=settings.RABBITMQ_HOST,
+        port=settings.RABBITMQ_PORT,
+    )
+    connection = pika.BlockingConnection(connection_parameters)
+    channel = connection.channel()
+    channel.basic_qos(prefetch_count=1)
+
+    channel.basic_consume(
+        queue=settings.RABBITMQ_QUEUE,
+        auto_ack=False,
+        on_message_callback=waveform_callback,
+    )
+    try:
+        channel.start_consuming()
+    except KeyboardInterrupt:
+        channel.stop_consuming()
+
+    connection.close()

src/controller.py

@@ -0,0 +1,59 @@
+"""Writes a frame of waveform data to a csv file."""
+
+import csv
+from datetime import datetime
+from pathlib import Path
+
+
+def create_file_name(
+    source_stream_id: str, observation_time: datetime, csn: str, units: str
+) -> str:
+    """Create a unique file name based on the patient contact serial number
+    (csn) the date, and the source system."""
+    datestring = observation_time.strftime("%Y-%m-%d")
+    units = units.replace("/", "p")
+    units = units.replace("%", "percent")
+    return f"{datestring}.{csn}.{source_stream_id}.{units}.csv"
+
+
+def write_frame(
+    waveform_data: dict,
+    source_stream_id: str,
+    observation_timestamp: float,
+    units: str,
+    sampling_rate: int,
+    mapped_location_string: str,
+    csn: str,
+    mrn: str,
+) -> bool:
+    """Appends a frame of waveform data to a csv file (creates file if it
+    doesn't exist.
+
+    :return: True if write was successful.
+    """
+    observation_datetime = datetime.fromtimestamp(observation_timestamp)
+
+    out_path = "waveform-export/"
+    Path(out_path).mkdir(exist_ok=True)
+
+    filename = out_path + create_file_name(
+        source_stream_id, observation_datetime, csn, units
+    )
+    with open(filename, "a") as fileout:
+        wv_writer = csv.writer(fileout, delimiter=",")
+        waveform_data = waveform_data.get("value", "")
+
+        wv_writer.writerow(
+            [
+                csn,
+                mrn,
+                source_stream_id,
+                units,
+                sampling_rate,
+                observation_timestamp,
+                mapped_location_string,
+                waveform_data,
+            ]
+        )
+
+    return True

src/csv_writer.py

@@ -0,0 +1,55 @@
+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]

src/db.py

@@ -13,6 +13,8 @@ def get_from_env(env_var, setting_name=None):
 get_from_env("UDS_PASSWORD")
 get_from_env("UDS_HOST")
 get_from_env("UDS_PORT")
+get_from_env("UDS_CONNECT_TIMEOUT")
+get_from_env("UDS_QUERY_TIMEOUT")
 get_from_env("SCHEMA_NAME")
 get_from_env("RABBITMQ_USERNAME")
 get_from_env("RABBITMQ_PASSWORD")

waveform_controller/settings.py src/settings.pywaveform_controller/settings.py renamed to src/settings.py

@@ -14,5 +14,5 @@ INNER JOIN {schema_name}.location_visit lv
 INNER JOIN {schema_name}.location loc
   ON lv.location_id = loc.location_id
 WHERE loc.location_string = %(location_string)s
-  AND hv.valid_from BETWEEN %(start_datetime)s AND %(end_datetime)s
-ORDER by hv.valid_from DESC
+  AND lv.admission_datetime <= %(observation_datetime)s
+  AND ( lv.discharge_datetime >= %(observation_datetime)s OR lv.discharge_datetime IS NULL )