Upload DICOM images with FTPS

.github/workflows/main.yml

@@ -91,8 +91,7 @@ jobs:
       - name: Run tests
         working-directory: pixl_core/tests
         run: |
-          docker compose up --build --exit-code-from test
-          docker compose down
+          pytest
         env:
           ENV: test
 
@@ -235,7 +234,7 @@ jobs:
         name: Move cache
         run: |
           rm -rf /tmp/.buildx-cache
-          mv /tmp/.buildx-cache-new /tmp/.buildx-cache   
+          mv /tmp/.buildx-cache-new /tmp/.buildx-cache
 
       - name: Init Python
         uses: actions/setup-python@v4

.gitignore

@@ -147,3 +147,5 @@ dmypy.json
 
 # project specific files
 /exports/
+**/test_producer.csv
+pixl_core/tests/ftp-server/mounts/data/*

cli/tests/test_database.py

@@ -18,7 +18,6 @@
 import pytest
 from core.database import Extract, Image
 from core.patient_queue.message import Message
-from dateutil.tz import UTC
 from pixl_cli._database import filter_exported_or_add_to_db
 from sqlalchemy.orm import Session
 
@@ -32,7 +31,7 @@ def _make_message(project_name: str, accession_number: str, mrn: str) -> Message
         mrn=mrn,
         study_date=STUDY_DATE,
         procedure_occurrence_id=1,
-        omop_es_timestamp=datetime.datetime.now(tz=UTC),
+        omop_es_timestamp=datetime.datetime.now(tz=datetime.timezone.utc),
     )
 
 
@@ -56,7 +55,7 @@ def rows_in_session(db_session) -> Session:
         study_date=STUDY_DATE,
         mrn="mrn",
         extract=extract,
-        exported_at=datetime.datetime.now(tz=UTC),
+        exported_at=datetime.datetime.now(tz=datetime.timezone.utc),
     )
     image_not_exported = Image(
         accession_number="234",

orthanc/orthanc-anon/plugin/pixl.py

@@ -31,12 +31,17 @@
 
 import requests
 import yaml
+from core import upload
 from decouple import config
 from pydicom import dcmread
 
 import orthanc
 import pixl_dcmd
 
+ORTHANC_USERNAME = config("ORTHANC_USERNAME")
+ORTHANC_PASSWORD = config("ORTHANC_PASSWORD")
+ORTHANC_URL = "http://localhost:8042"
+
 
 def AzureAccessToken():
     """
@@ -75,9 +80,6 @@ def AzureDICOMTokenRefresh():
 
     orthanc.LogWarning("Refreshing Azure DICOM token")
 
-    ORTHANC_USERNAME = config("ORTHANC_USERNAME")
-    ORTHANC_PASSWORD = config("ORTHANC_PASSWORD")
-
     AZ_DICOM_TOKEN_REFRESH_SECS = int(config("AZ_DICOM_TOKEN_REFRESH_SECS"))
     AZ_DICOM_ENDPOINT_NAME = config("AZ_DICOM_ENDPOINT_NAME")
     AZ_DICOM_ENDPOINT_URL = config("AZ_DICOM_ENDPOINT_URL")
@@ -105,7 +107,7 @@ def AzureDICOMTokenRefresh():
 
     headers = {"content-type": "application/json"}
 
-    url = "http://localhost:8042/dicom-web/servers/" + AZ_DICOM_ENDPOINT_NAME
+    url = ORTHANC_URL + "/dicom-web/servers/" + AZ_DICOM_ENDPOINT_NAME
 
     try:
         requests.put(
@@ -131,12 +133,9 @@ def SendViaStow(resourceId):
     Makes a POST API call to upload the resource to a dicom-web server
     using orthanc credentials as authorisation
     """
-    ORTHANC_USERNAME = config("ORTHANC_USERNAME")
-    ORTHANC_PASSWORD = config("ORTHANC_PASSWORD")
-
     AZ_DICOM_ENDPOINT_NAME = config("AZ_DICOM_ENDPOINT_NAME")
 
-    url = "http://localhost:8042/dicom-web/servers/" + AZ_DICOM_ENDPOINT_NAME + "/stow"
+    url = ORTHANC_URL + "/dicom-web/servers/" + AZ_DICOM_ENDPOINT_NAME + "/stow"
 
     headers = {"content-type": "application/json"}
 
@@ -156,6 +155,29 @@ def SendViaStow(resourceId):
         orthanc.LogError("Failed to send via STOW")
 
 
+def SendViaFTPS(resourceId: str) -> None:
+    """
+    Makes a POST API call to upload the resource to a dicom-web server
+    using orthanc credentials as authorisation
+    """
+    # Query orthanc-anon for the study
+    query = f"{ORTHANC_URL}/studies/{resourceId}/archive"
+    try:
+        response_study = requests.get(query, auth=(ORTHANC_USERNAME, ORTHANC_PASSWORD), timeout=10)
+        success_code = 200
+        if response_study.status_code != success_code:
+            msg = "Could not download archive of resource '%s'"
+            raise RuntimeError(msg, resourceId)
+    except requests.exceptions.RequestException:
+        orthanc.LogError(f"Failed to query'{resourceId}'")
+
+    # get the zip content
+    zip_content = response_study.content
+    logging.info("Downloaded data for resource %s", resourceId)
+
+    upload.upload_dicom_image(zip_content, resourceId)
+
+
 def ShouldAutoRoute():
     """
     Checks whether ORTHANC_AUTOROUTE_ANON_TO_AZURE environment variable is

pixl_core/README.md

@@ -11,14 +11,8 @@ pip install -e .
 ### Test
 
 ```bash
-pip install .[test]
-pytest -m "not pika"
-```
-or the full set with
-```bash
-cd tests
-docker compose up --build --exit-code-from test
-docker compose down
+pip install -e .[test]
+pytest 
 ```
 
 ## Token buffer

pixl_core/src/core/queries.py

@@ -0,0 +1,77 @@
+#  Copyright (c) University College London Hospitals NHS Foundation Trust
+#
+#  Licensed under the Apache License, Version 2.0 (the "License");
+#  you may not use this file except in compliance with the License.
+#  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+#  limitations under the License.
+
+"""Interaction with the PIXL database."""
+
+from datetime import datetime
+
+from decouple import config
+from sqlalchemy import URL, create_engine
+from sqlalchemy.orm import Session, sessionmaker
+
+from core.database import Extract, Image
+
+url = URL.create(
+    drivername="postgresql+psycopg2",
+    username=config("PIXL_DB_USER", default="None"),
+    password=config("PIXL_DB_PASSWORD", default="None"),
+    host=config("PIXL_DB_HOST", default="None"),
+    port=config("PIXL_DB_PORT", default=1),
+    database=config("PIXL_DB_NAME", default="None"),
+)
+
+engine = create_engine(url)
+
+
+def get_project_slug_from_db(hashed_value: str) -> str:
+    """
+    Get the project slug from the PIXL database for a given hashed identifier.
+    Throws an exception if the image has already been exported.
+    """
+    PixlSession = sessionmaker(engine)
+    with PixlSession() as pixl_session, pixl_session.begin():
+        existing_image = _query_existing_image(pixl_session, hashed_value)
+
+        if existing_image.exported_at is not None:
+            msg = "Image already exported"
+            raise RuntimeError(msg)
+
+        existing_extract = (
+            pixl_session.query(Extract)
+            .filter(
+                Extract.extract_id == existing_image.extract_id,
+            )
+            .one()
+        )
+
+        return str(existing_extract.slug)
+
+
+def update_exported_at(hashed_value: str, date_time: datetime) -> None:
+    """Update the `exported_at` field for an image in the PIXL database"""
+    PixlSession = sessionmaker(engine)
+    with PixlSession() as pixl_session, pixl_session.begin():
+        existing_image = _query_existing_image(pixl_session, hashed_value)
+        existing_image.exported_at = date_time
+        pixl_session.add(existing_image)
+
+
+def _query_existing_image(pixl_session: Session, hashed_value: str) -> Image:
+    return (
+        pixl_session.query(Image)
+        .filter(
+            Image.hashed_identifier == hashed_value,
+        )
+        .one()
+    )

pixl_core/src/core/upload.py

@@ -0,0 +1,103 @@
+#  Copyright (c) University College London Hospitals NHS Foundation Trust
+#
+#  Licensed under the Apache License, Version 2.0 (the "License");
+#  you may not use this file except in compliance with the License.
+#  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+#  limitations under the License.
+"""Functionality to upload files to an endpoint."""
+
+from __future__ import annotations
+
+import ftplib
+import logging
+import os
+import ssl
+from datetime import datetime, timezone
+from ftplib import FTP_TLS
+from typing import TYPE_CHECKING, Any, BinaryIO
+
+if TYPE_CHECKING:
+    from socket import socket
+
+from core.queries import get_project_slug_from_db, update_exported_at
+
+logger = logging.getLogger(__name__)
+
+
+class ImplicitFtpTls(ftplib.FTP_TLS):
+    """
+    FTP_TLS subclass that automatically wraps sockets in SSL to support implicit FTPS.
+
+    https://stackoverflow.com/questions/12164470/python-ftp-implicit-tls-connection-issue
+    """
+
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
+        """Create instance from parent class."""
+        super().__init__(*args, **kwargs)
+        self._sock: socket | None = None
+
+    @property
+    def sock(self) -> socket | None:
+        """Return the socket."""
+        return self._sock
+
+    @sock.setter
+    def sock(self, value: socket) -> None:
+        """When modifying the socket, ensure that it is ssl wrapped."""
+        if value is not None and not isinstance(value, ssl.SSLSocket):
+            value = self.context.wrap_socket(value)
+        self._sock = value
+
+
+def upload_dicom_image(zip_content: BinaryIO, pseudo_anon_id: str) -> None:
+    """Top level way to upload an image."""
+    # rename destination to {project-slug}/{study-pseduonymised-id}.zip
+    remote_directory = get_project_slug_from_db(pseudo_anon_id)
+
+    # Create the remote directory if it doesn't exist
+    ftp = _connect_to_ftp()
+    _create_and_set_as_cwd(ftp, remote_directory)
+    command = f"STOR {pseudo_anon_id}.zip"
+    logger.debug("Running %s", command)
+
+    # Store the file using a binary handler
+    ftp.storbinary(command, zip_content)
+
+    # Close the FTP connection
+    ftp.quit()
+    logger.debug("Finished uploading!")
+
+    update_exported_at(pseudo_anon_id, datetime.now(tz=timezone.utc))
+
+
+def _connect_to_ftp() -> FTP_TLS:
+    # Set your FTP server details
+    ftp_host = os.environ["FTP_HOST"]
+    ftp_port = os.environ["FTP_PORT"]  # FTPS usually uses port 21
+    ftp_user = os.environ["FTP_USER_NAME"]
+    ftp_password = os.environ["FTP_USER_PASS"]
+
+    # Connect to the server and login
+    ftp = ImplicitFtpTls()
+    ftp.connect(ftp_host, int(ftp_port))
+    ftp.login(ftp_user, ftp_password)
+    ftp.prot_p()
+    return ftp
+
+
+def _create_and_set_as_cwd(ftp: FTP_TLS, project_dir: str) -> None:
+    try:
+        ftp.cwd(project_dir)
+        logger.info("'%s' exists on remote ftp, so moving into it", project_dir)
+    except ftplib.error_perm:
+        logger.info("creating '%s' on remote ftp and moving into it", project_dir)
+        # Directory doesn't exist, so create it
+        ftp.mkd(project_dir)
+        ftp.cwd(project_dir)