Implement matching of DICOM tag values

src/serie/actions.py

@@ -1,13 +1,20 @@
 import asyncio
+import asyncio
 import dataclasses
 from collections.abc import Iterable, Sequence
+from typing import Optional
 
-from aiochris.models import Plugin, Feed, PACSFile, PluginInstance
+import aiochris.models
+from aiochris.models import Plugin, Feed, PluginInstance
 from aiochris.types import ChrisURL
 
 from serie.clients import get_plugin, get_client
-from serie.dicom_series_metadata import DicomSeriesMetadataName
-from serie.models import OxidicomCustomMetadata, ChrisRunnableRequest
+from serie.models import (
+    OxidicomCustomMetadata,
+    ChrisRunnableRequest,
+    PacsFile,
+)
+from serie.series_file_pair import DicomSeriesFilePair
 
 
 @dataclasses.dataclass(frozen=True)
@@ -19,9 +26,21 @@ class ClientActions:
     auth: str | None
     url: ChrisURL
 
+    async def resolve_series(self, data: PacsFile) -> Optional[DicomSeriesFilePair]:
+        """
+        Check whether the given PACSFile is a "OxidicomAttemptedPushCount=*" file
+        (which signifies that the reception of a DICOM series is complete). If so,
+        get one of the DICOM instances of the series from *CUBE*.
+        """
+        if (ocm := OxidicomCustomMetadata.from_pacsfile(data)) is None:
+            return None
+        if (pacs_file := await self._get_first_dicom_of(ocm)) is None:
+            return None
+        return DicomSeriesFilePair(ocm, pacs_file)
+
     async def create_analysis(
         self,
-        oxm_file: OxidicomCustomMetadata,
+        series: DicomSeriesFilePair,
         runnables_request: Iterable[ChrisRunnableRequest],
         feed_name_template: str,
     ) -> Feed:
