feat: Support <Uebertragungsdatei> in AhbReader (adds flag is_on_uebertragungsdatei_level to AHB Segment model) (#132)

feat: Support `<Uebertragungsdatei>` in `AhbReader`

add flag to `Segment` model

Co-authored-by: Konstantin <[email protected]>

Author: hf-kklein

src/fundamend/models/anwendungshandbuch.py

@@ -87,6 +87,13 @@ class Segment(FundamendBaseModel):
     number: str  #: e.g. '00002'
     ahb_status: str | None  #: e.g. 'Muss'
     data_elements: tuple[DataElement | DataElementGroup, ...]
+    is_on_uebertragungsdatei_level: bool = False
+    """
+    true for e.g. S_UNA, S_UNB or S_UNZ which occur outside the regular AHB <M_FORMAT> structure.
+    We decided against introducing another (mostly useless) layer around the actual body of the Anwendungsfall.
+    This is only relevant (true) for MSCONS AHBs as of 2025-04-23 although the MIGs read like this _might_ also be true
+    for REMADV and INVOIC. We all know, how (self-)consistent BDEW documents are, even if you pay for them.
+    """
 
 
 class SegmentGroup(FundamendBaseModel):

src/fundamend/reader/ahbreader.py

@@ -119,7 +119,7 @@ def _to_data_element_group(element: ET.Element) -> DataElementGroup:
     )
 
 
-def _to_segment(element: ET.Element) -> Segment:
+def _to_segment(element: ET.Element, is_uebertragungsdatei_level: bool = False) -> Segment:
     assert _is_segment(element)
     data_elements: list[DataElement | DataElementGroup] = []
     for child in element:
@@ -138,6 +138,7 @@ def _to_segment(element: ET.Element) -> Segment:
         number=element.attrib["Number"],
         ahb_status=ahb_status,
         data_elements=tuple(data_elements),
+        is_on_uebertragungsdatei_level=is_uebertragungsdatei_level,
     )
 
 
@@ -231,17 +232,22 @@ def get_anwendungsfaelle(self) -> list[Anwendungsfall]:
             result.append(self._read_anwendungsfall(element))
         return result
 
-    def _iter_segments_and_segment_groups(self, element: ET.Element) -> list[SegmentGroup | Segment]:
+    def _iter_segments_and_segment_groups(
+        self, element: ET.Element, is_uebertragungsdatei_level: bool = False
+    ) -> list[SegmentGroup | Segment]:
         """recursive function that builds a list of all segments and segment groups"""
         result: list[Segment | SegmentGroup] = []
-        should_go_deeper = _is_anwendungsfall(element) or _is_format(element) or _is_uebertragungsdatei(element)
+        should_go_deeper = _is_anwendungsfall(element) or _is_format(element)
         if should_go_deeper:
             for sub_element in element:
                 result.extend(self._iter_segments_and_segment_groups(sub_element))
-        if _is_segment_group(element):
+        if _is_uebertragungsdatei(element):
+            for sub_element in element:
+                result.extend(self._iter_segments_and_segment_groups(sub_element, is_uebertragungsdatei_level=True))
+        elif _is_segment_group(element):
             result.append(_to_segment_group(element))
         elif _is_segment(element):
-            result.append(_to_segment(element))
+            result.append(_to_segment(element, is_uebertragungsdatei_level=is_uebertragungsdatei_level))
         return result
 
     def _read_anwendungsfall(self, original_element: ET.Element) -> Anwendungsfall:

src/fundamend/sqlmodels/anwendungshandbuch.py

@@ -184,6 +184,7 @@ class Segment(SQLModel, table=True):
     name: str = Field(index=True)  #: e.g. 'Beginn der Nachricht'
     number: str = Field(index=True)  #: e.g. '00002'
     ahb_status: str | None  #: e.g. 'Muss'
+    is_on_uebertragungsdatei_level: bool
     data_elements: list[DataElement] = Relationship(back_populates="segment")
     data_element_groups: list[DataElementGroup] = Relationship(back_populates="segment")
     position: Optional[int] = Field(default=None, index=True)
@@ -202,6 +203,7 @@ def from_model(cls, model: PydanticSegment, position: Optional[int] = None) -> "
             number=model.number,
             ahb_status=model.ahb_status,
             position=position,
+            is_on_uebertragungsdatei_level=model.is_on_uebertragungsdatei_level,
         )
         for _position, element in enumerate(model.data_elements):
             if isinstance(element, PydanticDataElement):
@@ -225,6 +227,7 @@ def to_model(self) -> PydanticSegment:
                     key=lambda y: y.position or 0,
                 )
             ),
+            is_on_uebertragungsdatei_level=self.is_on_uebertragungsdatei_level,
         )
 
 

unittests/__snapshots__/test_ahbreader.ambr

@@ -3,6 +3,7 @@
 
 import pytest
 
+from fundamend.models.anwendungshandbuch import Segment as AhbSegment
 from fundamend.models.messageimplementationguide import Segment as MigSegment
 from fundamend.reader import AhbReader, MigReader
 
@@ -61,7 +62,22 @@ def test_deserializing_all_migs() -> None:
         _ = reader.read()  # must not crash
 
 
-def test_uebertragungsdatei_flag_is_set() -> None:
+def test_uebertragungsdatei_level_flag_is_set_ahb() -> None:
+    if not is_private_submodule_checked_out():
+        pytest.skip("Skipping test because of missing private submodule")
+    reader = AhbReader(data_path / "FV2504" / "MSCONS_AHB_3_1f_Fehlerkorrektur_20250320.xml")
+    ahb13012 = [awf for awf in reader.read().anwendungsfaelle if awf.pruefidentifikator == "13012"][0]
+    unb_segment = ahb13012.elements[0]
+    assert isinstance(unb_segment, AhbSegment) and unb_segment.id == "UNB"
+    unz_segment = ahb13012.elements[-1]
+    assert isinstance(unz_segment, AhbSegment) and unz_segment.id == "UNZ"
+    unh_segment = [s for s in ahb13012.elements if isinstance(s, AhbSegment) and s.id == "UNH"][0]
+    assert unb_segment.is_on_uebertragungsdatei_level
+    assert unz_segment.is_on_uebertragungsdatei_level
+    assert not unh_segment.is_on_uebertragungsdatei_level
+
+
+def test_uebertragungsdatei_flag_is_set_mig() -> 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"