Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion docker/orthanc/Dockerfile
Original file line number Diff line number Diff line change
Expand Up @@ -32,7 +32,7 @@ FROM pixl_orthanc_uv AS pixl_orthanc_with_spec
# Do it in dead end build stage to discard this environment afterwards,
# and because the spec is only needed in orthanc-anon.
RUN uv venv
RUN uv pip install dicom-validator==0.7.3
RUN uv pip install dicom-validator==0.9.0
COPY ./orthanc/orthanc-anon/plugin/download_dicom_spec.py /etc/orthanc/download_dicom_spec.py
RUN --mount=type=cache,target=/root/.cache,id=dlspec \
python3 /etc/orthanc/download_dicom_spec.py
Expand Down
4 changes: 1 addition & 3 deletions orthanc/orthanc-anon/plugin/download_dicom_spec.py
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,4 @@
edition = "2024e"
download_path = str(Path.home() / "dicom-validator")
edition_reader = EditionReader(download_path)
destination = edition_reader.get_revision(edition, recreate_json=False)
json_path = Path(destination, "json")
EditionReader.load_dicom_info(json_path)
edition_reader.get_edition_path(edition)
2 changes: 1 addition & 1 deletion pixl_dcmd/pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,7 @@ dependencies = [
"arrow==1.4.0",
"deid==0.4.12",
"dicom-anonymizer==2.0.0",
"dicom-validator==0.7.3",
"dicom-validator==0.9.0",
"logger==1.4",
"pydicom==3.0.2",
"pydicom-data",
Expand Down
128 changes: 77 additions & 51 deletions pixl_dcmd/src/pixl_dcmd/dicom_helpers.py
Original file line number Diff line number Diff line change
Expand Up @@ -19,15 +19,26 @@
import typing
from contextlib import contextmanager, redirect_stdout
from dataclasses import dataclass
import logging
from io import StringIO
from pathlib import Path
from typing import Generator

from loguru import logger

from dicom_validator.spec_reader.edition_reader import EditionReader
from dicom_validator.tag_tools import tag_name_from_id
from dicom_validator.validator.error_handler import (
NullValidationResultHandler,
ValidationResultFormatter,
)
from dicom_validator.validator.iod_validator import IODValidator
from dicom_validator.validator.validation_result import (
DicomTag,
ModuleErrors,
Status,
TagError,
ValidationResult,
)
from pydicom import Dataset

from core.exceptions import PixlSkipInstanceError
Expand All @@ -44,64 +55,71 @@ def __init__(self, edition: str = "current"):
standard_path = str(Path.home() / "dicom-validator")
with _redirect_stdout_to_debug(logger):
edition_reader = EditionReader(standard_path)
destination = edition_reader.get_revision(self.edition, False)
json_path = Path(destination, "json")
self.dicom_info = EditionReader.load_dicom_info(json_path)
self.dicom_info = edition_reader.dicom_info_for_edition(self.edition)

def validate_original(self, dataset: Dataset) -> dict | None:
# Used to format errors introduced by de-identification
self.formatter = ValidationResultFormatter(self.dicom_info.dictionary)

def _validate(self, dataset: Dataset) -> ValidationResult:
"""Validate a pydicom Dataset using dicom-validator."""
return IODValidator(
dataset,
self.dicom_info,
error_handler=NullValidationResultHandler(),
).validate()

def _describe_error(self, tag: DicomTag, error: TagError) -> str:
tag_name = tag_name_from_id(tag.tag, self.dicom_info.dictionary)
return f"Tag {tag_name}{self.formatter.error_message(error)}"

def validate_original(self, dataset: Dataset) -> ModuleErrors | None:
"""Check pre-existing validation errors in a dataset.

Returns:
validation_errors: a dictionary of validation errors, or None
if dicom-validator raised a RuntimeError during validation.
module_errors: pre-existing validation errors, keyed by module
name then DICOM tag, or None if dicom-validator could not
validate the dataset at all (e.g. missing or unrecognised
SOP Class UID).
"""
validator = IODValidator(
dataset,
self.dicom_info,
log_level=logging.ERROR,
)
try:
errors: dict | None = validator.validate()
except RuntimeError as error:
result = self._validate(dataset)
if result.status not in (Status.Passed, Status.Failed):
logger.warning(
"Cannot check for pre-existing validation errors. "
"dicom-validator raised a RuntimeError during validation: {}",
error,
"dicom-validator returned status: {}",
result.status,
)
errors = None
return None

return errors
return result.module_errors

def validate_anonymised(
self, dataset: Dataset, original_errors: dict | None
) -> dict:
self, dataset: Dataset, original_errors: ModuleErrors | None
) -> dict[str, set[str]]:
"""Check validation errors introduced during de-identification.

Args:
original_errors: dict of errors returned by validate_original for
dataset before anonymisation, or None if the dataset hasn't
been validated for pre-existing errors.
original_errors: module_errors returned by validate_original for
the dataset before anonymisation, or None if the dataset
hasn't been validated for pre-existing errors.

Returns:
new_errors: dict of errors introduced by anonymisation. If
original_errors is None, all errors found after
anonymisation are returned, as it's not possible to tell
which of them pre-existed.
new_errors: human-readable errors introduced by anonymisation,
keyed by module name. If original_errors is None, all errors
found after anonymisation are returned, as it's not possible
to tell which of them pre-existed.

Raises:
PixlSkipInstanceError: If dicom-validator raises a RuntimeError
during validation.
PixlSkipInstanceError: If dicom-validator could not validate the
anonymised dataset at all (e.g. missing SOP Class UID).
"""
validator = IODValidator(
dataset,
self.dicom_info,
log_level=logging.ERROR,
)
try:
anon_errors: dict = validator.validate()
except RuntimeError as error:
msg = f"dicom-validator raised a RuntimeError when validating the anonymised dataset: {error}"
raise PixlSkipInstanceError(msg) from error
result = self._validate(dataset)
if result.status not in (Status.Passed, Status.Failed):
msg = (
"Cannot validate the anonymised dataset. "
f"dicom-validator returned status: {result.status}"
)
raise PixlSkipInstanceError(msg)
anon_errors = result.module_errors

