Add Twix, MRS and PMU raw data readers

.codespell-ignorewords

@@ -1,2 +1,4 @@
 nd
 te
+puls
+ptd

.pre-commit-config.yaml

@@ -9,7 +9,7 @@ repos:
       - id: check-yaml
       - id: check-added-large-files
   - repo: https://github.com/psf/black
-    rev: 22.3.0
+    rev: 24.10.0
     hooks:
       - id: black
         exclude: ^(fileformats/vendor/siemens/_version\.py)$
@@ -26,10 +26,11 @@ repos:
     rev: 7.0.0
     hooks:
       - id: flake8
+        additional_dependencies: [flake8-pyproject]
   - repo: https://github.com/pre-commit/mirrors-mypy
     rev: v1.11.2
     hooks:
       - id: mypy
-        args: [--strict, --install-types, --non-interactive]
+        args: [--strict, --no-warn-unused-ignores, --install-types, --non-interactive]
         exclude: tests
         additional_dependencies: [pytest, fileformats]

LICENSE

@@ -88,4 +88,4 @@ Creative Commons is not a party to its public licenses. Notwithstanding, Creativ
 
 Creative Commons may be contacted at creativecommons.org.
 
-Additional languages available: العربية, čeština, Deutsch, Ελληνικά, Español, euskara, suomeksi, français, hrvatski, Bahasa Indonesia, italiano, 日本語, 한국어, Lietuvių, latviski, te reo Māori, Nederlands, norsk, polski, português, română, русский, Slovenščina, svenska, Türkçe, українська, 中文, 華語. Please read the FAQ for more information about official translations.
+Additional languages available: العربية, čeština, Deutsch, Ελληνικά, Español, euskara, suomeksi, français, hrvatski, Bahasa Indonesia, italiano, 日本語, 한국어, Lietuvių, latviski, te reo Māori, Nederlands, norsk, polski, português, română, русский, Slovenščina, svenska, Türkçe, українська, 中文, 華語. Please read the FAQ for more information about official translations.

extras/LICENSE

@@ -88,4 +88,4 @@ Creative Commons is not a party to its public licenses. Notwithstanding, Creativ
 
 Creative Commons may be contacted at creativecommons.org.
 
-Additional languages available: العربية, čeština, Deutsch, Ελληνικά, Español, euskara, suomeksi, français, hrvatski, Bahasa Indonesia, italiano, 日本語, 한국어, Lietuvių, latviski, te reo Māori, Nederlands, norsk, polski, português, română, русский, Slovenščina, svenska, Türkçe, українська, 中文, 華語. Please read the FAQ for more information about official translations.
+Additional languages available: العربية, čeština, Deutsch, Ελληνικά, Español, euskara, suomeksi, français, hrvatski, Bahasa Indonesia, italiano, 日本語, 한국어, Lietuvių, latviski, te reo Māori, Nederlands, norsk, polski, português, română, русский, Slovenščina, svenska, Türkçe, українська, 中文, 華語. Please read the FAQ for more information about official translations.

extras/fileformats/extras/vendor/siemens/medimage.py

@@ -28,6 +28,8 @@
 from fileformats.core import extra_implementation, FileSet
 from fileformats.medimage.dicom import DicomImage
 from fileformats.vendor.siemens.medimage import (
+    SyngoMr_Xa_Puls,
+    SyngoMr_Xa_Rda,
     SyngoMi_Vr20b_RawData,
     SyngoMi_Vr20b_LargeRawData,
     SyngoMi_Vr20b_ListMode,
@@ -37,6 +39,7 @@
     SyngoMi_Vr20b_Normalisation,
     SyngoMi_Vr20b_Parameterisation,
     SyngoMi_Vr20b_CtSpl,
+    SyngoMr_Xa_Twix,
 )
 from fileformats.core.io import BinaryIOWindow
 
@@ -67,7 +70,7 @@ def siemens_pet_raw_data_read_metadata(
             pet_raw_data.dcm_hdr_size_int_offset,
         )
         dcm = pydicom.dcmread(window, specific_tags=specific_tags)
-    return DicomImage.pydicom_to_dict(dcm)
+    return DicomImage.pydicom_to_dict(dcm)  # type: ignore[no-any-return]
 
 
 @extra_implementation(SyngoMi_Vr20b_RawData.load_pydicom)
@@ -100,7 +103,7 @@ def siemens_petct_raw_data_read_metadata(
             *pet_raw_data.dicom_header_limits,
         )
         dcm = pydicom.dcmread(window, specific_tags=specific_tags)
-    return DicomImage.pydicom_to_dict(dcm)
+    return DicomImage.pydicom_to_dict(dcm)  # type: ignore[no-any-return]
 
 
 @extra_implementation(SyngoMi_Vr20b_RawData.load_pydicom)
@@ -210,3 +213,74 @@ def siemens_pet_parameterisation_generate_sample_data(
     generator: SampleFileGenerator,
 ) -> ty.List[Path]:
     return get_pet_replay_param_data(out_dir=generator.dest_dir)  # type: ignore[no-any-return]
+
+
+@extra_implementation(FileSet.read_metadata)
+def siemens_rda_read_metadata(
+    rda: SyngoMr_Xa_Rda,
+    **kwargs: ty.Any,
+) -> ty.Mapping[str, ty.Any]:
+    HEADER_END = b">>> End of header <<<"
+    raw = rda.read_contents()
+    end_idx = raw.find(HEADER_END)
+    if end_idx == -1:
+        return {}
+    header_text = raw[:end_idx].decode("latin-1")
+    metadata: ty.Dict[str, ty.Any] = {}
+    for line in header_text.splitlines():
+        if line.startswith(">>>"):
+            continue
+        if ": " in line:
+            key, value = line.split(": ", 1)
+            metadata[key.strip()] = value.strip()
+    return metadata
+
+
+@extra_implementation(FileSet.read_metadata)
+def siemens_puls_read_metadata(
+    puls: SyngoMr_Xa_Puls,
+    **kwargs: ty.Any,
+) -> ty.Mapping[str, ty.Any]:
+    text = puls.read_contents()
+    metadata: ty.Dict[str, ty.Any] = {}
+    for line in text.splitlines():
+        line = line.strip()
+        # Based off of one example file, ends in 6003
+        if not line or line == "6003":
+            continue
+        if ": " in line:
+            key, value = line.split(": ", 1)
+            metadata[key.strip()] = value.strip()
+    return metadata
+
+
+@extra_implementation(SyngoMr_Xa_Twix.read_twix)
+def twix_raw_data_read_twix(
+    twix_data: SyngoMr_Xa_Twix,
+    **kwargs: ty.Any,
+) -> ty.Any:
+    import twixtools
+
+    return twixtools.read_twix(str(twix_data.fspath), **kwargs)
+
+
+@extra_implementation(SyngoMr_Xa_Twix.map_twix)
+def twix_raw_data_map_twix(
+    twix_data: SyngoMr_Xa_Twix,
+    **kwargs: ty.Any,
+) -> ty.Any:
+    import twixtools
+
+    return twixtools.map_twix(str(twix_data.fspath), **kwargs)
+
+
+@extra_implementation(FileSet.read_metadata)
+def twix_raw_data_read_metadata(
+    twix_data: SyngoMr_Xa_Twix,
+    **kwargs: ty.Any,
+) -> ty.Mapping[str, ty.Any]:
+    import twixtools
+
+    scans = twixtools.read_twix(str(twix_data.fspath), parse_data=False, verbose=False)
+    # Return the parsed protocol headers from the last measurement (the actual scan)
+    return scans[-1].get("hdr", {})  # type: ignore[no-any-return]

extras/pyproject.toml

@@ -11,6 +11,7 @@ dependencies = [
     "fileformats-vendor-siemens",
     "medimages4tests >=0.3.0",
     "pydicom >=2.3.1",
+    "twixtools",
 ]
 license = {file = "LICENSE"}
 authors = [

fileformats/vendor/siemens/medimage/__init__.py

@@ -11,6 +11,9 @@
     SyngoMi_Vr20b_DynamicSinogramSeries,
     SyngoMi_Vr20b_CtSpl,
 )
+from .puls import SyngoMr_Xa_Puls
+from .rda import SyngoMr_Xa_Rda
+from .twix import SyngoMr_Xa_Twix
 
 Biograph128Vision_Vr20b_PetRawData = SyngoMi_Vr20b_RawData
 Biograph128Vision_Vr20b_PetLargeRawData = SyngoMi_Vr20b_LargeRawData
@@ -36,4 +39,7 @@
     "SyngoMi_Vr20b_Physio",
     "SyngoMi_Vr20b_DynamicSinogramSeries",
     "SyngoMi_Vr20b_CtSpl",
+    "SyngoMr_Xa_Puls",
+    "SyngoMr_Xa_Rda",
+    "SyngoMr_Xa_Twix",
 ]

fileformats/vendor/siemens/medimage/puls.py

@@ -0,0 +1,11 @@
+from fileformats.generic import UnicodeFile
+
+
+class SyngoMr_Xa_Puls(UnicodeFile):
+    """Siemens PMU (Physiological Monitoring Unit) pulse oximetry data.
+
+    Plain-text file containing pulse signal samples followed by
+    summary statistics and timing metadata.
+    """
+
+    ext = ".puls"

