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

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/test_ahb_serialisierung.py

@@ -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