feat: Support <Uebertragungsdatei> in MigReader (adds flag is_on_uebertragungsdatei_level to MIG Segment model) (#133)

feat: Support `<Uebertragungsdatei>` in `MigReader` (adds flag is_on_uebertragungsdatei_level to Segment model)

We decided AGAINST modeling this as an extra level in our data model because it adds no value and makes the models less uniform than they are right now (see e.g. PR #130 ). This is because some MIGs do contain an Übertragungsdatei on root level, some don't and the distinction is more pain than the value gained (at least for our usecases). Feel free to report back in issues if this doesn't suit your needs.

Author: hf-kklein

src/fundamend/models/messageimplementationguide.py

@@ -121,6 +121,17 @@ class Segment(FundamendBaseModel):
     status_specification: MigStatus  # e.g. M
     example: str | None  #: e.g. "NAD+MS+9900259000002::293'"
     data_elements: tuple[DataElement | DataElementGroup, ...]
+    is_on_uebertragungsdatei_level: bool = False
+    """
+    true only for those segments that are on the 'UEBERTRAGUNGSDATEI' level (meaning, outside the <M_FORMAT> tag).
+    Similar to the corresponding AHB tag, we decided against replicating this XML level in the fundamend data model.
+    This is the case for the UNA and UNZ segments in UTILMD, REMADV or INVOIC.
+    But other formats like UTILTS or ORDERS have no such level at all.
+    Basically this flag describes on which level you may aggregate single "Geschäftsvorfälle"/transactions in a single
+    Übertragungsdatei.
+    You may, for example, send multiple UTILMDs oR MSCONS messages batched/aggregated in one file, but not multiple
+    ORDERs.
+    """
 
 
 class SegmentGroup(FundamendBaseModel):

src/fundamend/reader/migreader.py

@@ -71,7 +71,7 @@ def _to_data_element_group(element: ET.Element) -> DataElementGroup:
     )
 
 
-def _to_segment(element: ET.Element) -> Segment:
+def _to_segment(element: ET.Element, is_on_uebertragungsdatei_level: bool = False) -> Segment:
     assert _is_segment(element)
     data_elements: list[DataElement | DataElementGroup] = []
     for child in element:
@@ -94,6 +94,7 @@ def _to_segment(element: ET.Element) -> Segment:
         example=element.attrib["Example"] or None,
         number=element.attrib["Number"],
         data_elements=tuple(data_elements),
+        is_on_uebertragungsdatei_level=is_on_uebertragungsdatei_level,
     )
 
 
@@ -180,18 +181,22 @@ def _iter_segments_and_segment_groups(self, element: ET.Element) -> list[Segment
 
     def read(self) -> MessageImplementationGuide:
         """
-        read the entire file and convert it to a MessageImplementationGuid instance
+        read the entire file and convert it to a MessageImplementationGuide instance
         """
-        segments_and_groups = []
+        segments_and_groups: list[Segment | SegmentGroup] = []
         root = self._element_tree.getroot()
         if _is_uebertragungsdatei(root):
-            for elem in root.iter():
-                if elem.tag.startswith("M_"):
-                    root = elem
-                    break
-        for element in root:
-            segments_and_groups.extend(self._iter_segments_and_segment_groups(element))
-
+            for elem in root:
+                if _is_segment(elem):
+                    segments_and_groups.append(_to_segment(elem, is_on_uebertragungsdatei_level=True))
+                elif elem.tag.startswith("M_"):
+                    for element in elem:
+                        segments_and_groups.extend(self._iter_segments_and_segment_groups(element))
+                else:
+                    raise ValueError(f"unexpected element: {elem.tag}")
+        else:
+            for element in root:
+                segments_and_groups.extend(self._iter_segments_and_segment_groups(element))
         result = MessageImplementationGuide(
             veroeffentlichungsdatum=self.get_publishing_date(),
             autor=self.get_author(),

unittests/__snapshots__/test_migreader.ambr

@@ -3,6 +3,7 @@
 
 import pytest
 
+from fundamend.models.messageimplementationguide import Segment as MigSegment
 from fundamend.reader import AhbReader, MigReader
 
 from .conftest import is_private_submodule_checked_out
@@ -58,3 +59,27 @@ def test_deserializing_all_migs() -> None:
     for mig_file_path in data_path.rglob("**/*MIG*.xml"):
         reader = MigReader(mig_file_path)
         _ = reader.read()  # must not crash
+
+
+def test_uebertragungsdatei_flag_is_set() -> None:
+    if not is_private_submodule_checked_out():
+        pytest.skip("Skipping test because of missing private submodule")
+    mig_file_path = data_path / "FV2504" / "UTILMD_MIG_Strom_S2_1_Fehlerkorrektur_20250320.xml"
+    reader = MigReader(mig_file_path)
+    mig = reader.read()
+    una_segment = [s for s in mig.elements if isinstance(s, MigSegment) and s.id == "UNA"][0]
+    assert una_segment.is_on_uebertragungsdatei_level is True
+    unh_segment = [s for s in mig.elements if isinstance(s, MigSegment) and s.id == "UNH"][0]
+    assert unh_segment.is_on_uebertragungsdatei_level is False
+    unz_segment = [s for s in mig.elements if isinstance(s, MigSegment) and s.id == "UNZ"][0]
+    assert unz_segment.is_on_uebertragungsdatei_level is True
+
+
+def test_uebertragungsdatei_flag_is_not_set_outside_uebertragungsdatei() -> None:
+    if not is_private_submodule_checked_out():
+        pytest.skip("Skipping test because of missing private submodule")
+    mig_file_path = data_path / "FV2504" / "UTILTS_MIG_1_1e_Fehlerkorrektur_20241018.xml"  # has no uebertragungsdatei
+    reader = MigReader(mig_file_path)
+    mig = reader.read()
+    unh_segment = [s for s in mig.elements if isinstance(s, MigSegment) and s.id == "UNH"][0]
+    assert unh_segment.is_on_uebertragungsdatei_level is False