@@ -35,13 +54,14 @@ async def create_analysis(
         pl_dircopy, pl_unstack_folders, plugins = await self._get_plugins(
             runnables_request
         )
-        dircopy_inst = await pl_dircopy.create_instance(dir=oxm_file.series_dir)
+        dircopy_inst = await pl_dircopy.create_instance(dir=series.series_dir)
         root_inst = await pl_unstack_folders.create_instance(previous=dircopy_inst)
         branches = (
             plugin.create_instance(previous=root_inst, **req.params)
             for plugin, req in zip(plugins, runnables_request)
         )
-        set_feed_name = self._set_feed_name(oxm_file, dircopy_inst, feed_name_template)
+        feed_name = _expand_variables(feed_name_template, series)
+        set_feed_name = self._set_feed_name(dircopy_inst, feed_name)
         feed, *_ = await asyncio.gather(set_feed_name, *branches)
         return feed
 
@@ -63,42 +83,32 @@ async def _get_plugins(
         pl_dircopy, pl_unstack_folders, *others = await asyncio.gather(*plugin_requests)  # noqa
         return pl_dircopy, pl_unstack_folders, others
 
-    async def _set_feed_name(
-        self, oxm_file: OxidicomCustomMetadata, plinst: PluginInstance, template: str
-    ) -> Feed:
-        """
-        Set the name of the plugin instance's feed using the given template and the related DICOM series.
-        """
-        feed, dicom = await asyncio.gather(
-            plinst.get_feed(), self._get_first_dicom_of(oxm_file)
-        )
-        name = template.format(**_feed_name_template_variables(oxm_file, dicom))
-        return await feed.set(name=name)
-
-    async def _get_first_dicom_of(self, oxm_file: OxidicomCustomMetadata) -> PACSFile:
+    async def _get_first_dicom_of(
+        self, oxm_file: OxidicomCustomMetadata
+    ) -> Optional[aiochris.models.PACSFile]:
         """
         Get an arbitrary DICOM file belonging to the series represented by ``oxm_file``.
         """
         chris = await get_client(self.url, self.auth)
-        dicom_file = await chris.search_pacsfiles(
+        return await chris.search_pacsfiles(
             pacs_identifier=oxm_file.pacs_identifier,
             PatientID=oxm_file.patient_id,
             StudyInstanceUID=oxm_file.study_instance_uid,
             SeriesInstanceUID=oxm_file.series_instance_uid,
         ).first()
-        if dicom_file is None:
-            raise ValueError(
-                "DICOM series not found. It is likely the given oxidicom custom file is invalid.",
-                oxm_file,
-            )
-        return dicom_file
+
+    @staticmethod
+    async def _set_feed_name(dircopy_inst: PluginInstance, name: str) -> Feed:
+        """
+        Get the feed of the given plugin instance, and set its feed name.
+        """
+        feed = await dircopy_inst.get_feed()
+        await feed.set(name=name)
+        return feed
 
 
-def _feed_name_template_variables(oxm_file: OxidicomCustomMetadata, dicom: PACSFile):
+def _expand_variables(template: str, series: DicomSeriesFilePair) -> str:
     """
-    Create a dict of values which can be used in a feed name template.
+    Expand the value of variables in ``template`` using field values from ``series``.
     """
-    values = {}
-    for name in DicomSeriesMetadataName:
-        values[name.value] = getattr(dicom, name, None) or getattr(oxm_file, name)
-    return values
+    return template.format(**series.to_dict())

src/serie/dicom_series_metadata.py

@@ -1,4 +1,5 @@
 import enum
+from typing import TypedDict
 
 
 class DicomSeriesMetadataName(enum.Enum):
@@ -19,3 +20,12 @@ class DicomSeriesMetadataName(enum.Enum):
     SeriesDescription = "SeriesDescription"
     pacs_identifier = "pacs_identifier"
     series_dir = "series_dir"
+
+
+DicomSeriesMetadata = TypedDict(
+    "DicomSeriesMetadata",
+    {name.value: None | bool | int | float | str for name in DicomSeriesMetadataName},
+)
+"""
+A :class:`dict` with the keys of :class:`DicomSeriesMetadataName`.
+"""

src/serie/match.py

@@ -0,0 +1,24 @@
+import re
+from collections.abc import Sequence
+
+from serie.dicom_series_metadata import DicomSeriesMetadata
+from serie.models import DicomSeriesMatcher
+from serie.series_file_pair import DicomSeriesFilePair
+
+
+def is_match(
+    series: DicomSeriesFilePair, conditions: Sequence[DicomSeriesMatcher]
+) -> bool:
+    """
+    :return: True if the series matches the conditions
+    """
+    series_dict = series.to_dict()
+    return all(_matches(cond, series_dict) for cond in conditions)
+
+
+def _matches(condition: DicomSeriesMatcher, series_dict: DicomSeriesMetadata) -> bool:
+    if condition.tag.value not in series_dict:
+        return False
+    value = series_dict[condition.tag.value]
+    flag = re.IGNORECASE if condition.case_sensitive else re.NOFLAG
+    return condition.regex.fullmatch(value) is not None

src/serie/models.py

@@ -3,7 +3,14 @@
 from typing import Literal, Optional, Self, ClassVar
 import re
 
-from pydantic import BaseModel, ConfigDict, NonNegativeInt, PastDatetime, Field
+from pydantic import (
+    BaseModel,
+    ConfigDict,
+    NonNegativeInt,
+    NonNegativeFloat,
+    PastDatetime,
+    Field,
+)
 
 from serie.dicom_series_metadata import DicomSeriesMetadataName
 
@@ -20,7 +27,7 @@ class PacsFile(BaseModel):
     patient_name: str = Field(alias="PatientName")
     patient_sex: Optional[str] = Field(alias="PatientSex")
     accession_number: str = Field(alias="AccessionNumber")
-    patient_age: Optional[NonNegativeInt] = Field(alias="PatientAge")
+    patient_age: Optional[NonNegativeFloat] = Field(alias="PatientAge")
     creation_date: PastDatetime = Field()
     pacs_id: NonNegativeInt
     patient_birth_date: Optional[PastDatetime] = Field(alias="PatientBirthDate")
@@ -160,7 +167,9 @@ class DicomSeriesPayload(BaseModel):
     hasura_id: str = Field(title="ID of event from Hasura")
 
     data: PacsFile = Field(title="The inserted DICOM file metadata")
-    match: Sequence[DicomSeriesMatcher] = Field(title="Which DICOM series to include")
+    match: Sequence[DicomSeriesMatcher] = Field(
+        title="Which DICOM series to include. Conditions are joined by AND."
+    )
     jobs: Sequence[ChrisRunnableRequest] = Field(
         title="Plugins or pipelines to run on the series data"
     )

src/serie/router.py

@@ -4,11 +4,13 @@
 from fastapi import Response, status, Header, APIRouter
 
 from serie.clients import BadAuthorizationError
+from serie.match import is_match
 from serie.models import DicomSeriesPayload, OxidicomCustomMetadata
 from serie.settings import get_settings
 from serie.actions import ClientActions
 
 router = APIRouter()
+"""The one and only router of SERIE!"""
 
 
 @router.post("/dicom_series/")
@@ -20,15 +22,18 @@ async def dicom_series(
     """
     Create *ChRIS* plugin instances and/or workflows on DICOM series data when an entire DICOM series is received.
     """
-    if (oxm_file := OxidicomCustomMetadata.from_pacsfile(payload.data)) is None:
+    settings = get_settings()
+    actions = ClientActions(auth=authorization, url=settings.chris_url)
+    if (series := await actions.resolve_series(payload.data)) is None:
+        response.status_code = status.HTTP_204_NO_CONTENT
+        return None
+    if not is_match(series, payload.match):
         response.status_code = status.HTTP_204_NO_CONTENT
         return None
 
-    settings = get_settings()
-    actions = ClientActions(auth=authorization, url=settings.chris_url)
     try:
         feed = await actions.create_analysis(
-            oxm_file, payload.jobs, payload.feed_name_template
+            series, payload.jobs, payload.feed_name_template
         )
     except BadAuthorizationError as e:
         response.status_code = status.HTTP_401_UNAUTHORIZED

src/serie/series_file_pair.py

@@ -0,0 +1,31 @@
+import dataclasses
+
+import aiochris.models
+
+from serie.dicom_series_metadata import DicomSeriesMetadata, DicomSeriesMetadataName
+from serie.models import OxidicomCustomMetadata
+
+
+@dataclasses.dataclass(frozen=True)
+class DicomSeriesFilePair:
+    """
+    A product type which can provide all the fields listed in :class:`DicomSeriesMetadataName`.
+    """
+
+    ocm: OxidicomCustomMetadata
+    pacs_file: aiochris.models.PACSFile
+
+    @property
+    def series_dir(self) -> str:
+        return self.ocm.series_dir
+
+    def to_dict(self) -> DicomSeriesMetadata:
+        """
+        Create a dict with all the keys from the variants of :class:`DicomSeriesMetadataName`.
+        """
+        values = {}
+        for name in DicomSeriesMetadataName:
+            values[name.value] = getattr(self.pacs_file, name.value, None) or getattr(
+                self.ocm, name.value
+            )
+        return values

tests/e2e_config.py

@@ -1,7 +1,7 @@
 EXAMPLE_DOWNLOAD_URL = "https://cube.chrisproject.org/api/v1/files/28617/None-1.2.276.0.7230010.3.1.4.3915146910.26736.1617073093.545.dcm"
 OXIDICOM_HOST = "oxidicom"
 OXIDICOM_PORT = 11111
-AE_TITLE = "SERIETEST"
+AE_TITLE = "SERIETESTE2E"
 OXIDICOM_AET = "ChRIS"
 CUBE_CONTAINER_ID = "chris"
 CHRIS_USERNAME = "serie"

tests/helpers.py

@@ -0,0 +1,52 @@
+import os
+from pathlib import Path
+
+import pydicom
+import pynetdicom
+import pytest
+import requests
+
+import tests.e2e_config as config
+
+
+def download_and_send_dicom(url: str, ae_title: str):
+    _send_dicom(_get_sample_dicom(url), ae_title)
+
+
+def _get_sample_dicom(url: str) -> Path:
+    tmp_dicom = Path(__file__).parent.parent / ".test_data" / _basename(url)
+    if tmp_dicom.exists():
+        return tmp_dicom
+    tmp_dicom.parent.mkdir(parents=True, exist_ok=True)
+    with requests.get(url, stream=True) as r:
+        r.raise_for_status()
+        with tmp_dicom.open("wb") as f:
+            for chunk in r.iter_content(chunk_size=8192):
+                f.write(chunk)
+    return tmp_dicom
+
+
+def _send_dicom(dicom_file: str | os.PathLike, ae_title: str):
+    """
+    See https://pydicom.github.io/pynetdicom/stable/examples/storage.html#storage-scu
+    """
+    ds = pydicom.dcmread(dicom_file)
+    ae = pynetdicom.AE(ae_title=ae_title)
+    ae.add_requested_context(ds.file_meta.MediaStorageSOPClassUID)
+    assoc = ae.associate(
+        config.OXIDICOM_HOST, config.OXIDICOM_PORT, ae_title=config.OXIDICOM_AET
+    )
+    if not assoc.is_established:
+        raise pytest.fail("Could not establish association with oxidicom")
+    status = assoc.send_c_store(ds)
+    if not status:
+        raise pytest.fail(
+            "Failed to send data to oxidicom: "
+            "connection timed out, was aborted or received invalid response"
+        )
+    assoc.release()
+
+
+def _basename(s: str) -> str:
+    split = s.rsplit("/", maxsplit=1)
+    return s if len(split) == 1 else split[1]

tests/test_e2e.py

@@ -76,7 +76,9 @@ async def _start_test(chris: ChrisClient):
     await asyncify(clear_dicoms_from_cube)(config.AE_TITLE)
     assert await chris.search_pacsfiles(pacs_identifier=config.AE_TITLE).count() == 0
 
-    await asyncify(download_and_send_dicom)(config.EXAMPLE_DOWNLOAD_URL, config.AE_TITLE)
+    await asyncify(download_and_send_dicom)(
+        config.EXAMPLE_DOWNLOAD_URL, config.AE_TITLE
+    )
 
     elapsed = 0
     while await chris.search_pacsfiles(pacs_identifier=config.AE_TITLE).count() == 0: