Add tests for error handler

- add error handler example
- add some rudimentary documentation
- minor refactoring

Author: mrbean-bremen

README.md

@@ -147,6 +147,43 @@ Support for more cases may be added in the future.
 Defined terms are _not_ checked, because they are allowed to be user-defined, which means
 that any value may be valid.
 
+### Using the IOD validator programmatically
+The commandline tool is based on the `DicomFileValidator` and `IODValidator` classes, which
+can also be used directly in another application, if you want to have better control of the output.
+To initialize such a class, you need the DICOM information (as a `DicomInfo` instance)
+read from the JSON files as described above.
+First, the needed edition of the DICOM standard has to be downloaded and processed.
+This needs to be done only once for a given edition:
+
+```py
+from pathlib import Path
+from dicom_validator.spec_reader.edition_reader import EditionReader
+
+reader = EditionReader()  # uses the default location <user home>/dicom-validator
+# if the edition is not downloaded yet, load_dicom_info will download and process it
+dicom_info = reader.load_dicom_info("2025b")  # could also use "current" for the current standard
+```
+Using this DICOM information, you can validate a DICOM file, or all DICOM files
+in a given directory:
+```py
+from dicom_validator.validator.dicom_file_validator import DicomFileValidator
+
+validator = DicomFileValidator(dicom_info)
+result = validator.validate(dicom_file_path)
+```
+This will create the same output as the command line tool. If you need your own error
+handling instead, you can write your own error handler, and use it instead:
+```py
+from dicom_validator.validator.html_error_handler import HtmlErrorHandler
+...
+handler = HtmlErrorHandler(dicom_info)
+validator = DicomFileValidator(dicom_info, error_handler=handler)
+validator.validate(dicom_file_path)
+with open("result.html", "w") as f:
+    f.write(handler.html)
+```
+This uses the `HtmlErrorHandler` class, which is included as a simple error handler example.
+
 ## dump_dcm_info
 
 This is a very simple DICOM dump tool, which uses

dicom_validator/spec_reader/edition_reader.py

@@ -51,8 +51,11 @@ class EditionReader:
     dict_info_json = "dict_info.json"
     uid_info_json = "uid_info.json"
 
-    def __init__(self, path: str | Path) -> None:
-        self.path = Path(path)
+    def __init__(self, path: str | Path | None = None) -> None:
+        if path is not None:
+            self.path = Path(path)
+        else:
+            self.path = Path.home() / "dicom-validator"
         self.logger = logging.getLogger()
         if not self.logger.hasHandlers():
             self.logger.addHandler(logging.StreamHandler(sys.stdout))

dicom_validator/tests/validator/conftest.py

@@ -2,10 +2,24 @@
 import logging
 from pathlib import Path
 
+import pydicom
 import pytest
+from pydicom import (
+    dcmread,
+    dcmwrite,
+    FileMetaDataset,
+    DataElement,
+    Sequence,
+    Dataset,
+    uid,
+)
+from pydicom.datadict import dictionary_VR
+from pydicom.filebase import DicomBytesIO
+from pydicom.tag import Tag
 
 from dicom_validator.spec_reader.edition_reader import EditionReader
 from dicom_validator.validator.dicom_info import DicomInfo
+from dicom_validator.validator.iod_validator import IODValidator
 
 CURRENT_EDITION = "2025d"
 
@@ -70,3 +84,52 @@ def disable_logging():
     logging.disable(logging.CRITICAL)
     yield
     logging.disable(logging.DEBUG)
+
+
+def new_data_set(tags, *, top_level: bool = True):
+    """Create a DICOM data set with the given attributes"""
+    tags = tags or {}
+    data_set = Dataset()
+    if not tags:
+        return data_set
+    for tag, value in tags.items():
+        tag = Tag(tag)  # raises for invalid tag
+        try:
+            vr = dictionary_VR(tag)
+        except KeyError:
+            vr = "LO"
+        if vr == "SQ":
+            items = []
+            for item_tags in value:
+                items.append(new_data_set(item_tags, top_level=False))
+            value = Sequence(items)
+        data_set[tag] = DataElement(tag, vr, value)
+    if not top_level:
+        # this is a sequence item
+        return data_set
+
+    # write the dataset into a file and read it back to ensure the real behavior
+    if "SOPInstanceUID" not in data_set:
+        data_set.SOPInstanceUID = "1.2.3"
+    data_set.file_meta = FileMetaDataset()
+    data_set.file_meta.TransferSyntaxUID = uid.ExplicitVRLittleEndian
+    fp = DicomBytesIO()
+    kwargs = (
+        {"write_like_original": False}
+        if int(pydicom.__version_info__[0]) < 3
+        else {"enforce_file_format": True}
+    )
+    dcmwrite(fp, data_set, **kwargs)
+    fp.seek(0)
+    return dcmread(fp)
+
+
+@pytest.fixture
+def validator(dicom_info, request):
+    marker = request.node.get_closest_marker("tag_set")
+    if marker is None:
+        tag_set = {}
+    else:
+        tag_set = marker.args[0]
+    data_set = new_data_set(tag_set)
+    return IODValidator(data_set, dicom_info, log_level=logging.WARNING)

