diff --git a/dev_requirements/requirements-linting.txt b/dev_requirements/requirements-linting.txt index 22d0c843..1093f83f 100644 --- a/dev_requirements/requirements-linting.txt +++ b/dev_requirements/requirements-linting.txt @@ -17,5 +17,5 @@ platformdirs==4.2.2 # via pylint pylint==3.2.6 # via -r dev_requirements/requirements-linting.in -tomlkit==0.12.5 +tomlkit==0.13.2 # via pylint diff --git a/json_schemas/generate_json_schemas.py b/json_schemas/generate_json_schemas.py index d789a6ad..d1b92d16 100644 --- a/json_schemas/generate_json_schemas.py +++ b/json_schemas/generate_json_schemas.py @@ -7,11 +7,10 @@ import pathlib from typing import List, Type +from kohlrahbi.models.anwendungshandbuch import DeepAnwendungshandbuchSchema from marshmallow import Schema from marshmallow_jsonschema import JSONSchema # type:ignore[import] -from maus.models.anwendungshandbuch import DeepAnwendungshandbuchSchema - schema_types: List[Type[Schema]] = [ DeepAnwendungshandbuchSchema # add other schemas as you like diff --git a/pyproject.toml b/pyproject.toml index 422e27f3..d63f5a39 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -28,7 +28,8 @@ dependencies = [ "attrs>=22.1.0", "marshmallow>=3.18.0", "more_itertools", - "efoli>=0.0.3" + "efoli>=0.0.3", + "kohlrahbi>=1.6.0" # add everything you add in requirements.in here ] dynamic = ["readme", "version"] diff --git a/requirements.in b/requirements.in index 4ffcaa2e..8e1ea2ea 100644 --- a/requirements.in +++ b/requirements.in @@ -6,4 +6,5 @@ xmltodict click lark efoli>=0.0.3 +kohlrahbi>=1.6.0 # whenever adding something here, don't forget to also add it to pyproject.toml dependencies diff --git a/requirements.txt b/requirements.txt index 20b50e9a..baf4b07c 100644 --- a/requirements.txt +++ b/requirements.txt @@ -2,25 +2,67 @@ # This file is autogenerated by pip-compile with Python 3.11 # by the following command: # -# pip-compile requirements.in +# pip-compile pyproject.toml # +annotated-types==0.7.0 + # via pydantic attrs==23.2.0 - # via -r requirements.in + # via maus (pyproject.toml) click==8.1.7 - # via -r requirements.in + # via kohlrahbi +colorama==0.4.6 + # via + # click + # colorlog +colorlog==6.8.2 + # via kohlrahbi efoli==1.1.0 - # via -r requirements.in -lark==1.2.2 - # via -r requirements.in + # via + # kohlrahbi + # maus (pyproject.toml) +et-xmlfile==1.1.0 + # via openpyxl +kohlrahbi==1.6.0 + # via maus (pyproject.toml) lxml==5.3.0 - # via -r requirements.in + # via python-docx marshmallow==3.21.3 - # via -r requirements.in -more-itertools==10.3.0 - # via -r requirements.in + # via maus (pyproject.toml) +more-itertools==10.4.0 + # via + # kohlrahbi + # maus (pyproject.toml) +numpy==2.1.0 + # via pandas +openpyxl==3.1.5 + # via kohlrahbi packaging==24.0 # via marshmallow +pandas==2.2.2 + # via kohlrahbi +pydantic==2.8.2 + # via kohlrahbi +pydantic-core==2.20.1 + # via pydantic +python-dateutil==2.9.0.post0 + # via pandas +python-docx==1.1.2 + # via kohlrahbi pytz==2024.1 - # via efoli -xmltodict==0.13.0 - # via -r requirements.in + # via + # efoli + # kohlrahbi + # pandas +six==1.16.0 + # via python-dateutil +tomlkit==0.13.2 + # via kohlrahbi +typing-extensions==4.12.2 + # via + # pydantic + # pydantic-core + # python-docx +tzdata==2024.1 + # via pandas +xlsxwriter==3.2.0 + # via kohlrahbi diff --git a/src/maus/cli.py b/src/maus/cli.py index 45166a91..a4c52a54 100644 --- a/src/maus/cli.py +++ b/src/maus/cli.py @@ -12,12 +12,9 @@ # click is only an optional dependency when maus is used as CLI tool raise +from kohlrahbi.models.anwendungshandbuch import DeepAnwendungshandbuch, FlatAnwendungshandbuch + from maus.mig_ahb_matching import to_deep_ahb -from maus.models.anwendungshandbuch import ( - DeepAnwendungshandbuch, - DeepAnwendungshandbuchSchema, - FlatAnwendungshandbuchSchema, -) from maus.models.message_implementation_guide import SegmentGroupHierarchySchema from maus.reader.mig_xml_reader import MigXmlReader @@ -79,7 +76,7 @@ def main( raise click.Abort() with open(flat_ahb_path, "r", encoding="utf-8") as flat_ahb_file: - flat_ahb = FlatAnwendungshandbuchSchema().load(json.load(flat_ahb_file)) + flat_ahb = FlatAnwendungshandbuch.model_validate(json.load(flat_ahb_file)) with open(sgh_path, "r", encoding="utf-8") as sgh_file: sgh = SegmentGroupHierarchySchema().loads(sgh_file.read()) @@ -93,14 +90,14 @@ def main( raise click.Abort() if output_path is not None: - maus_dict = DeepAnwendungshandbuchSchema().dump(maus) + maus_dict = maus.model_dump(mode="json") with open(output_path, "w", encoding="utf-8") as maus_file: json.dump(maus_dict, maus_file, indent=2, ensure_ascii=False, sort_keys=True) if check_path is not None: with open(check_path, "r", encoding="utf-8") as maus_file: - expected_maus: DeepAnwendungshandbuch = DeepAnwendungshandbuchSchema().loads(maus_file.read()) + expected_maus: DeepAnwendungshandbuch = DeepAnwendungshandbuch.model_validate_json(maus_file.read()) # reset the line index to make the comparison work # this is fine cause there is no logic built on top of the line index diff --git a/src/maus/maus_provider.py b/src/maus/maus_provider.py index cc108f29..3ff28541 100644 --- a/src/maus/maus_provider.py +++ b/src/maus/maus_provider.py @@ -10,8 +10,9 @@ from pathlib import Path from typing import Optional +from kohlrahbi.models.anwendungshandbuch import DeepAnwendungshandbuch + from maus.edifact import EdifactFormat, EdifactFormatVersion -from maus.models.anwendungshandbuch import DeepAnwendungshandbuch, DeepAnwendungshandbuchSchema # pylint:disable=too-few-public-methods @@ -60,7 +61,7 @@ def get_maus( try: with open(full_path, "r", encoding=self._encoding) as maus_infile: file_content_json = json.load(maus_infile) - maus = DeepAnwendungshandbuchSchema().load(file_content_json) + maus = DeepAnwendungshandbuch.model_validate(file_content_json) except FileNotFoundError: return None return maus diff --git a/src/maus/mig_ahb_matching.py b/src/maus/mig_ahb_matching.py index e8c3fb26..58a42df3 100644 --- a/src/maus/mig_ahb_matching.py +++ b/src/maus/mig_ahb_matching.py @@ -6,9 +6,9 @@ from itertools import groupby from typing import List, Optional, Sequence, Set, Tuple +from kohlrahbi.models.anwendungshandbuch import _VERSION, AhbLine, DeepAnwendungshandbuch, FlatAnwendungshandbuch from more_itertools import first, first_true, last -from maus.models.anwendungshandbuch import _VERSION, AhbLine, DeepAnwendungshandbuch, FlatAnwendungshandbuch from maus.models.edifact_components import ( DataElement, DataElementFreeText, @@ -98,7 +98,7 @@ def to_deep_ahb( Converts a flat ahb into a nested ahb using the provided segment hierarchy """ result = DeepAnwendungshandbuch(meta=flat_ahb.meta, lines=[]) - result.meta.maus_version = _VERSION + result.meta.maus_version = _VERSION # pylint:disable=assigning-non-slot parent_group_lists: List[List[SegmentGroup]] = [] used_stacks: Set[str] = set() # The following lists are _not_ a view into the lines that are going to follow (hence no name starting with "next") @@ -170,7 +170,7 @@ def to_deep_ahb( used_stacks.add(stack.to_json_path()) append_next_segments_here = segment_group.segments # type:ignore[assignment] if segment_group.discriminator == '$["Dokument"][0]["Nachricht"][0]': - result.lines.append(segment_group) + result.lines.append(segment_group) # pylint:disable=no-member append_next_sg_here = segment_group.segment_groups # type:ignore[assignment] parent_group_lists.append(result.lines) elif position.is_sub_location_of(previous_position): # pylint:disable=used-before-assignment diff --git a/src/maus/models/anwendungshandbuch.py b/src/maus/models/anwendungshandbuch.py deleted file mode 100644 index 243f6f2b..00000000 --- a/src/maus/models/anwendungshandbuch.py +++ /dev/null @@ -1,593 +0,0 @@ -# pylint:disable=too-few-public-methods -""" -This module contains classes that are required to model "Anwendungshandbücher" (AHB). There are two kinds of AHBs: -1. the "flat" AHB is very similar to the flat structure scraped from the official PDF files. It has no deep -structure. -2. the "nested" AHB which contains structural information (e.g. that a segment group is contained in -another segment group) -""" -import re -from typing import Callable, List, Optional, Sequence, Set -from uuid import UUID - -import attr.validators -import attrs -from marshmallow import Schema, fields, post_load # type:ignore[import] -from more_itertools import last, split_when - -from maus.models import _check_that_string_is_not_whitespace_or_empty -from maus.models.edifact_components import ( - DataElementFreeText, - DataElementValuePool, - Segment, - SegmentGroup, - SegmentGroupSchema, -) - -_VERSION = "0.3.0" #: version to be written into the deep ahb - - -# pylint:disable=too-many-instance-attributes -@attrs.define(auto_attribs=True, kw_only=True) -class AhbLine: - """ - An AhbLine is a single line inside the machine-readable, flat AHB. - """ - - guid: Optional[UUID] = attrs.field( - validator=attrs.validators.optional(attrs.validators.instance_of(UUID)) - ) #: optional key - # because the combination (segment group, segment, data element, name) is not guaranteed to be unique - # yes, it's actually that bad already - segment_group_key: Optional[str] = attrs.field( - validator=attrs.validators.optional(validator=attrs.validators.instance_of(str)) - ) - """ the segment group, e.g. 'SG5' """ - - segment_code: Optional[str] = attrs.field( - validator=attrs.validators.optional(validator=attrs.validators.instance_of(str)) - ) - """the segment, e.g. 'IDE'""" - - data_element: Optional[str] = attrs.field( - validator=attrs.validators.optional(validator=attrs.validators.instance_of(str)) - ) - """ the data element ID, e.g. '3224' """ - - segment_id: Optional[str] = attrs.field( - validator=attrs.validators.optional(validator=attrs.validators.instance_of(str)), default=None - ) - """ - the 5 digit segment id, e.g. '00003' for Nachrichten Kopfsegment - This is available since FV2410. - """ - - value_pool_entry: Optional[str] = attrs.field( - validator=attrs.validators.optional(attrs.validators.instance_of(str)) - ) - """ one of (possible multiple) allowed values, e.g. 'E01' or '293' """ - - name: Optional[str] = attrs.field(validator=attrs.validators.optional(validator=attrs.validators.instance_of(str))) - """the name, e.g. 'Meldepunkt'. It can be both the description of a field but also its meaning""" - - # Check the unittest test_csv_file_reading_11042 to see the different values of name. It's not only the grey fields - # where user input is expected but also the meanings / values of value pool entries. This means the exact meaning of - # name can only be determined in the context in which it is used. This is one of many shortcoming of the current AHB - # structure: Things in the same column don't necessarily mean the same thing. - ahb_expression: Optional[str] = attrs.field( - validator=attrs.validators.optional( - validator=attrs.validators.and_( - attrs.validators.instance_of(str), _check_that_string_is_not_whitespace_or_empty - ) - ) - ) - """a requirement indicator + an optional condition ("ahb expression"), e.g. 'Muss [123] O [456]' """ - # note: to parse expressions from AHBs consider using AHBicht: https://github.com/Hochfrequenz/ahbicht/ - conditions: Optional[str] = attrs.field( - validator=attrs.validators.optional(validator=attrs.validators.instance_of(str)), default=None - ) - """ - The condition text describes the text to the optional condition of the ahb expression. - E.g. '[492] This is a condition text. [999] And this is another one.' - """ - section_name: Optional[str] = attrs.field( - validator=attrs.validators.optional(validator=attrs.validators.instance_of(str)), default=None - ) - """ - The section name describes the purpose of a segment, e.g. "Nachrichten-Kopfsegment" or "Beginn der Nachricht" - """ - index: Optional[int] = attrs.field(validator=attrs.validators.optional(attrs.validators.ge(0)), default=None) - """ - index is a number that describes the position of the AHBLine inside the original PDF- and FlatAnwendungshandbuch. - """ - - def holds_any_information(self) -> bool: - """ - Returns true iff the line holds any information exception for just a GUID. - This is useful to filter out empty lines which are artefacts remaining from the scraping process. - """ - # https://stackoverflow.com/questions/47972143/using-attr-with-pylint - # pylint: disable=no-member - return ( - (self.segment_group_key is not None and len(self.segment_group_key.strip()) > 0) - or (self.segment_code is not None and len(self.segment_code.strip()) > 0) - or (self.data_element is not None and len(self.data_element.strip()) > 0) - or (self.value_pool_entry is not None and len(self.value_pool_entry.strip()) > 0) - or (self.name is not None and len(self.name.strip()) > 0) - or (self.ahb_expression is not None and len(self.ahb_expression.strip()) > 0) - ) - - def get_discriminator(self, include_name: bool = True) -> str: - """ - Generate a unique yet readable discriminator for this given line. - This discriminator is generated just from the line itself without any information on where it occurs. - It does neither know its position inside the AHB nor parent or sub groups. - """ - result: str - if self.segment_group_key: - result = f"{self.segment_group_key}->" - else: - result = "" - result += f"{self.segment_code}->{self.data_element}" - if self.name and include_name: - result += f" ({self.name})" - return result - - -class AhbLineSchema(Schema): - """ - A schema to (de-)serialize :class:`.AhbLine` - """ - - guid = fields.UUID(required=False, load_default=None) - segment_group_key = fields.String(required=False, load_default=None) - segment_code = fields.String(required=False, load_default=None) - segment_id = fields.String(required=False, load_default=None) - data_element = fields.String(required=False, load_default=None) - value_pool_entry = fields.String(required=False, load_default=None) - name = fields.String(required=False, load_default=None) - ahb_expression = fields.String(required=False, load_default=None) - conditions = fields.String(required=False, load_default=None) - section_name = fields.String(required=False, load_default=None) - index = fields.Int(required=False, load_default=None, dump_default=None) - - # pylint:disable=unused-argument - @post_load - def deserialize(self, data, **kwargs) -> AhbLine: - """ - Converts the barely typed data dictionary into an actual :class:`.AhbLine` - """ - return AhbLine(**data) - - -@attrs.define(auto_attribs=True, kw_only=True) -class AhbMetaInformation: - """ - Meta information about an AHB like e.g. its title, Prüfidentifikator, possible sender and receiver roles - """ - - pruefidentifikator: str #: identifies the message type (within a fixed format version) e.g. "11042" or "13012" - # there's more to come but for now we'll leave it as is, because we're just in a proof of concept phase - maus_version: Optional[str] = attrs.field( - validator=attrs.validators.optional(attrs.validators.instance_of(str)), # type:ignore[arg-type] - # https://github.com/Hochfrequenz/mig_ahb_utility_stack/issues/221 - default=_VERSION, - ) - """ - semantic version of maus used to create this document - """ - description: Optional[str] = attrs.field( - validator=attrs.validators.optional( - validator=attrs.validators.and_( - attrs.validators.instance_of(str), - _check_that_string_is_not_whitespace_or_empty, - ), - ), - default=None, - ) - """ - an optional description of the purpose of the pruefidentifikator; e.g. 'Anmeldung MSB' for 11042 - """ - direction: Optional[str] = attrs.field( - validator=attrs.validators.optional( - validator=attrs.validators.and_( - attrs.validators.instance_of(str), - _check_that_string_is_not_whitespace_or_empty, - ) - ), - default=None, - ) - """ - a stringly typed description of the roles of sender and receiver of the message - (the row name in the AHB is 'Kommunikation von'); - e.g. 'MSB an NB' for 11042 - """ - - -class AhbMetaInformationSchema(Schema): - """ - A schema to (de-)serialize :class:`.AhbMetaInformation` - """ - - pruefidentifikator = fields.String(required=True) - maus_version = fields.String(required=False, allow_none=True, dump_default=_VERSION) - description = fields.String(required=False, allow_none=True, dump_default=None) - direction = fields.String(required=False, allow_none=True, dump_default=None) - - # pylint:disable=unused-argument - @post_load - def deserialize(self, data, **kwargs) -> AhbMetaInformation: - """ - Converts the barely typed data dictionary into an actual :class:`.AhbMetaInformation` - """ - return AhbMetaInformation(**data) - - -def _remove_grouped_ahb_lines_containing_section_name( - grouped_ahb_lines: List[List[AhbLine]], section_name: str -) -> List[List[AhbLine]]: - """ - Removes all groups of ahb lines that contain a line with the given section name and returns a new list instance. - """ - return [goal for goal in grouped_ahb_lines if any(ahbl.section_name == section_name for ahbl in goal)] - - -# pylint:disable=unused-argument -def _check_that_nearly_all_lines_have_a_segment_group(instance, attribute, value: List[AhbLine]): - """ - Loops over all provided ahb lines and checks that only at the beginning and the end there are lines without a - segment group. In between every line has to have a segment group specified. This is necessary for the navigation - to work because it primarily focuses and relies on correct SG information in the lines . - """ - switches_from_no_sg_to_sg = list( - split_when(value, lambda x, y: x.segment_group_key is None and y.segment_group_key is not None) - ) - switches_from_no_sg_to_sg = _remove_grouped_ahb_lines_containing_section_name( - grouped_ahb_lines=switches_from_no_sg_to_sg, section_name="Abschnitts-Kontrollsegment" - ) - if len(switches_from_no_sg_to_sg) > 2: - raise ValueError(f"There is a None segment group in line {last(switches_from_no_sg_to_sg[1])}") - - -_data_element_pattern = re.compile(r"^\d{4}$|^\d{5}$|^[A-Za-z]+\d{4}$") - - -# pylint:disable=unused-argument -def _check_that_line_has_either_none_or_d4_data_element(instance, attribute, value: AhbLine): - """ - checks that the given line has either a None data element or a data element that matches - \\d{4},\\ d{5} or [A-Za-z]+\\d{4}. - AhbLine(data_element = 0001), AhbLine(data_element = 00001) or AhbLine(data_element = R0001) - """ - if value.data_element is None: - return - match = _data_element_pattern.match(value.data_element) - if match is None: - raise ValueError(f"The data_element '{value.data_element}' does not match {_data_element_pattern}") - - -_segment_group_key_pattern = re.compile(r"^SG\d+$") - - -# pylint:disable=unused-argument -def _check_that_line_has_either_none_or_matching_sg(instance, attribute, value: AhbLine): - """ - checks that the given line has a segment group key that is either None (for root) or matches SG\\d+ - """ - if value.segment_group_key is None: - return - match = _segment_group_key_pattern.match(value.segment_group_key) - if match is None: - raise ValueError( - f"The segment_group_key '{value.segment_group_key}' does not match {_segment_group_key_pattern}" - ) - - -_segment_code_pattern = re.compile(r"^[A-Z]+$") - - -# pylint:disable=unused-argument -def _check_that_line_has_either_none_az_segment_code(instance, attribute, value: AhbLine): - """ - checks that the given line has either a None segment code or a segment code that consists of all upper letters - """ - if value.segment_code is None: - return - match = _segment_code_pattern.match(value.segment_code) - if match is None: - raise ValueError(f"The segment_code '{value.segment_code}' does not match {_segment_code_pattern}") - - -@attrs.define(auto_attribs=True, kw_only=True) -class FlatAnwendungshandbuch: - """ - A flat Anwendungshandbuch (AHB) models an Anwendungshandbuch as combination of some meta data and an ordered list - of `.class:`.FlatAhbLine s. Basically a flat Anwendungshandbuch is the result of a simple scraping approach. - You can create instances of this class without knowing anything about the "under the hood" structure of AHBs or MIGs - """ - - meta: AhbMetaInformation = attrs.field(validator=attrs.validators.instance_of(AhbMetaInformation)) - """information about this AHB""" - - lines: List[AhbLine] = attrs.field( - validator=attrs.validators.deep_iterable( - member_validator=attrs.validators.and_( - attrs.validators.instance_of(AhbLine), - # The following checks are not baked into the AhbLine class itself, because they might be initialized - # with raw data that do not yet obey these strict validations. But as soon as we bundle them in a - # FlatAnwendungshandbuch, the sanitizations shall be applied. - _check_that_line_has_either_none_or_d4_data_element, - _check_that_line_has_either_none_or_matching_sg, - _check_that_line_has_either_none_az_segment_code, - ), - iterable_validator=attrs.validators.and_( - attrs.validators.instance_of(list), - _check_that_nearly_all_lines_have_a_segment_group, - ), - ) - ) #: ordered list lines as they occur in the AHB - - def get_segment_groups(self) -> List[Optional[str]]: - """ - :return: a set with all segment groups in this AHB in the order in which they occur - """ - return FlatAnwendungshandbuch._get_available_segment_groups(self.lines) - - @staticmethod - def _get_available_segment_groups(lines: List[AhbLine]) -> List[Optional[str]]: - """ - extracts the distinct segment groups from a list of ahb lines - :param lines: - :return: distinct segment groups, including None in the order in which they occur - """ - # this code is in a static method to make it easily accessible for fine grained unit testing - result: List[Optional[str]] = [] - for line in lines: - if line.segment_group_key not in result: - # an "in" check against a set would be faster but we want to preserve both order and readability - result.append(line.segment_group_key) - return result - - def sort_lines_by_segment_groups(self): - """ - sorts lines by segment groups while preserving the order inside the groups and the order between the groups. - """ - self.lines = FlatAnwendungshandbuch._sorted_lines_by_segment_groups(self.lines, self.get_segment_groups()) - - @staticmethod - def _sorted_lines_by_segment_groups(ahb_lines: Sequence[AhbLine], sg_order: List[Optional[str]]) -> List[AhbLine]: - """ - Calls sorted(...) on the provided list and returns a new list. - Its purpose is, that if a segment group in the AHB (read from top to bottom in the flat ahb/pdf) is interrupted - by other segment groups, the lines belonging to the same group will be next to each other. - This is useful to later use a groupby aggregation that only returns one group key per segment group. - - The sort is stable such that the existing order inside the segment groups is maintained. - - Note that this also means, that the order of the return lines no longer the same as in the flat AHB. - - If you provide this function a list: - [ - (SG2, Foo), - (SG2, Bar), - (SG3, ABC), - (SG3, DEF), - (SG2, SomethingElse) - ] - The result will be: - [ - (SG2, Foo), - (SG2, Bar), - (SG2, SomethingElse) - (SG3, ABC), - (SG3, DEF) - ] - - See the unittests for details. - """ - - # this code is in a static method to make it easily accessible for fine-grained unit testing - result: List[AhbLine] = sorted(ahb_lines, key=lambda x: x.segment_group_key or "") - result.sort(key=lambda ahb_line: sg_order.index(ahb_line.segment_group_key)) - return result - - -class FlatAnwendungshandbuchSchema(Schema): - """ - A schema to (de-)serialize :class:`.FlatAnwendungshandbuch` - """ - - meta = fields.Nested(AhbMetaInformationSchema) - lines = fields.List(fields.Nested(AhbLineSchema)) - - # pylint:disable=unused-argument - @post_load - def deserialize(self, data, **kwargs) -> FlatAnwendungshandbuch: - """ - Converts the barely typed data dictionary into an actual :class:`.FlatAnwendungshandbuch` - """ - return FlatAnwendungshandbuch(**data) - - -@attrs.define(auto_attribs=True, kw_only=True) -class DeepAhbInputReplacement: - """ - A container class that models replacements of inputs in the DeepAnwendungshandbuch - """ - - #: true iff a replacement is applicable - replacement_found: bool = attrs.field(validator=attrs.validators.instance_of(bool)) - input_replacement: Optional[str] = attrs.field( - validator=attrs.validators.optional(attr.validators.instance_of(str)) - ) - """ - The replacement for entered_input itself. Note that the replacement may be None even if a replacement is found. - This implies, that you must always check for replacement_found is True first and then, iff true, replace with the - replacement, even if it may be None/null. - """ - - -@attrs.define(auto_attribs=True, kw_only=True) -class DeepAnwendungshandbuch: - """ - The data of the AHB nested as described in the MIG. - """ - - meta: AhbMetaInformation = attrs.field(validator=attrs.validators.instance_of(AhbMetaInformation)) - """information about this AHB""" - - lines: List[SegmentGroup] = attrs.field( - validator=attrs.validators.deep_iterable( - member_validator=attrs.validators.instance_of(SegmentGroup), - iterable_validator=attrs.validators.instance_of(list), - ) - ) #: the nested data - - def reset_ahb_line_index(self) -> None: - """ - reset the ahb line index for all lines in the DeepAnwendungshandbuch - :return: nothing; modifies the instance - """ - for line in self.lines: - line.reset_ahb_line_index() - - @staticmethod - def _query_segment_group( - segment_group: SegmentGroup, predicate: Callable[[SegmentGroup], bool] - ) -> List[SegmentGroup]: - """ - recursively search for a segment group that matches the predicate - :return: return empty list if nothing was found, the matching segment groups otherwise - """ - result: List[SegmentGroup] = [] - if predicate(segment_group): - result.append(segment_group) - if segment_group.segment_groups is not None: - for sub_group in segment_group.segment_groups: - sub_result = DeepAnwendungshandbuch._query_segment_group(sub_group, predicate) - result += sub_result - return result - - def find_segment_groups(self, predicate: Callable[[SegmentGroup], bool]) -> List[SegmentGroup]: - """ - recursively search for segment group in this ahb that meets the predicate. - :return: list of segment groups that match the predicate; empty list otherwise - """ - result: List[SegmentGroup] = [] - for line in self.lines: - if line.segment_groups is not None: - for segment_group in line.segment_groups: - result += DeepAnwendungshandbuch._query_segment_group(segment_group, predicate) - for line_result in DeepAnwendungshandbuch._query_segment_group(line, predicate): - if line_result not in result: - result.append(line_result) - return result - - def find_segments( - self, - group_predicate: Callable[[SegmentGroup], bool] = lambda _: True, - segment_predicate: Callable[[Segment], bool] = lambda _: True, - ) -> List[Segment]: - """ - recursively search for segment characterised by the segment_predicate inside a group characterised by the - group_predicate. - :return: list of matching segments, empty list if nothing was found - """ - result: List[Segment] = [] - for segment_group in self.find_segment_groups(group_predicate): - result += segment_group.find_segments(segment_predicate) - for line in self.lines: - if line.segments is not None: - for segment in line.segments: - if segment_predicate(segment): - result.append(segment) - return result - - def replace_inputs_based_on_discriminator(self, replacement_func: Callable[[str], DeepAhbInputReplacement]) -> None: - """ - Replace all the entered_inputs in the entire DeepAnwendungshandbuch using the given replacement_func. - Note that this modifies this DeepAnwendungshandbuch instance (self). - """ - _replace_inputs_based_on_discriminator(self.lines, replacement_func) - - def get_all_value_pools(self) -> List[DataElementValuePool]: - """ - recursively find all value pools in the deep ahb - :return: a list of all value pools - """ - result: List[DataElementValuePool] = [] - added_discriminators: Set[Optional[str]] = set() - # checks like "str in set" are way faster than "value pool in list" - - def add_to_result(value_pool: DataElementValuePool): - if value_pool.discriminator in added_discriminators: - # assumption: the discriminator is truly unique in an AHB - return - if value_pool not in result: - added_discriminators.add(value_pool.discriminator) - result.append(value_pool) - - for segment in self.find_segments(): - for sub_result in segment.get_all_value_pools(): - add_to_result(sub_result) - for segment_group in self.find_segment_groups(lambda _: True): - for sub_result in segment_group.get_all_value_pools(): - add_to_result(sub_result) - return list(result) - - def get_all_expressions(self) -> List[str]: - """ - recursively iterate through the deep ahb and return all distinct expressions found - """ - result: Set[str] = set() - for segment in self.find_segments(): - if segment.ahb_expression: - result.add(segment.ahb_expression) - for data_element in segment.data_elements: - if isinstance(data_element, DataElementFreeText): - if data_element.ahb_expression: - result.add(data_element.ahb_expression) - elif isinstance(data_element, DataElementValuePool): - for value_pool_entry in data_element.value_pool: - if value_pool_entry.ahb_expression: - result.add(value_pool_entry.ahb_expression) - for segment_group in self.find_segment_groups(lambda _: True): - if segment_group.ahb_expression: - result.add(segment_group.ahb_expression) - return sorted(result) - - -def _replace_inputs_based_on_discriminator( - segment_groups: List[SegmentGroup], replacement_func: Callable[[str], DeepAhbInputReplacement] -) -> None: - """ - Replace all the entered_inputs in the entire list of segment groups using the given replacement_func. - """ - for segment_group in segment_groups: - if segment_group.segment_groups is not None: - _replace_inputs_based_on_discriminator(segment_group.segment_groups, replacement_func) - if segment_group.segments is None: - continue - for segment in segment_group.segments: - for data_element in segment.data_elements: - if data_element.discriminator is not None: - replacement_result = replacement_func(data_element.discriminator) - if replacement_result.replacement_found is True: - data_element.entered_input = replacement_result.input_replacement - - -class DeepAnwendungshandbuchSchema(Schema): - """ - A schema to (de-)serialize :class:`.DeepAnwendungshandbuch` - """ - - meta = fields.Nested(AhbMetaInformationSchema) - lines = fields.List(fields.Nested(SegmentGroupSchema)) - - # pylint:disable=unused-argument - @post_load - def deserialize(self, data, **kwargs) -> DeepAnwendungshandbuch: - """ - Converts the barely typed data dictionary into an actual :class:`.DeepAnwendungshandbuch` - """ - return DeepAnwendungshandbuch(**data) diff --git a/src/maus/navigation.py b/src/maus/navigation.py index dc413bfd..c4bc5198 100644 --- a/src/maus/navigation.py +++ b/src/maus/navigation.py @@ -9,9 +9,9 @@ from typing import Callable, List, Optional, Tuple, TypeVar, Union, overload import attrs +from kohlrahbi.models.anwendungshandbuch import AhbLine from more_itertools import first_true, last -from maus.models.anwendungshandbuch import AhbLine from maus.models.message_implementation_guide import SegmentGroupHierarchy T = TypeVar("T") diff --git a/src/maus/reader/flat_ahb_reader.py b/src/maus/reader/flat_ahb_reader.py index 5817b4da..709f304f 100644 --- a/src/maus/reader/flat_ahb_reader.py +++ b/src/maus/reader/flat_ahb_reader.py @@ -10,7 +10,8 @@ from pathlib import Path from typing import Dict, List, Literal, Optional, Sequence, Set, TextIO, Tuple, overload -from maus.models.anwendungshandbuch import _VERSION, AhbLine, AhbMetaInformation, FlatAnwendungshandbuch +from kohlrahbi.models.anwendungshandbuch import _VERSION, AhbLine, AhbMetaInformation, FlatAnwendungshandbuch + from maus.models.edifact_components import gabi_edifact_qualifier_pattern _pruefi_pattern = re.compile(r"^\d{5}$") #: five digits diff --git a/src/maus/transformation.py b/src/maus/transformation.py index a412bc97..2a1b4aab 100644 --- a/src/maus/transformation.py +++ b/src/maus/transformation.py @@ -25,7 +25,8 @@ import asyncio from typing import Any, Awaitable, Dict, List, Mapping, Optional, Protocol, TypeVar -from maus.models.anwendungshandbuch import DataElementValuePool, DeepAnwendungshandbuch +from kohlrahbi.models.anwendungshandbuch import DataElementValuePool, DeepAnwendungshandbuch + from maus.models.edifact_components import ValuePoolEntry EdifactData = TypeVar("EdifactData") diff --git a/tests/integration_tests/helpers.py b/tests/integration_tests/helpers.py index 46889908..d7e9b41e 100644 --- a/tests/integration_tests/helpers.py +++ b/tests/integration_tests/helpers.py @@ -4,14 +4,14 @@ from typing import Optional import attrs - -from maus.mig_ahb_matching import to_deep_ahb -from maus.models.anwendungshandbuch import ( +from kohlrahbi.models.anwendungshandbuch import ( DeepAnwendungshandbuch, DeepAnwendungshandbuchSchema, FlatAnwendungshandbuch, FlatAnwendungshandbuchSchema, ) + +from maus.mig_ahb_matching import to_deep_ahb from maus.models.message_implementation_guide import SegmentGroupHierarchy, SegmentGroupHierarchySchema from maus.reader.flat_ahb_reader import FlatAhbCsvReader from maus.reader.mig_xml_reader import MigXmlReader diff --git a/tests/unit_tests/example_data_11042.py b/tests/unit_tests/example_data_11042.py index 1cd3e133..d9f4c94d 100644 --- a/tests/unit_tests/example_data_11042.py +++ b/tests/unit_tests/example_data_11042.py @@ -4,7 +4,8 @@ from uuid import UUID -from maus.models.anwendungshandbuch import AhbLine, AhbMetaInformation, FlatAnwendungshandbuch +from kohlrahbi.models.anwendungshandbuch import AhbLine, AhbMetaInformation, FlatAnwendungshandbuch + from maus.models.message_implementation_guide import SegmentGroupHierarchy from maus.navigation import AhbLocation, AhbLocationLayer, _DifferentialAhbLineHierarchyChange diff --git a/tests/unit_tests/test_ahb.py b/tests/unit_tests/test_ahb.py deleted file mode 100644 index 6a789f56..00000000 --- a/tests/unit_tests/test_ahb.py +++ /dev/null @@ -1,967 +0,0 @@ -import uuid -from typing import List, Optional, Set - -import pytest # type:ignore[import] -from unit_tests.serialization_test_helper import assert_serialization_roundtrip # type:ignore[import] - -from maus.models.anwendungshandbuch import ( - AhbLine, - AhbLineSchema, - AhbMetaInformation, - AhbMetaInformationSchema, - DeepAhbInputReplacement, - DeepAnwendungshandbuch, - DeepAnwendungshandbuchSchema, - FlatAnwendungshandbuch, - FlatAnwendungshandbuchSchema, -) -from maus.models.edifact_components import ( - DataElementFreeText, - DataElementValuePool, - Segment, - SegmentGroup, - ValuePoolEntry, -) - -meta_x = AhbMetaInformation(pruefidentifikator="11042", maus_version="0.2.3") -meta_y = AhbMetaInformation(pruefidentifikator="11043", maus_version="0.2.3") - -line_x = AhbLine( - ahb_expression="Muss [1] O [2]", - conditions="[1] Test\n[2] Test", - segment_group_key="SG2", - segment_code="NAD", - data_element="3039", - value_pool_entry="E01", - name="MP-ID", - guid=uuid.UUID("12b1a98a-edf5-4177-89e5-a6d8a92c5fdc"), -) -line_y = AhbLine( - ahb_expression="Muss [3] O [4]", - conditions="[3] Test\n[4] Test", - segment_group_key="SG2", - segment_code="NAD", - data_element="3039", - value_pool_entry="E01", - name="MP-ID", - guid=uuid.UUID("12b1a98a-edf5-4177-89e5-a6d8a92c5fdc"), -) - - -class TestAhb: - """ - Tests the behaviour of the Anwendungshandbuch model - """ - - @pytest.mark.parametrize( - "ahb, expected_json_dict", - [ - pytest.param( - AhbMetaInformation(pruefidentifikator="11042", maus_version="0.2.3"), - {"pruefidentifikator": "11042", "maus_version": "0.2.3", "description": None, "direction": None}, - ), - pytest.param( - AhbMetaInformation( - pruefidentifikator="11042", maus_version="0.2.3", description="Foo", direction="bar" - ), - {"pruefidentifikator": "11042", "maus_version": "0.2.3", "description": "Foo", "direction": "bar"}, - ), - ], - ) - def test_ahb_meta_information_serialization_roundtrip(self, ahb: AhbMetaInformation, expected_json_dict: dict): - assert_serialization_roundtrip(ahb, AhbMetaInformationSchema(), expected_json_dict) - - @pytest.mark.parametrize( - "ahb_x, ahb_y, are_equal", - [ - pytest.param(meta_x, AhbMetaInformationSchema().loads(AhbMetaInformationSchema().dumps(meta_x)), True), - pytest.param(meta_x, meta_y, False), - ], - ) - def test_ahb_meta_information_equality(self, ahb_x: AhbMetaInformation, ahb_y: AhbMetaInformation, are_equal: bool): - actual = ahb_x == ahb_y - assert actual == are_equal - - @pytest.mark.parametrize( - "ahb_line, expected_json_dict", - [ - pytest.param( - AhbLine( - ahb_expression="Muss [1] O [2]", - conditions="[1] Test\n[2] Test", - segment_group_key="SG2", - segment_code="NAD", - segment_id="01234", - data_element="3039", - value_pool_entry="E01", - name="MP-ID", - section_name="Foo", - guid=uuid.UUID("12b1a98a-edf5-4177-89e5-a6d8a92c5fdc"), - ), - { - "ahb_expression": "Muss [1] O [2]", - "segment_group_key": "SG2", - "segment_code": "NAD", - "segment_id": "01234", - "data_element": "3039", - "value_pool_entry": "E01", - "name": "MP-ID", - "guid": "12b1a98a-edf5-4177-89e5-a6d8a92c5fdc", - "section_name": "Foo", - "conditions": "[1] Test\n[2] Test", - "index": None, - }, - ), - pytest.param( - AhbLine( - ahb_expression="Muss [1] O [2]", - conditions="[1] Test\n[2] Test", - segment_group_key="SG2", - segment_code="NAD", - segment_id=None, - data_element="3039", - value_pool_entry="E01", - name="MP-ID", - section_name="Foo", - guid=uuid.UUID("12b1a98a-edf5-4177-89e5-a6d8a92c5fdc"), - index=42, - ), - { - "ahb_expression": "Muss [1] O [2]", - "segment_group_key": "SG2", - "segment_code": "NAD", - "segment_id": None, - "data_element": "3039", - "value_pool_entry": "E01", - "name": "MP-ID", - "guid": "12b1a98a-edf5-4177-89e5-a6d8a92c5fdc", - "section_name": "Foo", - "conditions": "[1] Test\n[2] Test", - "index": 42, - }, - ), - ], - ) - def test_ahbline_serialization_roundtrip(self, ahb_line: AhbLine, expected_json_dict: dict): - assert_serialization_roundtrip(ahb_line, AhbLineSchema(), expected_json_dict) - - @pytest.mark.parametrize( - "line_x, line_y, are_equal", - [ - pytest.param(line_x, AhbLineSchema().loads(AhbLineSchema().dumps(line_x)), True), - pytest.param(line_x, line_y, False), - ], - ) - def test_ahb_line_equality(self, line_x: AhbLine, line_y: AhbLine, are_equal: bool): - actual = line_x == line_y - assert actual == are_equal - - @pytest.mark.parametrize( - "ahb_line, expected_holds_any_information", - [ - pytest.param( - AhbLine( - ahb_expression="Muss [1] O [2]", - conditions="[1] Test\n[2] Test", - segment_group_key="SG2", - segment_code="NAD", - data_element="3039", - value_pool_entry="E01", - name="MP-ID", - guid=uuid.UUID("12b1a98a-edf5-4177-89e5-a6d8a92c5fdc"), - ), - True, - ), - pytest.param( - AhbLine( - ahb_expression=None, - segment_group_key=None, - segment_code=None, - data_element=None, - value_pool_entry=None, - name=None, - guid=uuid.UUID("12b1a98a-edf5-4177-89e5-a6d8a92c5fdc"), - ), - False, - ), - pytest.param( - AhbLine( - ahb_expression=None, - segment_group_key="", - segment_code="", - data_element=" ", - value_pool_entry="", - name="", - guid=uuid.UUID("12b1a98a-edf5-4177-89e5-a6d8a92c5fdc"), - ), - False, - ), - ], - ) - def test_ahbline_holds_any_information(self, ahb_line: AhbLine, expected_holds_any_information: bool): - actual = ahb_line.holds_any_information() - assert actual == expected_holds_any_information - - @pytest.mark.parametrize( - "ahb_line, include_name, expected_discriminator", - [ - pytest.param( - AhbLine( - ahb_expression="Muss [1] O [2]", - conditions="[1] Test\n[2] Test", - segment_group_key="SG2", - segment_code="NAD", - data_element="3039", - value_pool_entry="E01", - name="MP-ID", - guid=uuid.UUID("12b1a98a-edf5-4177-89e5-a6d8a92c5fdc"), - ), - True, - "SG2->NAD->3039 (MP-ID)", - ), - pytest.param( - AhbLine( - ahb_expression="Muss [1] O [2]", - conditions="[1] Test\n[2] Test", - segment_group_key="SG2", - segment_code="NAD", - data_element="3039", - value_pool_entry="E01", - name="MP-ID", - guid=uuid.UUID("12b1a98a-edf5-4177-89e5-a6d8a92c5fdc"), - ), - False, - "SG2->NAD->3039", - ), - pytest.param( - AhbLine( - ahb_expression="Muss [1] O [2]", - conditions="[1] Test\n[2] Test", - segment_group_key=None, - segment_code="UNH", - data_element="0062", - guid=uuid.UUID("12b1a98a-edf5-4177-89e5-a6d8a92c5fdc"), - value_pool_entry=None, - name=None, - ), - True, - "UNH->0062", - ), - ], - ) - def test_ahbline_get_discriminator(self, ahb_line: AhbLine, include_name: bool, expected_discriminator: str): - actual = ahb_line.get_discriminator(include_name) - assert actual == expected_discriminator - - @pytest.mark.parametrize( - "flat_ahb, expected_json_dict", - [ - pytest.param( - FlatAnwendungshandbuch( - meta=AhbMetaInformation(pruefidentifikator="11042", maus_version="0.2.3"), - lines=[ - AhbLine( - ahb_expression="Muss [1] O [2]", - conditions="[1] Test\n[2] Test", - segment_group_key="SG2", - segment_code="NAD", - data_element="3039", - value_pool_entry="E01", - name="MP-ID", - guid=uuid.UUID("12b1a98a-edf5-4177-89e5-a6d8a92c5fdc"), - section_name="MP-ID Absender", - ) - ], - ), - { - "meta": { - "pruefidentifikator": "11042", - "maus_version": "0.2.3", - "description": None, - "direction": None, - }, - "lines": [ - { - "ahb_expression": "Muss [1] O [2]", - "segment_group_key": "SG2", - "segment_code": "NAD", - "data_element": "3039", - "value_pool_entry": "E01", - "segment_id": None, - "name": "MP-ID", - "guid": "12b1a98a-edf5-4177-89e5-a6d8a92c5fdc", - "section_name": "MP-ID Absender", - "conditions": "[1] Test\n[2] Test", - "index": None, - } - ], - }, - ), - ], - ) - def test_flatahb_serialization_roundtrip(self, flat_ahb: FlatAnwendungshandbuch, expected_json_dict: dict): - assert_serialization_roundtrip(flat_ahb, FlatAnwendungshandbuchSchema(), expected_json_dict) - - @pytest.mark.parametrize( - "ahb_x, ahb_y, are_equal", - [ - pytest.param( - FlatAnwendungshandbuch( - meta=meta_x, - lines=[line_x], - ), - FlatAnwendungshandbuch( - meta=meta_x, - lines=[line_x], - ), - True, - ), - pytest.param( - FlatAnwendungshandbuch( - meta=meta_x, - lines=[line_x], - ), - FlatAnwendungshandbuch( - meta=meta_y, - lines=[line_x], - ), - False, - id="different meta", - ), - pytest.param( - FlatAnwendungshandbuch( - meta=meta_x, - lines=[line_x], - ), - FlatAnwendungshandbuch( - meta=meta_x, - lines=[line_y], - ), - False, - id="different line", - ), - ], - ) - def test_flat_ahb_equality(self, ahb_x: FlatAnwendungshandbuch, ahb_y: FlatAnwendungshandbuch, are_equal: bool): - actual = ahb_x == ahb_y - assert actual == are_equal - - @pytest.mark.parametrize( - "deep_ahb, expected_json_dict", - [ - pytest.param( - DeepAnwendungshandbuch( - meta=AhbMetaInformation(pruefidentifikator="11042", maus_version="0.2.3"), - lines=[ - SegmentGroup( - ahb_expression="expr A", - discriminator="disc A", - segments=[ - Segment( - ahb_expression="expr B", - discriminator="disc B", - section_name="foo", - data_elements=[ - DataElementValuePool( - value_pool=[ - ValuePoolEntry(qualifier="HELLO", meaning="world", ahb_expression="X"), - ValuePoolEntry(qualifier="MAUS", meaning="rocks", ahb_expression="X"), - ], - discriminator="baz", - data_element_id="0123", - entered_input="HELLO", - ), - DataElementFreeText( - ahb_expression="Muss [1]", - entered_input="Hello Maus", - discriminator="bar", - data_element_id="4567", - ), - ], - ), - ], - segment_groups=[ - SegmentGroup( - discriminator="disc C", - ahb_expression="expr C", - segments=[ - Segment( - section_name="bar", - ahb_expression="expr Y", - discriminator="disc Y", - data_elements=[], - ) - ], - segment_groups=None, - ), - ], - ), - ], - ), - { - "meta": { - "pruefidentifikator": "11042", - "maus_version": "0.2.3", - "description": None, - "direction": None, - }, - "lines": [ - { - "ahb_expression": "expr A", - "discriminator": "disc A", - "segments": [ - { - "section_name": "foo", - "segment_id": None, - "ahb_expression": "expr B", - "discriminator": "disc B", - "data_elements": [ - { - "value_pool": [ - {"qualifier": "HELLO", "meaning": "world", "ahb_expression": "X"}, - {"qualifier": "MAUS", "meaning": "rocks", "ahb_expression": "X"}, - ], - "discriminator": "baz", - "data_element_id": "0123", - "value_type": "VALUE_POOL", - "entered_input": "HELLO", - }, - { - "ahb_expression": "Muss [1]", - "entered_input": "Hello Maus", - "discriminator": "bar", - "data_element_id": "4567", - "value_type": "TEXT", - }, - ], - } - ], - "segment_groups": [ - { - "ahb_expression": "expr C", - "discriminator": "disc C", - "segments": [ - { - "segment_id": None, - "section_name": "bar", - "ahb_expression": "expr Y", - "discriminator": "disc Y", - "data_elements": [], - } - ], - "segment_groups": None, - } - ], - }, - ], - }, - ), - ], - ) - def test_deep_ahb_serialization_roundtrip(self, deep_ahb: DeepAnwendungshandbuch, expected_json_dict: dict): - assert_serialization_roundtrip(deep_ahb, DeepAnwendungshandbuchSchema(), expected_json_dict) - - @pytest.mark.parametrize( - "ahb_x, ahb_y, are_equal", - [ - pytest.param( - DeepAnwendungshandbuch( - meta=AhbMetaInformation(pruefidentifikator="11042"), - lines=[ - SegmentGroup( - ahb_expression="expr A", - discriminator="disc A", - segments=[ - Segment( - ahb_expression="expr B", - discriminator="disc B", - data_elements=[ - DataElementValuePool( - value_pool=[ - ValuePoolEntry(qualifier="HELLO", meaning="world", ahb_expression="X"), - ValuePoolEntry(qualifier="MAUS", meaning="rocks", ahb_expression="X"), - ], - discriminator="baz", - entered_input="HELLO", - data_element_id="0123", - ), - DataElementFreeText( - ahb_expression="Muss [1]", - entered_input="Hello Maus", - discriminator="bar", - data_element_id="4567", - ), - ], - ), - ], - segment_groups=[], - ), - ], - ), - DeepAnwendungshandbuch( - meta=AhbMetaInformation(pruefidentifikator="11042"), - lines=[ - SegmentGroup( - ahb_expression="expr A", - discriminator="disc A", - segments=[ - Segment( - ahb_expression="expr B", - discriminator="disc B", - data_elements=[ - DataElementValuePool( - value_pool=[ - ValuePoolEntry(qualifier="HELLO", meaning="world", ahb_expression="X"), - ValuePoolEntry(qualifier="MAUS", meaning="rocks", ahb_expression="X"), - ], - discriminator="baz", - entered_input="MAUS", - data_element_id="0123", - ), - DataElementFreeText( - ahb_expression="Muss [1]", - entered_input="Hello Mice", - discriminator="bar", - data_element_id="4567", - ), - ], - ), - ], - segment_groups=[], - ), - ], - ), - False, - id="different lines", - ), - ], - ) - def test_deep_ahb_equality(self, ahb_x: DeepAnwendungshandbuch, ahb_y: DeepAnwendungshandbuch, are_equal: bool): - actual = ahb_x == ahb_y - assert actual == are_equal - - @pytest.mark.parametrize( - "lines, expected_group_names", - [ - pytest.param( - [ - AhbLine( - segment_group_key="Foo", - guid=None, - segment_code=None, - data_element=None, - value_pool_entry=None, - ahb_expression=None, - name=None, - ), - AhbLine( - segment_group_key="Bar", - guid=None, - segment_code=None, - data_element=None, - value_pool_entry=None, - ahb_expression=None, - name=None, - ), - AhbLine( - segment_group_key=None, - guid=None, - segment_code=None, - data_element=None, - value_pool_entry=None, - ahb_expression=None, - name=None, - ), - AhbLine( - segment_group_key="Bar", - guid=None, - segment_code=None, - data_element=None, - value_pool_entry=None, - ahb_expression=None, - name=None, - ), - ], - ["Foo", "Bar", None], - ) - ], - ) - def test_get_segment_groups(self, lines: List[AhbLine], expected_group_names: Set[Optional[str]]): - actual = FlatAnwendungshandbuch._get_available_segment_groups(lines) - assert actual == expected_group_names - - @pytest.mark.parametrize( - "unsorted_input, sg_order, expected_result", - [ - pytest.param( - [ - AhbLine( - segment_group_key="Foo", - guid=uuid.UUID("12b1a98a-edf5-4177-89e5-a6d8a92c5fdc"), - segment_code=None, - data_element=None, - value_pool_entry=None, - ahb_expression=None, - name=None, - ), - AhbLine( - segment_group_key="Bar", - guid=uuid.UUID("9ec5ddb3-3721-48be-9d57-8742e08aa7cf"), - segment_code=None, - data_element=None, - value_pool_entry=None, - ahb_expression=None, - name=None, - ), - AhbLine( - segment_group_key=None, - guid=uuid.UUID("7beb2471-0fd2-4b6b-8aae-a5d1f153972d"), - segment_code=None, - data_element=None, - value_pool_entry=None, - ahb_expression=None, - name=None, - ), - AhbLine( - segment_group_key="Bar", - guid=uuid.UUID("365e1f9d-0bb9-43ab-921b-9ffa16b6df86"), - segment_code=None, - data_element=None, - value_pool_entry=None, - ahb_expression=None, - name=None, - ), - AhbLine( - segment_group_key="Foo", - guid=uuid.UUID("8874a7c9-0143-4aa5-b7fe-5225bb25d2c5"), - segment_code=None, - data_element=None, - value_pool_entry=None, - ahb_expression=None, - name=None, - ), - ], - ["Foo", "Bar", None], - [ - AhbLine( - segment_group_key="Foo", - guid=uuid.UUID("12b1a98a-edf5-4177-89e5-a6d8a92c5fdc"), - segment_code=None, - data_element=None, - value_pool_entry=None, - ahb_expression=None, - name=None, - ), - AhbLine( - segment_group_key="Foo", - guid=uuid.UUID("8874a7c9-0143-4aa5-b7fe-5225bb25d2c5"), - segment_code=None, - data_element=None, - value_pool_entry=None, - ahb_expression=None, - name=None, - ), - AhbLine( - segment_group_key="Bar", - guid=uuid.UUID("9ec5ddb3-3721-48be-9d57-8742e08aa7cf"), - segment_code=None, - data_element=None, - value_pool_entry=None, - ahb_expression=None, - name=None, - ), - AhbLine( - segment_group_key="Bar", - guid=uuid.UUID("365e1f9d-0bb9-43ab-921b-9ffa16b6df86"), - segment_code=None, - data_element=None, - value_pool_entry=None, - ahb_expression=None, - name=None, - ), - AhbLine( - segment_group_key=None, - guid=uuid.UUID("7beb2471-0fd2-4b6b-8aae-a5d1f153972d"), - segment_code=None, - data_element=None, - value_pool_entry=None, - ahb_expression=None, - name=None, - ), - ], - ) - ], - ) - def test_sorted_segment_groups( - self, unsorted_input: List[AhbLine], sg_order: List[Optional[str]], expected_result: List[AhbLine] - ): - actual = FlatAnwendungshandbuch._sorted_lines_by_segment_groups(unsorted_input, sg_order) - assert actual == expected_result - - @pytest.mark.parametrize( - "original,expected", - [ - pytest.param( - DeepAnwendungshandbuch( - meta=AhbMetaInformation(pruefidentifikator="11042"), - lines=[ - SegmentGroup( - ahb_expression="expr A", - discriminator="disc A", - segments=[ - Segment( - ahb_expression="expr B", - discriminator="disc B", - section_name="foo", - data_elements=[ - DataElementFreeText( - ahb_expression="Muss [1]", - entered_input="Hello Maus", - discriminator="foo", - data_element_id="4567", - ), - ], - ), - ], - segment_groups=[ - SegmentGroup( - discriminator="disc C", - ahb_expression="expr C", - segments=[ - Segment( - section_name="bar", - ahb_expression="expr Y", - discriminator="disc Y", - data_elements=[ - DataElementFreeText( - ahb_expression="Muss [1]", - entered_input="Hello Elefant", - discriminator="abc", - data_element_id="4567", - ), - DataElementValuePool( - value_pool=[ - ValuePoolEntry( - qualifier="Hallo Ente", meaning="world", ahb_expression="X" - ), - ValuePoolEntry( - qualifier="MAUS", meaning="rocks", ahb_expression="X" - ), - ], - entered_input="Hello Ente", - discriminator="ente", - data_element_id="4567", - ), - DataElementFreeText( - ahb_expression="Muss [1]", - entered_input="this should stay untouched", - discriminator="qwert", - data_element_id="4567", - ), - DataElementFreeText( - ahb_expression="Muss [1]", - entered_input="Not none", - discriminator="replace_with_none_here", - data_element_id="4567", - ), - ], - ) - ], - segment_groups=None, - ), - ], - ), - ], - ), - DeepAnwendungshandbuch( - meta=AhbMetaInformation(pruefidentifikator="11042"), - lines=[ - SegmentGroup( - ahb_expression="expr A", - discriminator="disc A", - segments=[ - Segment( - ahb_expression="expr B", - discriminator="disc B", - section_name="foo", - data_elements=[ - DataElementFreeText( - ahb_expression="Muss [1]", - entered_input="bar", - discriminator="foo", - data_element_id="4567", - ), - ], - ), - ], - segment_groups=[ - SegmentGroup( - discriminator="disc C", - ahb_expression="expr C", - segments=[ - Segment( - section_name="bar", - ahb_expression="expr Y", - discriminator="disc Y", - data_elements=[ - DataElementFreeText( - ahb_expression="Muss [1]", - entered_input="xyz", - discriminator="abc", - data_element_id="4567", - ), - DataElementValuePool( - value_pool=[ - ValuePoolEntry( - qualifier="Hallo Ente", meaning="world", ahb_expression="X" - ), - ValuePoolEntry( - qualifier="MAUS", meaning="rocks", ahb_expression="X" - ), - ], - entered_input="quack", - discriminator="ente", - data_element_id="4567", - ), - DataElementFreeText( - ahb_expression="Muss [1]", - entered_input="this should stay untouched", - discriminator="qwert", - data_element_id="4567", - ), - DataElementFreeText( - ahb_expression="Muss [1]", - entered_input=None, - discriminator="replace_with_none_here", - data_element_id="4567", - ), - ], - ) - ], - segment_groups=None, - ), - ], - ), - ], - ), - ) - ], - ) - def test_replacing_data_element_inputs(self, original: DeepAnwendungshandbuch, expected: DeepAnwendungshandbuch): - assert original != expected - - def replacement_func(discriminator: str) -> DeepAhbInputReplacement: - if discriminator == "foo": - return DeepAhbInputReplacement(replacement_found=True, input_replacement="bar") - if discriminator == "abc": - return DeepAhbInputReplacement(replacement_found=True, input_replacement="xyz") - if discriminator == "ente": - return DeepAhbInputReplacement(replacement_found=True, input_replacement="quack") - if discriminator == "replace_with_none_here": - return DeepAhbInputReplacement(replacement_found=True, input_replacement=None) - return DeepAhbInputReplacement(replacement_found=False, input_replacement=None) - - original.replace_inputs_based_on_discriminator(replacement_func) - assert original == expected - - @pytest.mark.parametrize( - "deep_ahb, expected_result_length", - [ - pytest.param( - DeepAnwendungshandbuch( - meta=AhbMetaInformation(pruefidentifikator="11042"), - lines=[ - SegmentGroup( - ahb_expression="expr A", - discriminator="disc A", - segments=[ - Segment( - ahb_expression="expr B", - discriminator="disc B", - data_elements=[ - DataElementValuePool( - value_pool=[ - ValuePoolEntry(qualifier="HELLO", meaning="world", ahb_expression="X"), - ValuePoolEntry(qualifier="MAUS", meaning="rocks", ahb_expression="X"), - ], - discriminator="baz", - entered_input="MAUS", - data_element_id="0123", - ), - DataElementFreeText( - ahb_expression="Muss [1]", - entered_input="Hello Mice", - discriminator="bar", - data_element_id="4567", - ), - ], - ), - ], - segment_groups=[], - ), - ], - ), - 1, - ) - ], - ) - def test_deep_ahb_get_value_pools(self, deep_ahb: DeepAnwendungshandbuch, expected_result_length: int): - actual = deep_ahb.get_all_value_pools() - assert len(actual) == 1 - - -""" - _find_this_sg2 = SegmentGroup( - ahb_expression="expr A", - discriminator="SG2", - segments=[ - Segment( - ahb_expression="expr B", - discriminator="FOO", - data_elements=[ - DataElementValuePool( - value_pool=[ - ValuePoolEntry(qualifier="HELLO", meaning="world", ahb_expression="X"), - ValuePoolEntry(qualifier="MAUS", meaning="rocks", ahb_expression="X"), - ], - discriminator="baz", - entered_input="MAUS", - data_element_id="0123", - ), - DataElementFreeText( - ahb_expression="Muss [1]", - entered_input="Hello Mice", - discriminator="bar", - data_element_id="4567", - ), - ], - ), - ], - segment_groups=[], - ) - - @pytest.mark.parametrize( - "deep_ahb,location, expected_result", - [ - pytest.param( - DeepAnwendungshandbuch( - meta=AhbMetaInformation(pruefidentifikator="11042"), - lines=[ - _find_this_sg2, - ], - ), - AhbLocation( - layers=[ - AhbLocationLayer(segment_group_key="SG2", opening_segment_code="FOO", opening_qualifier="ASD") - ] - ), - _find_this_sg2, - ) - ], - ) - def test_deep_ahb_find_location( - self, deep_ahb: DeepAnwendungshandbuch, location: AhbLocation, expected_result: Optional[SegmentGroup] - ): - actual = find_location(deep_ahb, location) - assert actual == expected_result -""" diff --git a/tests/unit_tests/test_maus.py b/tests/unit_tests/test_maus.py index f7a255e9..3beb2c9d 100644 --- a/tests/unit_tests/test_maus.py +++ b/tests/unit_tests/test_maus.py @@ -1,10 +1,10 @@ from typing import List import pytest # type:ignore[import] +from kohlrahbi.models.anwendungshandbuch import AhbLine from unit_tests.serialization_test_helper import assert_serialization_roundtrip # type:ignore[import] from maus.mig_ahb_matching import merge_lines_with_same_data_element -from maus.models.anwendungshandbuch import AhbLine from maus.models.edifact_components import ( DataElement, DataElementFreeText, diff --git a/tests/unit_tests/test_maus_provider.py b/tests/unit_tests/test_maus_provider.py index b3791d93..729aea54 100644 --- a/tests/unit_tests/test_maus_provider.py +++ b/tests/unit_tests/test_maus_provider.py @@ -6,10 +6,10 @@ from pathlib import Path import pytest # type:ignore[import] +from kohlrahbi.models.anwendungshandbuch import AhbMetaInformation, DeepAnwendungshandbuch from maus.edifact import EdifactFormat, EdifactFormatVersion from maus.maus_provider import FileBasedMausProvider, MausProvider -from maus.models.anwendungshandbuch import AhbMetaInformation, DeepAnwendungshandbuch, DeepAnwendungshandbuchSchema class MyFooBarMausProvider(FileBasedMausProvider): @@ -50,7 +50,7 @@ def test_file_based_maus_provider_success(self, tmpdir_factory): meta=AhbMetaInformation(pruefidentifikator="11001", maus_version="0.2.3"), lines=[] ) with open(maus_root_dir / "FV2104/UTILMD/11001_maus.json", "w+") as maus_test_outfile: - deep_ahb_dict = DeepAnwendungshandbuchSchema().dump(example_maus) # create a dictionary + deep_ahb_dict = example_maus.model_dump(mode="json") # create a dictionary json.dump(deep_ahb_dict, maus_test_outfile) # dump the dictionary to the file provider: MausProvider = MyFooBarMausProvider(base_path=maus_root_dir) assert provider is not None diff --git a/tests/unit_tests/test_navigation.py b/tests/unit_tests/test_navigation.py index 0661d48c..6a5d4570 100644 --- a/tests/unit_tests/test_navigation.py +++ b/tests/unit_tests/test_navigation.py @@ -3,9 +3,9 @@ import pytest # type:ignore[import] from jsonpath_ng.ext import parse # type:ignore[import] # jsonpath is just installed in the tests +from kohlrahbi.models.anwendungshandbuch import AhbLine, _remove_grouped_ahb_lines_containing_section_name from more_itertools import last -from maus.models.anwendungshandbuch import AhbLine, _remove_grouped_ahb_lines_containing_section_name from maus.models.message_implementation_guide import SegmentGroupHierarchy from maus.navigation import ( AhbLocation, diff --git a/tests/unit_tests/test_searching_in_models.py b/tests/unit_tests/test_searching_in_models.py index d73430e8..d00a9f15 100644 --- a/tests/unit_tests/test_searching_in_models.py +++ b/tests/unit_tests/test_searching_in_models.py @@ -1,16 +1,15 @@ from typing import Callable, List import pytest # type:ignore[import] - -from maus.models.anwendungshandbuch import ( +from kohlrahbi.models.anwendungshandbuch import ( AhbMetaInformation, DataElementFreeText, DataElementValuePool, DeepAnwendungshandbuch, Segment, SegmentGroup, + ValuePoolEntry, ) -from maus.models.edifact_components import ValuePoolEntry class TestSearchingInModels: @@ -239,6 +238,7 @@ def test_find_segments_from_deep_ahb( entered_input="Hello Mice", discriminator="bar", data_element_id="4567", + free_text="", ), ], ), diff --git a/tests/unit_tests/test_transformation.py b/tests/unit_tests/test_transformation.py index 29c053e5..ae2465a8 100644 --- a/tests/unit_tests/test_transformation.py +++ b/tests/unit_tests/test_transformation.py @@ -2,8 +2,8 @@ import pytest # type:ignore[import] from jsonpath_ng.ext import parse # type:ignore[import] # jsonpath is just installed in the tests +from kohlrahbi.models.anwendungshandbuch import AhbMetaInformation, DeepAnwendungshandbuch -from maus.models.anwendungshandbuch import AhbMetaInformation, DeepAnwendungshandbuch from maus.models.edifact_components import ( DataElementFreeText, DataElementValuePool,