Adding pseudonym to wador views

adit-client/adit_client/client.py

@@ -1,22 +1,8 @@
 import importlib.metadata
 from typing import Iterator
 
-from dicognito.anonymizer import Anonymizer
-from dicognito.value_keeper import ValueKeeper
-from dicomweb_client import DICOMwebClient, session_utils
 from pydicom import Dataset
-
-DEFAULT_SKIP_ELEMENTS_ANONYMIZATION = [
-    "AcquisitionDate",
-    "AcquisitionDateTime",
-    "AcquisitionTime",
-    "ContentDate",
-    "ContentTime",
-    "SeriesDate",
-    "SeriesTime",
-    "StudyDate",
-    "StudyTime",
-]
+from pydicomweb_client import DICOMwebClient, session_utils
 
 
 class AditClient:
@@ -27,7 +13,6 @@ def __init__(
         verify: str | bool = True,
         trial_protocol_id: str | None = None,
         trial_protocol_name: str | None = None,
-        skip_elements_anonymization: list[str] | None = None,
     ) -> None:
         self.server_url = server_url
         self.auth_token = auth_token
@@ -36,11 +21,6 @@ def __init__(
         self.trial_protocol_name = trial_protocol_name
         self.__version__ = importlib.metadata.version("adit-client")
 
-        if skip_elements_anonymization is None:
-            self.skip_elements_anonymization = DEFAULT_SKIP_ELEMENTS_ANONYMIZATION
-        else:
-            self.skip_elements_anonymization = skip_elements_anonymization
-
     def search_for_studies(
         self, ae_title: str, query: dict[str, str] | None = None
     ) -> list[Dataset]:
@@ -76,24 +56,16 @@ def retrieve_study(
         """Retrieve all instances of a study."""
         images = self._create_dicom_web_client(ae_title).retrieve_study(study_uid)
 
-        anonymizer: Anonymizer | None = None
-        if pseudonym is not None:
-            anonymizer = self._setup_anonymizer()
-
-        return [self._handle_dataset(image, anonymizer, pseudonym) for image in images]
+        return [self._handle_dataset(image, pseudonym) for image in images]
 
     def iter_study(
         self, ae_title: str, study_uid: str, pseudonym: str | None = None
     ) -> Iterator[Dataset]:
         """Iterate over all instances of a study."""
         images = self._create_dicom_web_client(ae_title).iter_study(study_uid)
 
-        anonymizer: Anonymizer | None = None
-        if pseudonym is not None:
-            anonymizer = self._setup_anonymizer()
-
         for image in images:
-            yield self._handle_dataset(image, anonymizer, pseudonym)
+            yield self._handle_dataset(image, pseudonym)
 
     def retrieve_series(
         self,
@@ -107,11 +79,7 @@ def retrieve_series(
             study_uid, series_instance_uid=series_uid
         )
 
-        anonymizer: Anonymizer | None = None
-        if pseudonym is not None:
-            anonymizer = self._setup_anonymizer()
-
-        return [self._handle_dataset(image, anonymizer, pseudonym) for image in images]
+        return [self._handle_dataset(image, pseudonym) for image in images]
 
     def iter_series(
         self,
@@ -125,12 +93,8 @@ def iter_series(
             study_uid, series_instance_uid=series_uid
         )
 
-        anonymizer: Anonymizer | None = None
-        if pseudonym is not None:
-            anonymizer = self._setup_anonymizer()
-
         for image in images:
-            yield self._handle_dataset(image, anonymizer, pseudonym)
+            yield self._handle_dataset(image, pseudonym)
 
     def retrieve_image(
         self,
@@ -145,11 +109,7 @@ def retrieve_image(
             study_uid, series_uid, image_uid
         )
 
-        anonymizer: Anonymizer | None = None
-        if pseudonym is not None:
-            anonymizer = self._setup_anonymizer()
-
-        return self._handle_dataset(image, anonymizer, pseudonym)
+        return self._handle_dataset(image, pseudonym)
 
     def store_images(self, ae_title: str, images: list[Dataset]) -> Dataset:
         """Store images."""
@@ -175,15 +135,7 @@ def _create_dicom_web_client(self, ae_title: str) -> DICOMwebClient:
             },
         )
 
-    def _setup_anonymizer(self) -> Anonymizer:
-        anonymizer = Anonymizer()
-        for element in self.skip_elements_anonymization:
-            anonymizer.add_element_handler(ValueKeeper(element))
-        return anonymizer
-
-    def _handle_dataset(
-        self, ds: Dataset, anonymizer: Anonymizer | None, pseudonym: str | None
-    ) -> Dataset:
+    def _handle_dataset(self, ds: Dataset, pseudonym: str | None) -> Dataset:
         # Similar to what ADIT does in core/processors.py
 
         if self.trial_protocol_id is not None:
@@ -192,12 +144,6 @@ def _handle_dataset(
         if self.trial_protocol_name is not None:
             ds.ClinicalTrialProtocolName = self.trial_protocol_name
 
-        if pseudonym is not None:
-            assert anonymizer is not None
-            anonymizer.anonymize(ds)
-            ds.PatientID = pseudonym
-            ds.PatientName = pseudonym
-
         if pseudonym and self.trial_protocol_id:
             session_id = f"{ds.StudyDate}-{ds.StudyTime}"
             ds.PatientComments = (

adit/core/processors.py

@@ -8,11 +8,11 @@
 from pathlib import Path
 from typing import Callable
 
-from dicognito.anonymizer import Anonymizer
-from dicognito.value_keeper import ValueKeeper
 from django.conf import settings
 from pydicom import Dataset
 
+from adit.core.utils.pseudonymizer import Pseudonymizer
+
 from .errors import DicomError
 from .models import DicomAppSettings, DicomNode, DicomTask, TransferTask
 from .types import DicomLogEntry, ProcessingResult
@@ -186,12 +186,11 @@ def _download_to_folder(
         study_folder = patient_folder / f"{prefix}-{modalities}"
         os.makedirs(study_folder, exist_ok=True)
 
-        anonymizer = self._setup_anonymizer()
+        pseudonymizer = Pseudonymizer(pseudonym)
 
         modifier = partial(
             self._modify_dataset,
-            anonymizer,
-            pseudonym,
+            pseudonymizer,
         )
 
         if series_uids:
@@ -281,12 +280,6 @@ def _find_series_list(self, series_uid: str) -> list[ResultDataset]:
 
         return results
 
-    def _setup_anonymizer(self) -> Anonymizer:
-        anonymizer = Anonymizer()
-        for element in settings.SKIP_ELEMENTS_ANONYMIZATION:
-            anonymizer.add_element_handler(ValueKeeper(element))
-        return anonymizer
-
     def _download_study(
         self,
         patient_id: str,
@@ -379,17 +372,12 @@ def callback(ds: Dataset | None) -> None:
 
     def _modify_dataset(
         self,
-        anonymizer: Anonymizer,
-        pseudonym: str | None,
+        pseudonymizer: Pseudonymizer,
         ds: Dataset,
     ) -> None:
         """Optionally pseudonymize an incoming dataset with the given pseudonym
         and add the trial ID and name to the DICOM header if specified."""
-        if pseudonym:
-            anonymizer.anonymize(ds)
-
-            ds.PatientID = pseudonym
-            ds.PatientName = pseudonym
+        pseudonymizer.pseudonymize(ds)
 
         trial_protocol_id = self.transfer_task.job.trial_protocol_id
         trial_protocol_name = self.transfer_task.job.trial_protocol_name
@@ -400,6 +388,7 @@ def _modify_dataset(
         if trial_protocol_name:
             ds.ClinicalTrialProtocolName = trial_protocol_name
 
+        pseudonym = pseudonymizer.pseudonym
         if pseudonym and trial_protocol_id:
             session_id = f"{ds.StudyDate}-{ds.StudyTime}"
             ds.PatientComments = (

adit/core/utils/dicom_web_connector.py

@@ -6,9 +6,9 @@
 from pathlib import Path
 from typing import Callable, Iterator, NoReturn
 
-from dicomweb_client import DICOMwebClient
 from pydicom import Dataset
 from pydicom.errors import InvalidDicomError
+from pydicomweb_client import DICOMwebClient
 from requests import HTTPError
 
 from ..errors import DicomError, RetriableDicomError

adit/core/utils/pseudonymizer.py

@@ -0,0 +1,45 @@
+from dicognito.anonymizer import Anonymizer
+from dicognito.value_keeper import ValueKeeper
+from django.conf import settings
+from pydicom import Dataset
+
+
+class Pseudonymizer:
+    """
+    A utility class for pseudonymizing (or anonymizing) DICOM data.
+    """
+
+    def __init__(self, pseudonym: str | None = None) -> None:
+        """
+        Initialize the Pseudonymizer with an optional pseudonym.
+
+        :param pseudonym: The pseudonym to be used for the DICOM data.
+        """
+        self.anonymizer = self._setup_anonymizer()
+        self.pseudonym = pseudonym
+
+    def _setup_anonymizer(self) -> Anonymizer:
+        """
+        Set up the anonymizer instance and configure it to skip specific elements.
+
+        :return: An instance of the Anonymizer class.
+        """
+        anonymizer = Anonymizer()
+        for element in settings.SKIP_ELEMENTS_ANONYMIZATION:
+            anonymizer.add_element_handler(ValueKeeper(element))
+        return anonymizer
+
+    def pseudonymize(
+        self,
+        ds: Dataset,
+    ) -> None:
+        """
+        Pseudonymize the given DICOM dataset using the anonymizer and optional pseudonym.
+
+        :param ds: The DICOM dataset to be pseudonymized.
+        """
+        if self.pseudonym:
+            # Replace PatientID and PatientName with the provided pseudonym.
+            self.anonymizer.anonymize(ds)  # Apply anonymization to the dataset.
+            ds.PatientID = self.pseudonym
+            ds.PatientName = self.pseudonym

adit/core/utils/testing_helpers.py

@@ -7,11 +7,11 @@
 import pandas as pd
 from adit_radis_shared.accounts.factories import GroupFactory
 from adit_radis_shared.common.utils.testing_helpers import add_permission
-from dicomweb_client import DICOMwebClient
 from django.conf import settings
 from django.core.management import call_command
 from playwright.sync_api import FilePayload
 from pydicom import Dataset
+from pydicomweb_client import DICOMwebClient
 from pynetdicom.association import Association
 from pynetdicom.status import Status
 

adit/core/validators.py

@@ -68,3 +68,11 @@ def validate_series_number(value: str) -> None:
             raise ValueError()
     except ValueError:
         raise ValidationError(f"Invalid series number: {value}")
+
+
+def validate_pseudonym(pseudonym: str) -> None:
+    no_backslash_char_validator(pseudonym)
+    no_control_chars_validator(pseudonym)
+    no_wildcard_chars_validator(pseudonym)
+    if len(pseudonym) > 64:
+        raise ValidationError("Pseudonym string too long (max 64 characters).")