if original_errors is None:
logger.warning(
Expand All @@ -110,18 +128,26 @@ def validate_anonymised(
"Errors found after anonymisation: {}",
anon_errors,
)
return anon_errors

diff_errors: dict = {}
for key in anon_errors:
if key in original_errors:
# Keep only errors introduced after the anonymisation
# The keys of the dictionary containt the actual errors
diff = set(anon_errors[key]) - set(original_errors[key])
if diff:
diff_errors[key] = diff
original_errors = ModuleErrors()

diff_errors: dict[str, set[str]] = {}
for module_name, anon_tag_errors in anon_errors.items():
if module_name in original_errors:
# keep tags with new errors or errors that have changed
original_tag_errors = original_errors[module_name]
new_tag_errors = {
tag: error
for tag, error in anon_tag_errors.items()
if (tag, error) not in original_tag_errors.items()
}
else:
diff_errors[key] = anon_errors[key]
new_tag_errors = anon_tag_errors

if new_tag_errors:
diff_errors[module_name] = {
self._describe_error(tag, error)
for tag, error in new_tag_errors.items()
}

return diff_errors

Expand Down
65 changes: 39 additions & 26 deletions pixl_dcmd/tests/test_dicom_validator.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@

import pytest
from core.exceptions import PixlSkipInstanceError
from dicom_validator.validator.validation_result import ErrorCode
from pixl_dcmd.dicom_helpers import DicomValidator
from pixl_dcmd.main import anonymise_dicom
from pydicom import Dataset
Expand Down Expand Up @@ -85,10 +86,7 @@ def test_validation_fails_after_invalid_tag_modification(
assert len(validation_result) == 1
assert "Patient" in validation_result.keys()
assert len(validation_result["Patient"]) == 1
assert (
"Tag (0010,0010) (Patient's Name) is missing"
in validation_result["Patient"].keys()
)
assert "Tag (0010,0010) (Patient's Name) is missing" in validation_result["Patient"]


@pytest.fixture()
Expand All @@ -103,54 +101,69 @@ def dicom_with_malformed_sequence_tag(vanilla_dicom_image_DX: Dataset) -> Datase
return vanilla_dicom_image_DX


def test_validate_original_survives_runtime_error(
def test_validate_original_reports_malformed_sequence(
dicom_with_malformed_sequence_tag: Dataset,
) -> None:
"""
GIVEN a DICOM dataset that makes dicom-validator raise a RuntimeError
GIVEN a DICOM dataset with a malformed sequence tag
WHEN the original dataset is validated
THEN None is returned rather than a dictionary of pre-existing errors
THEN an InvalidSequence error is returned
"""
validator = DicomValidator()
original_errors = validator.validate_original(dicom_with_malformed_sequence_tag)
assert original_errors is None

error_codes = {
error.code
for tag_errors in original_errors.values()
for error in tag_errors.values()
}
assert ErrorCode.InvalidSequence in error_codes


def test_validate_anonymised_returns_all_errors_when_original_unknown(
dicom_with_malformed_sequence_tag: Dataset,
vanilla_dicom_image_DX: Dataset,
) -> None:
"""
GIVEN an anonymised dataset that has not been validated for pre-existing errors
WHEN the anonymised dataset is validated
THEN all errors found are returned
"""
validator = DicomValidator()
original_errors = validator.validate_original(dicom_with_malformed_sequence_tag)
assert original_errors is None

# delete problematic element
del dicom_with_malformed_sequence_tag.DerivationCodeSequence
# delete a required element to introduce an error
del dicom_with_malformed_sequence_tag.PatientName
del vanilla_dicom_image_DX.PatientName

validation_result = validator.validate_anonymised(
dicom_with_malformed_sequence_tag, original_errors
)
validation_result = validator.validate_anonymised(vanilla_dicom_image_DX, None)
assert "Patient" in validation_result.keys()


def test_validate_anonymised_raises_skip_instance_error_on_runtime_error(
dicom_with_malformed_sequence_tag: Dataset,
@pytest.fixture()
def dicom_missing_sop_class_uid(vanilla_dicom_image_DX: Dataset) -> Dataset:
"""A DICOM dataset with no SOP Class UID, which dicom-validator cannot validate."""
del vanilla_dicom_image_DX.SOPClassUID
return vanilla_dicom_image_DX


def test_validate_original_returns_none_when_dataset_cannot_be_validated(
dicom_missing_sop_class_uid: Dataset,
) -> None:
"""
GIVEN a DICOM dataset that dicom-validator cannot validate at all
WHEN the original dataset is validated
THEN None is returned
"""
validator = DicomValidator()
original_errors = validator.validate_original(dicom_missing_sop_class_uid)
assert original_errors is None


def test_validate_anonymised_raises_skip_instance_error_when_dataset_cannot_be_validated(
dicom_missing_sop_class_uid: Dataset,
) -> None:
"""
GIVEN an anonymised dataset that causes dicom-validator to raise a RuntimeError
GIVEN an anonymised DICOM dataset that dicom-validator cannot validate at all
WHEN the anonymised dataset is validated
THEN a PixlSkipInstanceError is raised
"""
validator = DicomValidator()
original_errors = validator.validate_original(dicom_with_malformed_sequence_tag)

with pytest.raises(PixlSkipInstanceError):
validator.validate_anonymised(
dicom_with_malformed_sequence_tag, original_errors
)
validator.validate_anonymised(dicom_missing_sop_class_uid, None)
8 changes: 4 additions & 4 deletions uv.lock

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

Loading