dicom_validator/tests/validator/test_error_handler.py

@@ -0,0 +1,62 @@
+from pathlib import Path
+
+import pytest
+from pydicom import uid
+
+from dicom_validator.validator.dicom_file_validator import DicomFileValidator
+from dicom_validator.validator.html_error_handler import HtmlErrorHandler
+from dicom_validator.validator.error_handler import (
+    ValidationResultHandler,
+)
+from dicom_validator.validator.validation_result import (
+    ValidationResult,
+)
+
+
+class GenericErrorHandler(ValidationResultHandler):
+    def __init__(self):
+        self.logs = []
+
+    def handle_validation_start(self, result: ValidationResult):
+        self.logs.append("Starting Validation")
+
+    def handle_validation_result(self, result: ValidationResult):
+        self.logs.append("Finished Validation")
+        self.logs.append(f"Status: {result.status.name}")
+        self.logs.append(f"Error: {result.errors}")
+
+
+@pytest.mark.tag_set(
+    {
+        "SOPClassUID": uid.CTImageStorage,
+        "PatientName": "XXX",
+        "PatientID": "ZZZ",
+    }
+)
+def test_generic_error_handler(validator) -> None:
+    handler = GenericErrorHandler()
+    validator.handler = handler
+    result = validator.validate()
+    nr_errors = result.errors
+    assert handler.logs == [
+        "Starting Validation",
+        "Finished Validation",
+        "Status: Failed",
+        f"Error: {nr_errors}",
+    ]
+
+
+@pytest.fixture(scope="module")
+def dicom_fixture_path():
+    yield Path(__file__).parent.parent / "fixtures" / "dicom"
+
+
+def test_html_error_handler(dicom_info, dicom_fixture_path) -> None:
+    rtdose_path = dicom_fixture_path / "rtdose.dcm"
+    handler = HtmlErrorHandler(dicom_info)
+    validator = DicomFileValidator(dicom_info, error_handler=handler)
+    validator.validate(rtdose_path)
+    assert (
+        '<h3><a href="https://dicom.nema.org/medical/dicom/current/output/chtml/part03'
+        '/sect_C.8.8.html">RT Series</a></h3>' in handler.html
+    )

dicom_validator/tests/validator/test_iod_validator.py

@@ -1,69 +1,14 @@
 import logging
 
-import pydicom
 import pytest
-from pydicom import DataElement, uid, Sequence, dcmwrite, dcmread
-from pydicom.datadict import dictionary_VR
-from pydicom.dataset import Dataset, FileMetaDataset
-from pydicom.filebase import DicomBytesIO
-from pydicom.tag import Tag
+from pydicom import uid
 
 from dicom_validator.tests.utils import has_tag_error
-from dicom_validator.validator.iod_validator import IODValidator
 from dicom_validator.validator.validation_result import Status, ErrorCode
 
 pytestmark = pytest.mark.usefixtures("disable_logging")
 
 
