Skip to content

Commit

Permalink
generate sgh files from the nested nachrichtenstrukturjson files (#83)
Browse files Browse the repository at this point in the history 
* introduce segment group hierarchies

* linter issues

* Added unittests for sgh

* Update src/migmose/mig/segmentgrouphierarchies.py

Co-authored-by: konstantin <konstantin.klein@hochfrequenz.de>

* Update src/migmose/mig/segmentgrouphierarchies.py

Co-authored-by: konstantin <konstantin.klein@hochfrequenz.de>

* add reference to maus segmentgrouphierarchy class

* modified unittest

* linter

* modified to_json for segementgrouphierarchy

---------

Co-authored-by: konstantin <konstantin.klein@hochfrequenz.de>
  • Loading branch information
@DeltaDaniel @hf-kklein
DeltaDaniel and hf-kklein authored May 29, 2024
1 parent ef19d55 commit 44b529c
Show file tree
Hide file tree
Showing 6 changed files with 280 additions and 2 deletions.
21 changes: 19 additions & 2 deletions src/migmose/__main__.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@
from migmose.mig.nachrichtenstrukturtabelle import NachrichtenstrukturTabelle
from migmose.mig.nestednachrichtenstruktur import NestedNachrichtenstruktur
from migmose.mig.reducednestednachrichtenstruktur import ReducedNestedNachrichtenstruktur
from migmose.mig.segmentgrouphierarchies import SegmentGroupHierarchy
from migmose.parsing import find_file_to_format, parse_raw_nachrichtenstrukturzeile


Expand Down Expand Up @@ -52,8 +53,8 @@
@click.option(
"-ft",
"--file-type",
type=click.Choice(["csv", "nested_json", "reduced_nested_json"], case_sensitive=False),
default=["csv", "nested_json", "reduced_nested_json"],
type=click.Choice(["csv", "nested_json", "reduced_nested_json", "sgh_json"], case_sensitive=False),
default=["csv", "nested_json", "reduced_nested_json", "sgh_json"],
help="Defines the output format. Choose between csv and nested_json and reduced_nested_json. Default is csv.",
multiple=True,
)
Expand Down Expand Up @@ -121,6 +122,22 @@ def main(
output_dir_for_format,
)
reduced_nested_nachrichtenstruktur.to_json(m_format, output_dir_for_format)
if "sgh_json" in file_type:
nested_nachrichtenstruktur, _ = NestedNachrichtenstruktur.create_nested_nachrichtenstruktur(
nachrichtenstrukturtabelle
)
reduced_nested_nachrichtenstruktur = (
ReducedNestedNachrichtenstruktur.create_reduced_nested_nachrichtenstruktur(nested_nachrichtenstruktur)
)
sgh = SegmentGroupHierarchy.create_segmentgroup_hierarchy(reduced_nested_nachrichtenstruktur)
# Save the reduced nested Nachrichtenstruktur as json
logger.info(
"💾 Saving reduced nested Nachrichtenstruktur for {} and {} as json to {}.",
m_format,
format_version,
output_dir_for_format,
)
sgh.to_json(m_format, output_dir_for_format)


if __name__ == "__main__":
Expand Down
60 changes: 60 additions & 0 deletions src/migmose/mig/segmentgrouphierarchies.py
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,60 @@
"""
contains class for trees consisting of segments of mig tables
"""

import json
from pathlib import Path
from typing import Optional

from loguru import logger
from maus.edifact import EdifactFormat
from pydantic import BaseModel, Field

from migmose.mig.reducednestednachrichtenstruktur import ReducedNestedNachrichtenstruktur


class SegmentGroupHierarchy(BaseModel):
"""
This module contains the same model for a segment group hierarchy as used by the MAUS library,
which can be found here:
(https://github.com/Hochfrequenz/mig_ahb_utility_stack/blob/eae2d84d600f42257cf364190da486aa011a5882/src/maus/models/message_implementation_guide.py#L17). # pylint: disable=line-too-long
However, instead of using the attrs class, we are creating a Pydantic class.
"""

opening_segment: Optional[str] = None
segment_group: Optional[str] = None
sub_hierarchy: list[Optional["SegmentGroupHierarchy"]] = Field(default_factory=list)

@classmethod
def create_segmentgroup_hierarchy(
cls, reduced_nested_nachrichtenstruktur: ReducedNestedNachrichtenstruktur
) -> "SegmentGroupHierarchy":
"""init Segmentrgroup Hierarchy"""
segment_group: Optional[str] = None
opening_segment: Optional[str] = None
if reduced_nested_nachrichtenstruktur.header_linie is not None:
segment_group = reduced_nested_nachrichtenstruktur.header_linie.bezeichnung
if (
any(reduced_nested_nachrichtenstruktur.segmente)
and reduced_nested_nachrichtenstruktur.segmente[0] is not None
):
opening_segment = reduced_nested_nachrichtenstruktur.segmente[0].bezeichnung
sub_hierarchy: list[Optional["SegmentGroupHierarchy"]] = []
for sub_segmentgroup in reduced_nested_nachrichtenstruktur.segmentgruppen:
if sub_segmentgroup is not None:
sub_hierarchy.append(cls.create_segmentgroup_hierarchy(sub_segmentgroup))

return SegmentGroupHierarchy(
segment_group=segment_group, opening_segment=opening_segment, sub_hierarchy=sub_hierarchy
)

def to_json(self, message_type: EdifactFormat, output_dir: Path) -> None:
"""
writes the reduced NestedNachrichtenstruktur as json
"""
output_dir.mkdir(parents=True, exist_ok=True)
file_path = output_dir.joinpath("sgh.json")
structured_json = self.model_dump(mode="json")
with open(file_path, "w", encoding="utf-8") as json_file:
json.dump(structured_json, json_file, indent=4)
logger.info("Wrote segmentgroup hierarchy (sgh.json) for {} to {}", message_type, file_path)
66 changes: 66 additions & 0 deletions unittests/test_expected_data/FV2310/IFTSTA/sgh.json
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,66 @@
{
"opening_segment": "UNH",
"segment_group": null,
"sub_hierarchy": [
{
"opening_segment": "NAD",
"segment_group": "SG1",
"sub_hierarchy": [
{
"opening_segment": "CTA",
"segment_group": "SG2",
"sub_hierarchy": []
}
]
},
{
"opening_segment": "EQD",
"segment_group": "SG4",
"sub_hierarchy": [
{
"opening_segment": "LOC",
"segment_group": "SG6",
"sub_hierarchy": []
},
{
"opening_segment": "STS",
"segment_group": "SG7",
"sub_hierarchy": []
}
]
},
{
"opening_segment": "CNI",
"segment_group": "SG14",
"sub_hierarchy": [
{
"opening_segment": "STS",
"segment_group": "SG15",
"sub_hierarchy": [
{
"opening_segment": "EFI",
"segment_group": "SG16",
"sub_hierarchy": []
},
{
"opening_segment": "NAD",
"segment_group": "SG17",
"sub_hierarchy": [
{
"opening_segment": "CTA",
"segment_group": "SG18",
"sub_hierarchy": []
}
]
},
{
"opening_segment": "GID",
"segment_group": "SG25",
"sub_hierarchy": []
}
]
}
]
}
]
}
22 changes: 22 additions & 0 deletions unittests/test_expected_data/FV2310/ORDCHG/sgh.json
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,22 @@
{
"opening_segment": "UNH",
"segment_group": null,
"sub_hierarchy": [
{
"opening_segment": "RFF",
"segment_group": "SG1",
"sub_hierarchy": []
},
{
"opening_segment": "NAD",
"segment_group": "SG3",
"sub_hierarchy": [
{
"opening_segment": "CTA",
"segment_group": "SG6",
"sub_hierarchy": []
}
]
}
]
}
59 changes: 59 additions & 0 deletions unittests/test_expected_data/FV2310/UTILMD/sgh.json
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,59 @@
{
"opening_segment": "UNH",
"segment_group": null,
"sub_hierarchy": [
{
"opening_segment": "RFF",
"segment_group": "SG1",
"sub_hierarchy": []
},
{
"opening_segment": "NAD",
"segment_group": "SG2",
"sub_hierarchy": [
{
"opening_segment": "CTA",
"segment_group": "SG3",
"sub_hierarchy": []
}
]
},
{
"opening_segment": "IDE",
"segment_group": "SG4",
"sub_hierarchy": [
{
"opening_segment": "LOC",
"segment_group": "SG5",
"sub_hierarchy": []
},
{
"opening_segment": "RFF",
"segment_group": "SG6",
"sub_hierarchy": []
},
{
"opening_segment": "SEQ",
"segment_group": "SG8",
"sub_hierarchy": [
{
"opening_segment": "QTY",
"segment_group": "SG9",
"sub_hierarchy": []
},
{
"opening_segment": "CCI",
"segment_group": "SG10",
"sub_hierarchy": []
}
]
},
{
"opening_segment": "NAD",
"segment_group": "SG12",
"sub_hierarchy": []
}
]
}
]
}
54 changes: 54 additions & 0 deletions unittests/test_segment_group_hierarchies.py
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,54 @@
import json

import pytest
from maus.edifact import EdifactFormat, EdifactFormatVersion
from maus.models.message_implementation_guide import SegmentGroupHierarchy as MausSGH
from maus.models.message_implementation_guide import SegmentGroupHierarchySchema

from migmose.mig.nachrichtenstrukturtabelle import NachrichtenstrukturTabelle
from migmose.mig.nestednachrichtenstruktur import NestedNachrichtenstruktur
from migmose.mig.reducednestednachrichtenstruktur import ReducedNestedNachrichtenstruktur
from migmose.mig.segmentgrouphierarchies import SegmentGroupHierarchy
from migmose.parsing import find_file_to_format, parse_raw_nachrichtenstrukturzeile
from unittests import expected_output_dir, path_to_test_edi_energy_mirror_repo


class TestSegmentGroupHierarchy:
"""test class for create_reduced_nested_nachrichtenstruktur"""

@pytest.mark.parametrize(
"message_format",
[
pytest.param(EdifactFormat.ORDCHG, id="ORDCHG"),
pytest.param(EdifactFormat.UTILMD, id="UTILMD"),
pytest.param(EdifactFormat.IFTSTA, id="IFTSTA"),
],
)
def test_create_create_segmentgroup_hierarchy(self, message_format: EdifactFormat, tmp_path):
"""test if the reduced nested nachrichtenstruktur is created correctly"""
file_path_dict = find_file_to_format(
[message_format], path_to_test_edi_energy_mirror_repo, EdifactFormatVersion.FV2310
)
file_path = file_path_dict[message_format]
raw_lines = parse_raw_nachrichtenstrukturzeile(file_path)
nachrichtenstrukturtabelle = NachrichtenstrukturTabelle.create_nachrichtenstruktur_tabelle(raw_lines)
nested_nachrichtenstruktur, _ = NestedNachrichtenstruktur.create_nested_nachrichtenstruktur(
nachrichtenstrukturtabelle
)
reduced_nested_nachrichtenstruktur = ReducedNestedNachrichtenstruktur.create_reduced_nested_nachrichtenstruktur(
nested_nachrichtenstruktur
)
sgh = SegmentGroupHierarchy.create_segmentgroup_hierarchy(reduced_nested_nachrichtenstruktur)
sgh.to_json(message_format, tmp_path)
with open(tmp_path / "sgh.json", "r", encoding="utf-8") as file1:
actual_sgh_json = json.load(file1)
with open(
expected_output_dir / EdifactFormatVersion.FV2310 / message_format / "sgh.json",
"r",
encoding="utf-8",
) as sgh_file:
expected_sgh_json = json.load(sgh_file)
assert actual_sgh_json == expected_sgh_json
with open(tmp_path / "sgh.json", "r", encoding="utf-8") as sgh_file:
sgh = SegmentGroupHierarchySchema().loads(sgh_file.read())
assert isinstance(sgh, MausSGH)

0 comments on commit 44b529c

Please sign in to comment.