added csv output for Nachrichtenstrukturtabelle

src/migmose/__main__.py

@@ -7,7 +7,7 @@
 import click
 from maus.edifact import EdifactFormat
 
-from migmose.mig.nachrichtenstruktur import NachrichtenstrukturTabelle
+from migmose.mig.nachrichtenstrukturtabelle import NachrichtenstrukturTabelle
 from migmose.mig.nestednachrichtenstruktur import NestedNachrichtenstruktur
 from migmose.parsing import find_file_to_format, parse_raw_nachrichtenstrukturzeile
 
@@ -44,11 +44,12 @@ def main(input_dir: Path, output_dir, message_format: list[EdifactFormat]) -> No
     dict_files = find_file_to_format(message_format, input_dir)
     for m_format, file in dict_files.items():
         raw_lines = parse_raw_nachrichtenstrukturzeile(file)
-        nachrichtenstrukturtabelle = NachrichtenstrukturTabelle.init_raw_table(raw_lines)
+        nachrichtenstrukturtabelle = NachrichtenstrukturTabelle.create_nachrichtenstruktur_tabelle(raw_lines)
+        nachrichtenstrukturtabelle.to_csv(m_format, output_dir)
         nested_nachrichtenstruktur, _ = NestedNachrichtenstruktur.create_nested_nachrichtenstruktur(
             nachrichtenstrukturtabelle
         )
-        nested_nachrichtenstruktur.output_as_json(m_format, output_dir)
+        nested_nachrichtenstruktur.to_json(m_format, output_dir)
 
 
 if __name__ == "__main__":

src/migmose/mig/nachrichtenstrukturtabelle.py

@@ -0,0 +1,45 @@
+"""
+contains class mig tables
+"""
+
+import csv
+from pathlib import Path
+
+from loguru import logger
+from maus.edifact import EdifactFormat
+from pydantic import BaseModel
+
+from migmose.mig.nachrichtenstrukturzeile import NachrichtenstrukturZeile
+
+
+class NachrichtenstrukturTabelle(BaseModel):
+    """
+    class for mig tables
+    """
+
+    lines: list[NachrichtenstrukturZeile]
+
+    @classmethod
+    def create_nachrichtenstruktur_tabelle(cls, raw_lines: list[str]) -> "NachrichtenstrukturTabelle":
+        """
+        reads table as list of raw lines and returns a NachrichtenstrukturTabelle
+        consisting of NachrichtenstrukturZeilen
+        """
+        collected_lines: list[NachrichtenstrukturZeile] = []
+        for raw_line in raw_lines:
+            collected_lines.append(NachrichtenstrukturZeile.init_raw_lines(raw_line))
+        return cls(lines=collected_lines)
+
+    def to_csv(self, message_type: EdifactFormat, output_dir: Path) -> None:
+        """
+        writes the NestedNachrichtenstruktur as json
+        """
+        output_dir.mkdir(parents=True, exist_ok=True)
+        file_path = output_dir.joinpath(f"{message_type}_nachrichtenstruktur.csv")
+        fieldnames = list(NachrichtenstrukturZeile.model_fields.keys())
+        with open(file_path, "w", encoding="utf-8") as csv_file:
+            writer = csv.DictWriter(csv_file, fieldnames=fieldnames)
+            writer.writeheader()
+            for nachrichtenstruktur_zeile in self.lines:
+                writer.writerow(nachrichtenstruktur_zeile.model_dump())
+        logger.info(f"Wrote Nachrichtenstruktur for {message_type} to {file_path}")

src/migmose/mig/nestednachrichtenstruktur.py

@@ -11,7 +11,7 @@
 from maus.edifact import EdifactFormat
 from pydantic import BaseModel
 
-from migmose.mig.nachrichtenstruktur import NachrichtenstrukturTabelle
+from migmose.mig.nachrichtenstrukturtabelle import NachrichtenstrukturTabelle
 from migmose.mig.nachrichtenstrukturzeile import NachrichtenstrukturZeile
 
 