fileformats/vendor/siemens/medimage/rda.py

@@ -0,0 +1,27 @@
+from fileformats.core import validated_property
+from fileformats.core.exceptions import FormatMismatchError
+from fileformats.medimage.raw.mri import Rda
+
+HEADER_START = b">>> Begin of header <<<"
+
+
+class SyngoMr_Xa_Rda(Rda):  # type: ignore[misc]
+    """Siemens MRI spectroscopy data in .rda format.
+
+    The file starts with a plain-text header delimited by
+    ``>>> Begin of header <<<`` and ``>>> End of header <<<`` markers,
+    followed by binary spectroscopy data.
+
+    Inherits extension and binary classifier.
+    """
+
+    @validated_property
+    def is_valid_rda(self) -> bool:
+        """Validate that the file begins with the expected header marker."""
+        first_bytes = self.read_contents(size=len(HEADER_START), offset=0)
+        if first_bytes != HEADER_START:
+            raise FormatMismatchError(
+                f"File {self.fspath!r} does not start with the expected "
+                f"RDA header marker (got {first_bytes!r})"
+            )
+        return True

fileformats/vendor/siemens/medimage/syngo_mi.py

@@ -28,7 +28,7 @@
     import pydicom
 
 
-class SyngoMi_Vr20b_RawData(PetRawData):
+class SyngoMi_Vr20b_RawData(PetRawData):  # type: ignore[misc]
     """PET raw data format as produced by Siemens Biograph 128 Vision. It is used to
     store a range of raw PET data such as list-mode, calibration and sinogram files.
 
@@ -47,7 +47,7 @@ def read_tag(self, tag: ty.Tuple[int, int]) -> ty.Union[str, bytes, None]:
                 *self.dicom_header_limits,
             )
             value = get_dicom_tag(window, tag)
-        return value
+        return value  # type: ignore[no-any-return]
 
     @validated_property
     def image_type(self) -> str:
@@ -146,41 +146,41 @@ def dicom_header_limits(self) -> ty.Tuple[int, int]:  # type: ignore[override]
         )
 
 
-class SyngoMi_Vr20b_ListMode(SyngoMi_Vr20b_LargeRawData, PetListMode):
+class SyngoMi_Vr20b_ListMode(SyngoMi_Vr20b_LargeRawData, PetListMode):  # type: ignore[misc]
     expected_image_type = "PET_LISTMODE"
 
 
-class SyngoMi_Vr20b_Sinogram(SyngoMi_Vr20b_LargeRawData, PetSinogram):
+class SyngoMi_Vr20b_Sinogram(SyngoMi_Vr20b_LargeRawData, PetSinogram):  # type: ignore[misc]
     "histogrammed projection data in a reconstruction-friendly format"
 
     expected_image_type = "PET_EM_SINOGRAM"
 
 
-class SyngoMi_Vr20b_DynamicSinogram(SyngoMi_Vr20b_LargeRawData, PetSinogram):
+class SyngoMi_Vr20b_DynamicSinogram(SyngoMi_Vr20b_LargeRawData, PetSinogram):  # type: ignore[misc]
     "histogrammed projection data in a reconstruction-friendly format"
 
     expected_image_type = "PET_SINO_DYNAMIC"
 
 
-class SyngoMi_Vr20b_CountRate(SyngoMi_Vr20b_LargeRawData, PetCountRate):
+class SyngoMi_Vr20b_CountRate(SyngoMi_Vr20b_LargeRawData, PetCountRate):  # type: ignore[misc]
     "number of prompt/random/single events per unit time"
 
     expected_image_type = "PET_COUNTRATE"
 
 
-class SyngoMi_Vr20b_Parameterisation(SyngoMi_Vr20b_LargeRawData, PetParameterisation):
+class SyngoMi_Vr20b_Parameterisation(SyngoMi_Vr20b_LargeRawData, PetParameterisation):  # type: ignore[misc]
     "number of prompt/random/single events per unit time"
 
     expected_image_type = "PET_REPLAY_PARAM"
 
 
-class SyngoMi_Vr20b_Normalisation(SyngoMi_Vr20b_LargeRawData, PetNormalisation):
+class SyngoMi_Vr20b_Normalisation(SyngoMi_Vr20b_LargeRawData, PetNormalisation):  # type: ignore[misc]
     "normalisation scan or the current cross calibration factor"
 
     expected_image_type = "PET_CALIBRATION"
 
 
-class SyngoMi_Vr20b_Physio(SyngoMi_Vr20b_LargeRawData, PetPhysio):
+class SyngoMi_Vr20b_Physio(SyngoMi_Vr20b_LargeRawData, PetPhysio):  # type: ignore[misc]
     "normalisation scan or the current cross calibration factor"
 
     expected_image_type = "PET_PHYSIO"

fileformats/vendor/siemens/medimage/tests/test_puls.py

@@ -0,0 +1,40 @@
+import pytest
+from fileformats.core.exceptions import FormatMismatchError
+from fileformats.vendor.siemens.medimage import SyngoMr_Xa_Puls
+
+
+def _make_puls(tmp_path, name, content):  # codespell:ignore puls
+    p = tmp_path / name
+    p.write_text(content)
+    return p
+
+
+MINIMAL_PULS = (
+    "1 4 40 280 5003\n"
+    "ECG  Freq Per: 0 0\n"
+    "PULS Freq Per: 72 830\n"
+    "LogStartMDHTime:  49470570\n"
+    "LogStopMDHTime:   49583052\n"
+    "6003\n"
+)
+
+
+def test_valid_puls(tmp_path):
+    p = _make_puls(tmp_path, "test.puls", MINIMAL_PULS)
+    puls = SyngoMr_Xa_Puls(p)
+    assert puls.fspath == p
+
+
+def test_wrong_extension(tmp_path):
+    p = _make_puls(tmp_path, "test.txt", MINIMAL_PULS)
+    with pytest.raises(FormatMismatchError):
+        SyngoMr_Xa_Puls(p)
+
+
+def test_metadata_parsing(tmp_path):
+    p = _make_puls(tmp_path, "test.puls", MINIMAL_PULS)
+    puls = SyngoMr_Xa_Puls(p)
+    metadata = puls.read_metadata()
+    assert metadata["PULS Freq Per"] == "72 830"
+    assert metadata["LogStartMDHTime"] == "49470570"
+    assert metadata["LogStopMDHTime"] == "49583052"

fileformats/vendor/siemens/medimage/tests/test_rda.py

@@ -0,0 +1,74 @@
+import pytest
+from fileformats.core.exceptions import FormatMismatchError
+from fileformats.vendor.siemens.medimage import SyngoMr_Xa_Rda
+
+HEADER_START = b">>> Begin of header <<<"
+HEADER_END = b">>> End of header <<<"
+
+
+def _make_rda(
+    tmp_path, name, header_lines=None, start_marker=True, binary_tail=b"\x00" * 64
+):
+    """Helper to create a synthetic .rda file."""
+    parts = []
+    if start_marker:
+        parts.append(HEADER_START + b"\n")
+    if header_lines is not None:
+        for line in header_lines:
+            parts.append(line.encode("latin-1") + b"\n")
+    parts.append(HEADER_END + b"\n")
+    parts.append(binary_tail)
+    p = tmp_path / name
+    p.write_bytes(b"".join(parts))
+    return p
+
+
+def test_valid_rda(tmp_path):
+    rda_file = _make_rda(tmp_path, "test.rda", header_lines=["PatientName: John"])
+    rda = SyngoMr_Xa_Rda(rda_file)
+    assert rda.is_valid_rda is True
+
+
+def test_wrong_extension(tmp_path):
+    rda_file = _make_rda(tmp_path, "test.txt", header_lines=["PatientName: John"])
+    with pytest.raises(FormatMismatchError):
+        SyngoMr_Xa_Rda(rda_file)
+
+
+def test_missing_header_marker(tmp_path):
+    rda_file = _make_rda(
+        tmp_path, "test.rda", header_lines=["PatientName: John"], start_marker=False
+    )
+    with pytest.raises(FormatMismatchError):
+        SyngoMr_Xa_Rda(rda_file)
+
+
+def test_metadata_parsing(tmp_path):
+    header_lines = [
+        "PatientName: James",
+        "PatientID: 12345",
+        "TR: 2000",
+        "TE: 30",
+    ]
+    rda_file = _make_rda(tmp_path, "test.rda", header_lines=header_lines)
+    rda = SyngoMr_Xa_Rda(rda_file)
+    metadata = rda.read_metadata()
+    assert metadata["PatientName"] == "James"
+    assert metadata["PatientID"] == "12345"
+    assert metadata["TR"] == "2000"
+    assert metadata["TE"] == "30"
+
+
+def test_metadata_empty_header(tmp_path):
+    rda_file = _make_rda(tmp_path, "test.rda", header_lines=[])
+    rda = SyngoMr_Xa_Rda(rda_file)
+    metadata = rda.read_metadata()
+    assert metadata == {}
+
+
+def test_metadata_colon_in_value(tmp_path):
+    header_lines = ["ScanTime: 10:30:00"]
+    rda_file = _make_rda(tmp_path, "test.rda", header_lines=header_lines)
+    rda = SyngoMr_Xa_Rda(rda_file)
+    metadata = rda.read_metadata()
+    assert metadata["ScanTime"] == "10:30:00"