diff --git a/CHANGES.md b/CHANGES.md index 09254d9..2d461d8 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -3,8 +3,12 @@ The released versions correspond to PyPi releases. ## Unreleased +### Features +* added checking of enumerated values defined for a specific index + ### Fixes * fixed exception on parsing some older DICOM standard versions +* fixed checking of multi-valued tags with defined enum values (see [#87](../../issues/87)) ## [Version 0.5.0](https://pypi.python.org/pypi/dicom-validator/0.5.0) (2024-01-25) Adds enum checks and fixes a regression. diff --git a/README.md b/README.md index 579068a..fb91f4f 100644 --- a/README.md +++ b/README.md @@ -75,7 +75,7 @@ the first time. This checks a given DICOM file, or all DICOM files recursively in a given directory, for correct tags for the related SOP class. The presence or absence of the tag and the presence of a tag value are checked, and in the -case of an enumeration defined for the value, the value is also check for validity. +case that an enumeration is defined for the value, the value is also checked for validity. More checks may be added later. This is done by looking up all required and optional modules for this SOP class, and checking the tags for these modules. Tags that are not allowed or diff --git a/dicom_validator/spec_reader/edition_reader.py b/dicom_validator/spec_reader/edition_reader.py index b0dac68..caf52a0 100644 --- a/dicom_validator/spec_reader/edition_reader.py +++ b/dicom_validator/spec_reader/edition_reader.py @@ -75,19 +75,19 @@ def get_editions(self, update=True): # no need to update the edition dir more than once a month update = (today - modified_date).days > 30 else: - with open(editions_path) as f: + with open(editions_path, encoding="utf8") as f: update = not json.load(f) else: update = True if update: self.update_edition() if editions_path.exists(): - with open(editions_path) as json_file: + with open(editions_path, encoding="utf8") as json_file: return json.load(json_file) def read_from_html(self): html_path = self.path / self.html_filename - with open(html_path) as html_file: + with open(html_path, encoding="utf8") as html_file: contents = html_file.read() parser = EditionParser() parser.feed(contents) @@ -98,7 +98,7 @@ def write_to_json(self): editions = self.read_from_html() if editions: json_path = self.path / self.json_filename - with open(json_path, "w") as json_file: + with open(json_path, "w", encoding="utf8") as json_file: json_file.write(json.dumps(editions)) def get_edition(self, revision): @@ -163,7 +163,7 @@ def get_chapter(self, revision, chapter, destination, is_current): @staticmethod def load_info(json_path, info_json): - with open(json_path / info_json) as info_file: + with open(json_path / info_json, encoding="utf8") as info_file: return json.load(info_file) @classmethod @@ -204,13 +204,13 @@ def create_json_files(cls, docbook_path, json_path): if chapter in chapter_info: for uid in chapter_info[chapter]: definition[uid] = iod_info[chapter] - with open(json_path / cls.iod_info_json, "w") as info_file: + with open(json_path / cls.iod_info_json, "w", encoding="utf8") as info_file: info_file.write(cls.dump_description(definition)) - with open(json_path / cls.module_info_json, "w") as info_file: + with open(json_path / cls.module_info_json, "w", encoding="utf8") as info_file: info_file.write(cls.dump_description(part3reader.module_descriptions())) - with open(json_path / cls.dict_info_json, "w") as info_file: + with open(json_path / cls.dict_info_json, "w", encoding="utf8") as info_file: info_file.write(cls.dump_description(dict_info)) - with open(json_path / cls.uid_info_json, "w") as info_file: + with open(json_path / cls.uid_info_json, "w", encoding="utf8") as info_file: info_file.write(cls.dump_description(part6reader.all_uids())) cls.write_current_version(json_path) print("Done!") @@ -250,11 +250,11 @@ def is_current_version(json_path): version_path = json_path / "version" if not version_path.exists(): return False - with open(version_path) as f: + with open(version_path, encoding="utf8") as f: return f.read() >= __version__ @staticmethod def write_current_version(json_path): version_path = json_path / "version" - with open(version_path, "w") as f: + with open(version_path, "w", encoding="utf8") as f: f.write(__version__) diff --git a/dicom_validator/spec_reader/enum_parser.py b/dicom_validator/spec_reader/enum_parser.py index 40218df..a512e2a 100644 --- a/dicom_validator/spec_reader/enum_parser.py +++ b/dicom_validator/spec_reader/enum_parser.py @@ -1,4 +1,5 @@ -from typing import Callable, Optional, Dict +import re +from typing import Callable, Optional, Dict, List, Any from pydicom.valuerep import VR, INT_VR, STR_VR @@ -14,66 +15,75 @@ class EnumParser: """Parses enumerated values for a tag.""" docbook_ns = "{http://docbook.org/ns/docbook}" - nr_direct_enums = 0 - nr_linked_enums = 0 + enum_with_value_regex = re.compile(r"Enumerated Values for Value (\d):") def __init__( self, find_section: Callable[[str], Optional[ElementTree.Element]] ) -> None: self._find_section = find_section - self._enum_cache: Dict[str, ValuesType] = {} + self._enum_cache: Dict[str, Dict] = {} - def parse(self, node: ElementTree.Element, vr: VR) -> ValuesType: + def parse(self, node: ElementTree.Element, vr: VR) -> List[Dict]: """Searches for enumerated values in the tag description and in linked sections. Returns a list of the allowed values, or an empty list if none found. """ - var_list = node.find(self.docbook_ns + "variablelist") - if var_list is not None: - enums = self.parse_variable_list(var_list) - else: + var_lists = node.findall(self.docbook_ns + "variablelist") + enum_lists = [self.parse_variable_list(e) for e in var_lists] + enum_lists = [e for e in enum_lists if e] + if not enum_lists: enums = self.parse_linked_variablelist(node) - if enums: + if enums: + enum_lists.append(enums) + + if enum_lists: if vr == VR.AT: - # print( - # f"Ignoring enum values: " - # f"{', '.join([str(e) for e in enums])} with VR {vr}" - # ) return [] # this is included in INT_VRs, but won't work if vr in INT_VR: - int_enums: ValuesType = [] - for e in enums: - assert isinstance(e, str) - if e.endswith("H"): - int_enums.append(int(e[:-1], 16)) - else: - int_enums.append(int(e)) - self.__class__.nr_direct_enums += 1 - return int_enums + for enums in enum_lists: + int_enums: ValuesType = [] + for e in enums["val"]: + if isinstance(e, str): + if e.endswith("H"): + int_enums.append(int(e[:-1], 16)) + else: + int_enums.append(int(e)) + else: + int_enums.append(e) + enums["val"] = int_enums + return enum_lists if vr in STR_VR: - self.__class__.nr_direct_enums += 1 - return enums + return enum_lists # any other VR does not make sense here - # print( - # f"Ignoring enum values: " - # f"{', '.join([str(e) for e in enums])} with VR {vr}" - # ) return [] - def parse_variable_list(self, var_list) -> ValuesType: + def parse_variable_list(self, var_list) -> Dict: # we assume that a variablelist contains enumerated values or defined terms # we ignore defined terms, as they do not limit the possible values title = var_list.find(self.docbook_ns + "title") + if title is None: + return {} + value_match = self.enum_with_value_regex.match(title.text) + if value_match: + index = int(value_match.group(1)) + else: + index = 0 + if index == 0 and title.text.lower() != "enumerated values:": + return {} # TODO: handle cases with conditions - if title is None or title.text.lower() not in ("enumerated values:",): - return [] terms = [] for item in var_list.findall(self.docbook_ns + "varlistentry"): term = item.find(self.docbook_ns + "term") if term is not None: terms.append(term.text) - return terms + result: Dict[str, Any] = {} + if terms: + result["val"] = terms + if index > 0: + result["index"] = index - def parse_linked_variablelist(self, node) -> ValuesType: + return result + + def parse_linked_variablelist(self, node) -> Dict: for xref in node.findall(f"{self.docbook_ns}para/{self.docbook_ns}xref"): link = xref.attrib.get("linkend") if link and link.startswith("sect_"): @@ -84,8 +94,5 @@ def parse_linked_variablelist(self, node) -> ValuesType: var_list = section.find(f"{self.docbook_ns}variablelist") if var_list is not None: self._enum_cache[link] = self.parse_variable_list(var_list) - if self._enum_cache[link]: - self.__class__.nr_linked_enums += 1 - # print("LINK:", link, self._enum_cache[link]) return self._enum_cache[link] - return [] + return {} diff --git a/dicom_validator/tests/spec_reader/conftest.py b/dicom_validator/tests/spec_reader/conftest.py index 14d235e..b4bb7d2 100644 --- a/dicom_validator/tests/spec_reader/conftest.py +++ b/dicom_validator/tests/spec_reader/conftest.py @@ -47,7 +47,9 @@ def dict_info(revision_path): from dicom_validator.spec_reader.edition_reader import EditionReader json_fixture_path = revision_path / "json" - with open(json_fixture_path / EditionReader.dict_info_json) as info_file: + with open( + json_fixture_path / EditionReader.dict_info_json, encoding="utf8" + ) as info_file: info = json.load(info_file) yield info diff --git a/dicom_validator/tests/spec_reader/test_edition_reader.py b/dicom_validator/tests/spec_reader/test_edition_reader.py index 888ed43..476a533 100644 --- a/dicom_validator/tests/spec_reader/test_edition_reader.py +++ b/dicom_validator/tests/spec_reader/test_edition_reader.py @@ -23,7 +23,7 @@ def __init__(self, path, contents=""): self.html_contents = contents def retrieve(self, html_path): - with open(html_path, "w") as html_file: + with open(html_path, "w", encoding="utf8") as html_file: html_file.write(self.html_contents) diff --git a/dicom_validator/tests/spec_reader/test_enum_parser.py b/dicom_validator/tests/spec_reader/test_enum_parser.py index aaba4f1..4b16881 100644 --- a/dicom_validator/tests/spec_reader/test_enum_parser.py +++ b/dicom_validator/tests/spec_reader/test_enum_parser.py @@ -65,7 +65,7 @@ def test_single_enum(self, parser): NO """ - assert parser.parse(section(content), VR.SH) == ["NO"] + assert parser.parse(section(content), VR.SH) == [{"val": ["NO"]}] def test_single_enum_with_extra_tag(self, parser): content = """ @@ -77,7 +77,7 @@ def test_single_enum_with_extra_tag(self, parser): NO """ - assert parser.parse(section(content), VR.SH) == ["NO"] + assert parser.parse(section(content), VR.SH) == [{"val": ["NO"]}] def test_two_enums(self, parser): content = """ @@ -89,7 +89,7 @@ def test_two_enums(self, parser): NO """ - assert parser.parse(section(content), VR.SH) == ["YES", "NO"] + assert parser.parse(section(content), VR.SH) == [{"val": ["YES", "NO"]}] def test_int_enums(self, parser): content = """ @@ -101,7 +101,7 @@ def test_int_enums(self, parser): 0001 """ - assert parser.parse(section(content), VR.US) == [0, 1] + assert parser.parse(section(content), VR.US) == [{"val": [0, 1]}] def test_hex_enums(self, parser): content = """ @@ -113,7 +113,7 @@ def test_hex_enums(self, parser): 0011H """ - assert parser.parse(section(content), VR.US) == [16, 17] + assert parser.parse(section(content), VR.US) == [{"val": [16, 17]}] def test_linked_enum(self): content = """Bla blah, see @@ -139,4 +139,32 @@ def test_linked_enum(self): """ parser = EnumParser(lambda s: section(linked, s)) - assert parser.parse(section(content), VR.SH) == ["GEOMETRY", "FIDUCIAL"] + assert parser.parse(section(content), VR.SH) == [ + {"val": ["GEOMETRY", "FIDUCIAL"]} + ] + + def test_value_for_value(self, parser): + content = """ + + Enumerated Values for Value 1: + + DERIVED + + + + + + + Enumerated Values for Value 2: + + PRIMARY + + + + + + """ + assert parser.parse(section(content), VR.CS) == [ + {"index": 1, "val": ["DERIVED"]}, + {"index": 2, "val": ["PRIMARY"]}, + ] diff --git a/dicom_validator/tests/spec_reader/test_part3_reader.py b/dicom_validator/tests/spec_reader/test_part3_reader.py index 6b7e0d6..0eb5f83 100644 --- a/dicom_validator/tests/spec_reader/test_part3_reader.py +++ b/dicom_validator/tests/spec_reader/test_part3_reader.py @@ -223,7 +223,7 @@ def test_parsed_enum_values(self, reader): assert "(0082,0036)" in description assert "enums" in description["(0082,0036)"] enums = description["(0082,0036)"]["enums"] - assert enums == ["FAILURE", "WARNING", "INFORMATIVE"] + assert enums == [{"val": ["FAILURE", "WARNING", "INFORMATIVE"]}] @pytest.mark.parametrize( "revision", ["2015b", "2023c"], indirect=True, scope="session" @@ -236,4 +236,4 @@ def test_linked_enum_values(self, reader): # Device Motion Execution Mode tag = description["(300A,0450)"]["items"]["(300A,0451)"] assert "enums" in tag - assert tag["enums"] == ["CONTINUOUS", "TRIGGERED", "AUTOMATIC"] + assert tag["enums"] == [{"val": ["CONTINUOUS", "TRIGGERED", "AUTOMATIC"]}] diff --git a/dicom_validator/tests/test_cmdline_tools.py b/dicom_validator/tests/test_cmdline_tools.py index f24e550..2801bc6 100644 --- a/dicom_validator/tests/test_cmdline_tools.py +++ b/dicom_validator/tests/test_cmdline_tools.py @@ -3,7 +3,6 @@ import pytest -from dicom_validator.spec_reader.enum_parser import EnumParser from dicom_validator.validate_iods import main @@ -43,5 +42,3 @@ def test_validate_sr(revision, caplog, standard_path, dicom_fixture_path): # regression test for #9 assert "Unknown SOPClassUID" not in caplog.text assert "Tag (0008,1070) (Operators' Name) is missing" in caplog.text - - print(f"Direct: {EnumParser.nr_direct_enums}, linked: {EnumParser.nr_linked_enums}") diff --git a/dicom_validator/tests/validator/conftest.py b/dicom_validator/tests/validator/conftest.py index da41108..a1265a2 100644 --- a/dicom_validator/tests/validator/conftest.py +++ b/dicom_validator/tests/validator/conftest.py @@ -35,21 +35,27 @@ def json_fixture_path(standard_path): @pytest.fixture(scope="session") def iod_info(json_fixture_path): - with open(json_fixture_path / EditionReader.iod_info_json) as info_file: + with open( + json_fixture_path / EditionReader.iod_info_json, encoding="utf8" + ) as info_file: info = json.load(info_file) yield info @pytest.fixture(scope="session") def dict_info(json_fixture_path): - with open(json_fixture_path / EditionReader.dict_info_json) as info_file: + with open( + json_fixture_path / EditionReader.dict_info_json, encoding="utf8" + ) as info_file: info = json.load(info_file) yield info @pytest.fixture(scope="session") def module_info(json_fixture_path): - with open(json_fixture_path / EditionReader.module_info_json) as info_file: + with open( + json_fixture_path / EditionReader.module_info_json, encoding="utf8" + ) as info_file: info = json.load(info_file) yield info diff --git a/dicom_validator/tests/validator/test_iod_validator.py b/dicom_validator/tests/validator/test_iod_validator.py index bf4023a..9e3b1d4 100644 --- a/dicom_validator/tests/validator/test_iod_validator.py +++ b/dicom_validator/tests/validator/test_iod_validator.py @@ -429,9 +429,8 @@ def test_conditional_includes(self, validator): "BitsStored": 1, } ) - def test_enum_value(self, validator): + def test_invalid_enum_value(self, validator): result = validator.validate() - print(result) # Presentation LUT Shape: incorrect enum value assert has_tag_error( @@ -455,3 +454,106 @@ def test_enum_value(self, validator): "value is not allowed", "(value: 1, allowed: 8, 9, 10, 11, 12, 13, 14, 15, 16)", ) + + @pytest.mark.tag_set( + { + # MR Image Storage + "SOPClassUID": "1.2.840.10008.5.1.4.1.1.4", + "PatientName": "XXX", + "PatientID": "ZZZ", + "ScanningSequence": ["SE", "EP"], + } + ) + def test_valid_multi_valued_enum(self, validator): + result = validator.validate() + + # Scanning Sequence - all values allowed + assert not has_tag_error( + result, + "MR Image", + "(0018,0020)", + "value is not allowed", + ) + + @pytest.mark.tag_set( + { + # MR Image Storage + "SOPClassUID": "1.2.840.10008.5.1.4.1.1.4", + "PatientName": "XXX", + "PatientID": "ZZZ", + "ScanningSequence": ["SE", "EP", "IV"], + } + ) + def test_invalid_multi_valued_enum(self, validator): + result = validator.validate() + + # Scanning Sequence - IV not allowed + assert has_tag_error( + result, + "MR Image", + "(0018,0020)", + "value is not allowed", + "(value: IV, allowed: SE, IR, GR, EP, RM)", + ) + + @pytest.mark.tag_set( + { + # Ophthalmic Optical Coherence Tomography B-scan Volume Analysis Storage + "SOPClassUID": "1.2.840.10008.5.1.4.1.1.77.1.5.8", + "PatientName": "XXX", + "PatientID": "ZZZ", + "ImageType": ["ORIGINAL", "PRIMARY"], + } + ) + def test_valid_indexed_enum(self, validator): + result = validator.validate() + + # Image Type - both values correct + assert not has_tag_error( + result, + "Ophthalmic Optical Coherence Tomography B-scan Volume Analysis Image", + "(0008,0008)", + "value is not allowed", + ) + + @pytest.mark.tag_set( + { + # Ophthalmic Optical Coherence Tomography B-scan Volume Analysis Storage + "SOPClassUID": "1.2.840.10008.5.1.4.1.1.77.1.5.8", + "PatientName": "XXX", + "PatientID": "ZZZ", + "ImageType": ["ORIGINAL", "SECONDARY"], + } + ) + def test_invalid_indexed_enum(self, validator): + result = validator.validate() + + # Image Type - second value incorrect + assert has_tag_error( + result, + "Ophthalmic Optical Coherence Tomography B-scan Volume Analysis Image", + "(0008,0008)", + "value is not allowed", + "(value: SECONDARY, allowed: PRIMARY)", + ) + + @pytest.mark.tag_set( + { + # Ophthalmic Optical Coherence Tomography B-scan Volume Analysis Storage + "SOPClassUID": "1.2.840.10008.5.1.4.1.1.77.1.5.8", + "PatientName": "XXX", + "PatientID": "ZZZ", + "ImageType": ["PRIMARY", "ORIGINAL"], + } + ) + def test_indexed_enum_incorrect_index(self, validator): + result = validator.validate() + + # Image Type - allowed values at incorrect index + assert has_tag_error( + result, + "Ophthalmic Optical Coherence Tomography B-scan Volume Analysis Image", + "(0008,0008)", + "value is not allowed", + "(value: ORIGINAL, allowed: PRIMARY)", + ) diff --git a/dicom_validator/validator/iod_validator.py b/dicom_validator/validator/iod_validator.py index 1a055ab..e636257 100644 --- a/dicom_validator/validator/iod_validator.py +++ b/dicom_validator/validator/iod_validator.py @@ -4,6 +4,7 @@ from dataclasses import dataclass from pydicom import Sequence +from pydicom.multival import MultiValue from pydicom.tag import Tag from dicom_validator.spec_reader.condition import ( @@ -391,12 +392,20 @@ def _validate_attribute(self, tag_id, attribute): if value is None or isinstance(value, Sequence) and not value: error_kind = "empty" if value is not None and "enums" in attribute: - if value not in attribute["enums"]: - error_kind = "value is not allowed" - extra_msg = ( - f" (value: {value}, allowed: " - f"{', '.join([str(e) for e in attribute['enums']])})" - ) + if not isinstance(value, MultiValue): + value = [value] + for i, v in enumerate(value): + for enums in attribute["enums"]: + # if an index is there, we only check the value for the + # correct index; otherwise there will only be one entry + if "index" in enums and int(enums["index"]) != i + 1: + continue + if v not in enums["val"]: + error_kind = "value is not allowed" + extra_msg = ( + f" (value: {v}, allowed: " + f"{', '.join([str(e) for e in enums['val']])})" + ) if error_kind is not None: extra_msg = extra_msg or self._condition_message(condition_dict)