@@ -104,7 +104,7 @@ def create_nested_nachrichtenstruktur(
                     )
         return cls(header_linie=header_line, segmente=collected_segments, segmentgruppen=collected_segmentgroups), i
 
-    def output_as_json(self, message_type: EdifactFormat, output_dir: Path) -> None:
+    def to_json(self, message_type: EdifactFormat, output_dir: Path) -> None:
         """
         writes the NestedNachrichtenstruktur as json
         """

unittests/test_data/ORDCHG_MIG_1_1_info_20230331_v2_nested_nachrichtenstruktur.csv

@@ -0,0 +1,19 @@
+zaehler,nr,bezeichnung,standard_status,bdew_status,standard_maximale_wiederholungen,bdew_maximale_wiederholungen,ebene,inhalt
+0010,1,UNH,M,M,1,1,0,Nachrichten-Kopfsegment
+0020,2,BGM,M,M,1,1,0,Beginn der Nachricht
+0030,3,DTM,M,M,35,1,1,Nachrichtendatum
+0090,,SG1,C,D,9999,1,1,Referenz der Anfrage/Bestellung
+0100,4,RFF,M,M,1,1,1,Nachrichtennummer der Anfrage/Bestellung
+0090,,SG1,C,D,9999,1,1,Referenz des Vorgangs der Anfrage/Bestellung
+0100,5,RFF,M,M,1,1,1,Vorgangsnummer der Anfrage/Bestellung
+0090,,SG1,C,R,9999,1,1,Prüfidentifikator
+0100,6,RFF,M,M,1,1,1,Prüfidentifikator
+0150,,SG3,C,R,99,1,1,MP-ID Absender
+0160,7,NAD,M,M,1,1,1,MP-ID Absender
+0250,,SG6,C,D,5,1,2,Kontaktinformationen
+0260,8,CTA,M,M,1,1,2,Ansprechpartner
+0270,9,COM,C,R,5,5,3,Kommunikationsverbindung
+0150,,SG3,C,R,99,1,1,MP-ID Empfänger
+0160,10,NAD,M,M,1,1,1,MP-ID Empfänger
+2380,11,UNS,M,M,1,1,0,Abschnitts-Kontrollsegment
+2450,12,UNT,M,M,1,1,0,Nachrichten-Endesegment

unittests/test_nachrichtenstrukturtabelle.py

@@ -1,6 +1,9 @@
+import csv
 from pathlib import Path
 
-from migmose.mig.nachrichtenstruktur import NachrichtenstrukturTabelle
+from maus.edifact import EdifactFormat
+
+from migmose.mig.nachrichtenstrukturtabelle import NachrichtenstrukturTabelle
 from migmose.parsing import parse_raw_nachrichtenstrukturzeile
 
 
@@ -14,12 +17,35 @@ def test_init_nachrichtenstrukturtabelle(self):
             "\t0580\t527\tRFF\tC\tD\t9\t9\t3\tReferenz auf die ID einer Messlokation",
             "\t0590\tRFF\tC\tD\t9\t9\t3\tReferenz auf die ID einer Messlokation",
         ]
-        nachrichtenstrukturtabelle = NachrichtenstrukturTabelle.init_raw_table(raw_lines)
+        nachrichtenstrukturtabelle = NachrichtenstrukturTabelle.create_nachrichtenstruktur_tabelle(raw_lines)
         assert len(nachrichtenstrukturtabelle.lines) == len(raw_lines)
         assert nachrichtenstrukturtabelle.lines[0].zaehler == "0580"
 
     def test_parse(self):
         file_path = Path("unittests/test_data/ORDCHG_MIG_1_1_info_20230331_v2.docx")
         raw_lines = parse_raw_nachrichtenstrukturzeile(file_path)
-        nachrichtenstrukturtabelle = NachrichtenstrukturTabelle.init_raw_table(raw_lines)
+        nachrichtenstrukturtabelle = NachrichtenstrukturTabelle.create_nachrichtenstruktur_tabelle(raw_lines)
         assert len(nachrichtenstrukturtabelle.lines) == 18
