Add SR "get_evidence" methods

src/highdicom/sr/sop.py

@@ -1,11 +1,13 @@
 """Module for SOP Classes of Structured Report (SR) IODs."""
 import datetime
+from itertools import chain
 import logging
 from collections import defaultdict
 from copy import deepcopy
 from os import PathLike
 from typing import (
     Any,
+    Generator,
     cast,
     Mapping,
     List,
@@ -351,6 +353,111 @@ def content(self) -> ContentSequence:
         """highdicom.sr.value_types.ContentSequence: SR document content"""
         return self._content
 
+    def get_evidence(
+        self,
+        current_procedure_only: bool = False,
+    ) -> List[Tuple[UID, UID, UID, UID]]:
+        """Get a list of all SOP Instances listed as evidence in this SR.
+
+        Parameters
+        ----------
+        current_procedure_only: bool, optional
+            If True, return only evidence created in order to satisfy the
+            current requested procedure (found in the
+            *CurrentRequestedProcedureEvidenceSequence*). If False, also
+            include other evidence (found in the
+            *PertinentOtherEvidenceSequence*).
+
+        Returns
+        -------
+        List[Tuple[highdicom.UID, highdicom.UID, highdicom.UID, highdicom.UID]]:
+            List of tuples of UIDs, each representing a single instance. Each
+            tuple consists of (StudyInstanceUID, SeriesInstanceUID,
+            SOPInstanceUID, SOPClassUID).
+
+        """
+        def extract_evidence(
+            sequence: DataElementSequence,
+        ) -> Generator[Tuple[UID, UID, UID, UID], None, None]:
+            for item in sequence:
+                for series_ds in item.ReferencedSeriesSequence:
+                    for instance_ds in series_ds.ReferencedSOPSequence:
+                        yield (
+                            UID(item.StudyInstanceUID),
+                            UID(series_ds.SeriesInstanceUID),
+                            UID(instance_ds.ReferencedSOPInstanceUID),
+                            UID(instance_ds.ReferencedSOPClassUID),
+                        )
+
+        current_evidence_seq = self.get('CurrentRequestedProcedureEvidenceSequence')
+        if current_evidence_seq is not None:
+            current_evidence = extract_evidence(current_evidence_seq)
+        else:
+            current_evidence = []
+
+        other_evidence_seq = self.get('PertinentOtherEvidenceSequence')
+        if other_evidence_seq is not None and not current_procedure_only:
+            other_evidence = extract_evidence(other_evidence_seq)
+        else:
+            other_evidence = []
+
+        evidence = list(chain(current_evidence, other_evidence))
+
+        # Deduplicate the list while preserving order
+        evidence = list(dict.fromkeys(evidence))
+
+        return evidence
+
+    def get_evidence_series(
+        self,
+        current_procedure_only: bool = False,
+    ) -> List[Tuple[UID, UID]]:
+        """Get a list of all series listed as evidence in this SR.
+
+        Parameters
+        ----------
+        current_procedure_only: bool, optional
+            If True, return only evidence created in order to satisfy the
+            current requested procedure (found in the
+            *CurrentRequestedProcedureEvidenceSequence*). If False, also
+            include other evidence (found in the
+            *PertinentOtherEvidenceSequence*).
+
+        Returns
+        -------
+        List[Tuple[highdicom.UID, highdicom.UID]]:
+            List of tuples of UIDs, each representing a single series. Each
+            tuple consists of (StudyInstanceUID, SeriesInstanceUID).
+
+        """
+        def extract_evidence_series(
+            sequence: DataElementSequence,
+        ) -> Generator[Tuple[UID, UID], None, None]:
+            for item in sequence:
+                for series_ds in item.ReferencedSeriesSequence:
+                    yield (
+                        UID(item.StudyInstanceUID),
+                        UID(series_ds.SeriesInstanceUID),
+                    )
+
+        current_evidence_seq = self.get('CurrentRequestedProcedureEvidenceSequence')
+        if current_evidence_seq is not None:
+            current_evidence = extract_evidence_series(current_evidence_seq)
+        else:
+            current_evidence = []
+
+        other_evidence_seq = self.get('PertinentOtherEvidenceSequence')
+        if other_evidence_seq is not None and not current_procedure_only:
+            other_evidence = extract_evidence_series(other_evidence_seq)
+        else:
+            other_evidence = []
+
+        evidence = list(chain(current_evidence, other_evidence))
+
+        # Deduplicate the list while preserving order
+        evidence = list(dict.fromkeys(evidence))
+
+        return evidence
 
 class EnhancedSR(_SR):
 

tests/test_sr.py

@@ -3772,6 +3772,20 @@ def test_construction(self):
             performed_procedure_codes=self._performed_procedures
         )
         assert report.SOPClassUID == '1.2.840.10008.5.1.4.1.1.88.22'