-def new_data_set(tags, *, top_level: bool = True):
-    """Create a DICOM data set with the given attributes"""
-    tags = tags or {}
-    data_set = Dataset()
-    if not tags:
-        return data_set
-    for tag, value in tags.items():
-        tag = Tag(tag)  # raises for invalid tag
-        try:
-            vr = dictionary_VR(tag)
-        except KeyError:
-            vr = "LO"
-        if vr == "SQ":
-            items = []
-            for item_tags in value:
-                items.append(new_data_set(item_tags, top_level=False))
-            value = Sequence(items)
-        data_set[tag] = DataElement(tag, vr, value)
-    if not top_level:
-        # this is a sequence item
-        return data_set
-
-    # write the dataset into a file and read it back to ensure the real behavior
-    if "SOPInstanceUID" not in data_set:
-        data_set.SOPInstanceUID = "1.2.3"
-    data_set.file_meta = FileMetaDataset()
-    data_set.file_meta.TransferSyntaxUID = pydicom.uid.ExplicitVRLittleEndian
-    fp = DicomBytesIO()
-    kwargs = (
-        {"write_like_original": False}
-        if int(pydicom.__version_info__[0]) < 3
-        else {"enforce_file_format": True}
-    )
-    dcmwrite(fp, data_set, **kwargs)
-    fp.seek(0)
-    return dcmread(fp)
-
-
-@pytest.fixture
-def validator(dicom_info, request):
-    marker = request.node.get_closest_marker("tag_set")
-    if marker is None:
-        tag_set = {}
-    else:
-        tag_set = marker.args[0]
-    data_set = new_data_set(tag_set)
-    return IODValidator(data_set, dicom_info, logging.WARNING)
-
-
 class TestIODValidator:
     """Tests IODValidator.
     Note: some of the fixture data are not consistent with the DICOM Standard.

dicom_validator/tests/validator/test_iod_validator_func_groups.py

@@ -83,7 +83,7 @@ def validator(dicom_info, request):
     marker = request.node.get_closest_marker("per_frame_macros")
     per_frame_macros = {} if marker is None else marker.args[0]
     data_set = new_data_set(shared_macros, per_frame_macros)
-    return IODValidator(data_set, dicom_info, logging.WARNING)
+    return IODValidator(data_set, dicom_info, log_level=logging.WARNING)
 
 
 FRAME_ANATOMY = {
@@ -128,7 +128,7 @@ def test_missing_func_groups(self, dicom_info: DicomInfo, caplog):
         data_set = new_data_set({}, {})
         del data_set[0x52009229]
         del data_set[0x52009230]
-        validator = IODValidator(data_set, dicom_info, logging.WARNING)
+        validator = IODValidator(data_set, dicom_info, log_level=logging.WARNING)
         result = validator.validate()
         group_result = self.ensure_group_result(result)
         assert DicomTag(0x5200_9229) in group_result

dicom_validator/validator/dicom_file_validator.py

@@ -1,18 +1,18 @@
 import logging
 import os
-import sys
 from os import PathLike
 
 from pydicom import config, dcmread
 from pydicom.errors import InvalidDicomError
 
+from dicom_validator.validator.dicom_info import DicomInfo
 from dicom_validator.validator.error_handler import ValidationResultHandler
 from dicom_validator.validator.iod_validator import IODValidator
-from dicom_validator.validator.dicom_info import DicomInfo
 from dicom_validator.validator.validation_result import ValidationResult, Status
 
 
 class DicomFileValidator:
+    """Validates a single DICOM file or all DICOM files in a directory."""
 
     def __init__(
         self,
@@ -22,37 +22,51 @@ def __init__(
         suppress_vr_warnings: bool = False,
         error_handler: ValidationResultHandler | None = None,
     ) -> None:
+        """Initializes a DicomFileValidator object.
+
+        Parameters
+        ----------
+        dicom_info : DicomInfo
+            DICOM information as read from the JSON files created from the standard
+        log_level : int
+            The log level to use with the default error handler
+        force_read: bool
+            If `True`, the DICOM file is tried to read even if it may not be valid
+        suppress_vr_warnings: bool
+            By default, values not valid for the given VR are reported, using the
+            `pydicom` validation. If set to `True`, this validation is not performed.
+        error_handler: ValidationResultHandler or None
+            If set, this handler will be used to handle the validation result,
+            otherwise the default handler is used that logs errors to the console.
+        """
         self._dicom_info = dicom_info
-        self.logger = logging.getLogger()
-        self.logger.level = log_level
-        if not self.logger.hasHandlers():
-            self.logger.addHandler(logging.StreamHandler(sys.stdout))
+        self.log_level = log_level
         self._force_read = force_read
         self._suppress_vr_warnings = suppress_vr_warnings
         self._error_handler = error_handler
 
     def validate(self, path: str | PathLike) -> dict[str, ValidationResult]:
-        errors: dict[str, ValidationResult] = {}
+        results: dict[str, ValidationResult] = {}
         path = os.fspath(path)
         if not os.path.exists(path):
-            errors[path] = ValidationResult(status=Status.MissingFile, errors=1)
-            self.logger.warning('\n"%s" does not exist - skipping', path)
+            results[path] = ValidationResult(
+                file_path=path, status=Status.MissingFile, errors=1
+            )
         else:
             if os.path.isdir(path):
-                errors.update(self.validate_dir(path))
+                results.update(self.validate_dir(path))
             else:
-                errors.update(self.validate_file(path))
-        return errors
+                results.update(self.validate_file(path))
+        return results
 
     def validate_dir(self, dir_path: str) -> dict[str, ValidationResult]:
-        errors: dict[str, ValidationResult] = {}
+        results: dict[str, ValidationResult] = {}
         for root, _, names in os.walk(dir_path):
             for name in names:
-                errors.update(self.validate(os.path.join(root, name)))
-        return errors
+                results.update(self.validate(os.path.join(root, name)))
+        return results
 
     def validate_file(self, file_path: str) -> dict[str, ValidationResult]:
-        self.logger.info('\nProcessing DICOM file "%s"', file_path)
         try:
             # dcmread calls validate_value by default. If values don't match
             # required VR (value representation), it emits a warning but
@@ -63,14 +77,18 @@ def validate_file(self, file_path: str) -> dict[str, ValidationResult]:
             data_set = dcmread(file_path, defer_size=1024, force=self._force_read)
 
         except InvalidDicomError:
-            self.logger.error(f"Invalid DICOM file: {file_path}")
-            return {file_path: ValidationResult(status=Status.InvalidFile, errors=1)}
+            return {
+                file_path: ValidationResult(
+                    file_path=file_path, status=Status.InvalidFile, errors=1
+                )
+            }
         return {
             file_path: IODValidator(
                 data_set,
                 self._dicom_info,
-                self.logger.level,
+                log_level=self.log_level,
                 suppress_vr_warnings=self._suppress_vr_warnings,
                 error_handler=self._error_handler,
+                file_path=file_path,
             ).validate()
         }