+
+    def test_to_csv(self, tmp_path):
+        input_file = Path("unittests/test_data/ORDCHG_MIG_1_1_info_20230331_v2.docx")
+        message_format = EdifactFormat.ORDCHG
+        output_dir = tmp_path / Path("output")
+        raw_lines = parse_raw_nachrichtenstrukturzeile(input_file)
+        nachrichtenstrukturtabelle = NachrichtenstrukturTabelle.create_nachrichtenstruktur_tabelle(raw_lines)
+        nachrichtenstrukturtabelle.to_csv(message_format, output_dir)
+
+        file_path = output_dir / Path(f"{message_format}_nachrichtenstruktur.csv")
+
+        assert file_path.exists()
+
+        reference_file = Path("unittests/test_data/ORDCHG_MIG_1_1_info_20230331_v2_nested_nachrichtenstruktur.csv")
+
+        with open(file_path, "r", encoding="utf-8") as csv_file:
+            reader = csv.DictReader(csv_file)
+
+            with open(reference_file, "r", encoding="utf-8") as reference_csv:
+                reference_reader = csv.DictReader(reference_csv)
+                assert reader.fieldnames == reference_reader.fieldnames
+                for row, reference_row in zip(reader, reference_reader):
+                    assert row == reference_row

unittests/test_nestednachrichtenstruktur.py

@@ -3,7 +3,7 @@
 
 from maus.edifact import EdifactFormat
 
-from migmose.mig.nachrichtenstruktur import NachrichtenstrukturTabelle
+from migmose.mig.nachrichtenstrukturtabelle import NachrichtenstrukturTabelle
 from migmose.mig.nestednachrichtenstruktur import NestedNachrichtenstruktur
 from migmose.parsing import parse_raw_nachrichtenstrukturzeile
 
@@ -13,26 +13,26 @@ class TestNestedNachrichtenstruktur:
     A class with pytest unit tests for nested Nachrichtenstrukturen.
     """
 
-    def test_structure_table(self):
+    def test_create_nested_nachrichtenstruktur(self):
         file_path = Path("unittests/test_data/ORDCHG_MIG_1_1_info_20230331_v2.docx")
         raw_lines = parse_raw_nachrichtenstrukturzeile(file_path)
-        nachrichtenstrukturtabelle = NachrichtenstrukturTabelle.init_raw_table(raw_lines)
+        nachrichtenstrukturtabelle = NachrichtenstrukturTabelle.create_nachrichtenstruktur_tabelle(raw_lines)
         nested_nachrichtenstruktur, _ = NestedNachrichtenstruktur.create_nested_nachrichtenstruktur(
             nachrichtenstrukturtabelle
         )
         assert len(nested_nachrichtenstruktur.segmente) == 5
         assert len(nested_nachrichtenstruktur.segmentgruppen[3].segmentgruppen) == 1
 
-    def test_output_as_json(self, tmp_path):
+    def test_to_json(self, tmp_path):
         input_file = Path("unittests/test_data/ORDCHG_MIG_1_1_info_20230331_v2.docx")
         message_format = EdifactFormat.ORDCHG
         output_dir = tmp_path / Path("output")
         raw_lines = parse_raw_nachrichtenstrukturzeile(input_file)
-        nachrichtenstrukturtabelle = NachrichtenstrukturTabelle.init_raw_table(raw_lines)
+        nachrichtenstrukturtabelle = NachrichtenstrukturTabelle.create_nachrichtenstruktur_tabelle(raw_lines)
         nested_nachrichtenstruktur, _ = NestedNachrichtenstruktur.create_nested_nachrichtenstruktur(
             nachrichtenstrukturtabelle
         )
-        nested_nachrichtenstruktur.output_as_json(message_format, output_dir)
+        nested_nachrichtenstruktur.to_json(message_format, output_dir)
 
         file_path = output_dir / Path(f"{message_format}_nested_nachrichtenstruktur.json")