+        evidence = report.get_evidence()
+        assert len(evidence) == 1
+        assert evidence[0] == (
+            self._ref_dataset.StudyInstanceUID,
+            self._ref_dataset.SeriesInstanceUID,
+            self._ref_dataset.SOPInstanceUID,
+            self._ref_dataset.SOPClassUID,
+        )
+        evidence_series = report.get_evidence_series()
+        assert len(evidence_series) == 1
+        assert evidence_series[0] == (
+            self._ref_dataset.StudyInstanceUID,
+            self._ref_dataset.SeriesInstanceUID,
+        )
 
     def test_construction_content_is_sequence(self):
         report = EnhancedSR(
@@ -3913,6 +3927,21 @@ def test_construction(self):
         with pytest.raises(AttributeError):
             assert report.PertinentOtherEvidenceSequence
 
+        evidence = report.get_evidence()
+        assert len(evidence) == 1
+        assert evidence[0] == (
+            self._ref_dataset.StudyInstanceUID,
+            self._ref_dataset.SeriesInstanceUID,
+            self._ref_dataset.SOPInstanceUID,
+            self._ref_dataset.SOPClassUID,
+        )
+        evidence_series = report.get_evidence_series()
+        assert len(evidence_series) == 1
+        assert evidence_series[0] == (
+            self._ref_dataset.StudyInstanceUID,
+            self._ref_dataset.SeriesInstanceUID,
+        )
+
     def test_construction_content_is_sequence(self):
         report = ComprehensiveSR(
             evidence=[self._ref_dataset],
@@ -4029,6 +4058,96 @@ def test_evidence_multiple_studies(self):
         with pytest.raises(AttributeError):
             assert report.PertinentOtherEvidenceSequence
 
+        evidence = report.get_evidence()
+        assert len(evidence) == 2
+        assert evidence[0] == (
+            self._ref_dataset.StudyInstanceUID,
+            self._ref_dataset.SeriesInstanceUID,
+            self._ref_dataset.SOPInstanceUID,
+            self._ref_dataset.SOPClassUID,
+        )
+        assert evidence[1] == (
+            ref_dataset.StudyInstanceUID,
+            ref_dataset.SeriesInstanceUID,
+            ref_dataset.SOPInstanceUID,
+            ref_dataset.SOPClassUID,
+        )
+        evidence_series = report.get_evidence_series()
+        assert len(evidence_series) == 2
+        assert evidence_series[0] == (
+            self._ref_dataset.StudyInstanceUID,
+            self._ref_dataset.SeriesInstanceUID,
+        )
+        assert evidence_series[1] == (
+            ref_dataset.StudyInstanceUID,
+            ref_dataset.SeriesInstanceUID,
+        )
+
+    def test_current_and_other_evidence(self):
+        ref_dataset2 = deepcopy(self._ref_dataset)
+        ref_dataset2.SeriesInstanceUID = '1.2.3'
+        ref_dataset2.SOPInstanceUID = '1.2.3'
+
+        report = Comprehensive3DSR(
+            evidence=[self._ref_dataset, ref_dataset2],
+            content=self._content,
+            series_instance_uid=self._series_instance_uid,
+            series_number=self._series_number,
+            sop_instance_uid=self._sop_instance_uid,
+            instance_number=self._instance_number,
+            institution_name=self._institution_name,
+            institutional_department_name=self._department_name,
+            manufacturer=self._manufacturer
+        )
+        ref_evd_items = report.CurrentRequestedProcedureEvidenceSequence
+        assert len(ref_evd_items) == 1
+        unref_evd_items = report.PertinentOtherEvidenceSequence
+        assert len(unref_evd_items) == 1
+
+        evidence = report.get_evidence()
+        assert len(evidence) == 2
+        assert evidence[0] == (
+            self._ref_dataset.StudyInstanceUID,
+            self._ref_dataset.SeriesInstanceUID,
+            self._ref_dataset.SOPInstanceUID,
+            self._ref_dataset.SOPClassUID,
+        )
+        assert evidence[1] == (
+            ref_dataset2.StudyInstanceUID,
+            ref_dataset2.SeriesInstanceUID,
+            ref_dataset2.SOPInstanceUID,
+            ref_dataset2.SOPClassUID,
+        )
+        evidence_series = report.get_evidence_series()
+        assert len(evidence_series) == 2
+        assert evidence_series[0] == (
+            self._ref_dataset.StudyInstanceUID,
+            self._ref_dataset.SeriesInstanceUID,
+        )
+        assert evidence_series[1] == (
+            ref_dataset2.StudyInstanceUID,
+            ref_dataset2.SeriesInstanceUID,
+        )
+
+        current_evidence = report.get_evidence(
+            current_procedure_only=True
+        )
+        assert len(current_evidence) == 1
+        assert current_evidence[0] == (
+            self._ref_dataset.StudyInstanceUID,
+            self._ref_dataset.SeriesInstanceUID,
+            self._ref_dataset.SOPInstanceUID,
+            self._ref_dataset.SOPClassUID,
+        )
+        current_evidence_series = report.get_evidence_series(
+            current_procedure_only=True
+        )
+        assert len(current_evidence_series) == 1
+        assert current_evidence_series[0] == (
+            self._ref_dataset.StudyInstanceUID,
+            self._ref_dataset.SeriesInstanceUID,
+        )
+
     def test_srread(self):
         report = ComprehensiveSR(
             evidence=[self._ref_dataset],