maus.models package

Submodules

maus.models.anwendungshandbuch module

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)

Bases: object

An AhbLine is a single line inside the machine-readable, flat AHB.

ahb_expression: str | None: a requirement indicator + an optional condition (“ahb expression”), e.g. ‘Muss [123] O [456]’

data_element: str | None: the data element ID, e.g. ‘3224’

get_discriminator(include_name: bool = True) → str[source]: 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.

guid: UUID | None: optional key

holds_any_information() → bool[source]: 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.

index: int | None: index is a number that describes the position of the AHBLine inside the original PDF- and FlatAnwendungshandbuch.

name: str | None: the name, e.g. ‘Meldepunkt’. It can be both the description of a field but also its meaning

section_name: str | None: The section name describes the purpose of a segment, e.g. “Nachrichten-Kopfsegment” or “Beginn der Nachricht”

segment_code: str | None: the segment, e.g. ‘IDE’

segment_group_key: str | None: the segment group, e.g. ‘SG5’

value_pool_entry: str | None: one of (possible multiple) allowed values, e.g. ‘E01’ or ‘293’

Bases: Schema

A schema to (de-)serialize AhbLine

deserialize(data, **kwargs) → AhbLine[source]: Converts the barely typed data dictionary into an actual AhbLine

opts: SchemaOpts = <marshmallow.schema.SchemaOpts object>

class maus.models.anwendungshandbuch.AhbMetaInformation(*, pruefidentifikator: str, maus_version: str | None = '0.3.0', description: str | None = None, direction: str | None = None)[source]

Bases: object

Meta information about an AHB like e.g. its title, Prüfidentifikator, possible sender and receiver roles

description: str | None: an optional description of the purpose of the pruefidentifikator; e.g. ‘Anmeldung MSB’ for 11042

direction: str | 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

maus_version: str | None: semantic version of maus used to create this document

pruefidentifikator: str: identifies the message type (within a fixed format version) e.g. “11042” or “13012”

Bases: Schema

A schema to (de-)serialize AhbMetaInformation

deserialize(data, **kwargs) → AhbMetaInformation[source]: Converts the barely typed data dictionary into an actual AhbMetaInformation

opts: SchemaOpts = <marshmallow.schema.SchemaOpts object>

class maus.models.anwendungshandbuch.DeepAhbInputReplacement(*, replacement_found: bool, input_replacement: str | None)[source]

Bases: object

A container class that models replacements of inputs in the DeepAnwendungshandbuch

input_replacement: str | None: 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.

replacement_found: bool: true iff a replacement is applicable

class maus.models.anwendungshandbuch.DeepAnwendungshandbuch(*, meta: AhbMetaInformation, lines: List[SegmentGroup])[source]

Bases: object

The data of the AHB nested as described in the MIG.

find_segment_groups(predicate: Callable[[SegmentGroup], bool]) → List[SegmentGroup][source]: recursively search for segment group in this ahb that meets the predicate. :return: list of segment groups that match the predicate; empty list otherwise

find_segments(group_predicate: ~typing.Callable[[~maus.models.edifact_components.SegmentGroup], bool] = <function DeepAnwendungshandbuch.<lambda>>, segment_predicate: ~typing.Callable[[~maus.models.edifact_components.Segment], bool] = <function DeepAnwendungshandbuch.<lambda>>) → List[Segment][source]: 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

get_all_expressions() → List[str][source]: recursively iterate through the deep ahb and return all distinct expressions found

get_all_value_pools() → List[DataElementValuePool][source]: recursively find all value pools in the deep ahb :return: a list of all value pools

lines: List[SegmentGroup]: the nested data

meta: AhbMetaInformation: information about this AHB

replace_inputs_based_on_discriminator(replacement_func: Callable[[str], DeepAhbInputReplacement]) → None[source]: Replace all the entered_inputs in the entire DeepAnwendungshandbuch using the given replacement_func. Note that this modifies this DeepAnwendungshandbuch instance (self).

reset_ahb_line_index() → None[source]: reset the ahb line index for all lines in the DeepAnwendungshandbuch :return: nothing; modifies the instance

Bases: Schema

A schema to (de-)serialize DeepAnwendungshandbuch

deserialize(data, **kwargs) → DeepAnwendungshandbuch[source]: Converts the barely typed data dictionary into an actual DeepAnwendungshandbuch

opts: SchemaOpts = <marshmallow.schema.SchemaOpts object>

class maus.models.anwendungshandbuch.FlatAnwendungshandbuch(*, meta: AhbMetaInformation, lines: List[AhbLine])[source]

Bases: object

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

get_segment_groups() → List[str | None][source]

Returns:: a set with all segment groups in this AHB in the order in which they occur

lines: List[AhbLine]: ordered list lines as they occur in the AHB

meta: AhbMetaInformation: information about this AHB

sort_lines_by_segment_groups()[source]: sorts lines by segment groups while preserving the order inside the groups and the order between the groups.

Bases: Schema

A schema to (de-)serialize FlatAnwendungshandbuch

deserialize(data, **kwargs) → FlatAnwendungshandbuch[source]: Converts the barely typed data dictionary into an actual FlatAnwendungshandbuch

opts: SchemaOpts = <marshmallow.schema.SchemaOpts object>

maus.models.edifact_components module

EDIFACT components are data structures on different hierarchical levels inside an EDIFACT message. Components contain not only EDIFACT composits but also segments and segment groups.

class maus.models.edifact_components.DataElement(*, discriminator: str | None, data_element_id: str, entered_input: str | None, value_type: DataElementDataType | None = None)[source]

Bases: ABC

A data element holds specific kinds of data. It is defined in EDIFACT. At least for the German energy market communication any data element has a 4 digit key. For example in UTILMD the data element that holds the 13 digit market partner ID is data element ‘3039’

data_element_id: str: the ID of the data element (e.g. “0062”) for the Nachrichten-Referenznummer

discriminator: str | None: The discriminator uniquely identifies the data element. The discriminator is None if the data element was not found in the MIG.

entered_input: str | None: If the message which is evaluated contains data for this data element, this is set to a value which is not None. The field can either carry a free text or an element from a value pool (depending on the value_type)

value_type: DataElementDataType | None: The value_type allows to describe which type of data we’re expecting to be used within this data element. The value_type does not discriminate the type of the data element itself.

class maus.models.edifact_components.DataElementDataType(value)[source]

Bases: str, Enum

The Data Element Data Type describes with which kind of data element we’re dealing with in a data element. This information is set but not used anywhere inside MAUS directly but more of a “service” to downstream code.

DATETIME = 'DATETIME': a datetime string, usually as RFC3339

TEXT = 'TEXT': plain text, e.g. a name

VALUE_POOL = 'VALUE_POOL': the user can choose between different possible values

class maus.models.edifact_components.DataElementFreeText(*, discriminator: str | None, data_element_id: str, entered_input: str | None, value_type: DataElementDataType | None = DataElementDataType.TEXT, ahb_expression: str)[source]

Bases: DataElement

A DataElementFreeText is a data element that allows entering arbitrary data. This is the main difference to the DataElementValuePool which has a finite set of allowed values attached.

ahb_expression: str: any freetext data element has an ahb expression attached. Could be ‘X’ but also ‘M [13]’

value_type: DataElementDataType | None: The value_type allows to describe which type of data we’re expecting to be used within this data element. The value_type does not discriminate the type of the data element itself.

Bases: DataElementSchema

A Schema to serialize DataElementFreeTexts

deserialize(data, **kwargs) → DataElementFreeText[source]: Converts the barely typed data dictionary into an actual DataElementFreeText

opts: SchemaOpts = <marshmallow.schema.SchemaOpts object>

Bases: Schema

A Schema to (de-)serialize DataElements

opts: SchemaOpts = <marshmallow.schema.SchemaOpts object>

class maus.models.edifact_components.DataElementValuePool(*, discriminator: str | None, data_element_id: str, entered_input: str | None, value_type: DataElementDataType | None = DataElementDataType.VALUE_POOL, value_pool: List[ValuePoolEntry])[source]

Bases: DataElement

A DataElementValuePool is a data element with a finite set of allowed values. These allowed values are referred to as keys of the “value pool”. The value pool is defined both on MIG level and AHB level The set of values allowed according to the AHB is always a subset of the values allowed according to the MIG.

has_value_pool_which_is_subset_of(entries: Iterable[str]) → bool[source]: returns true iff all qualifiers from the data elements value pool are found in entries

replace_value_pool(edifact_to_domain_mapping: Mapping[str, str], meaning_qualifier_merger: Callable[[str, str], str] | None = None) → None[source]

If your data model comes from another domain than edifact the value pool from the AHBs, you can, in general, not use the same keys. Think e.g. of the Transaktionsgrund in UTILMD. In EDIFACT you might have the values: Edifact Domain: - E01 = Einzug/Auszug - E03 = Wechsel - Z33 = Auszug/Stilllegung … But if, in your application, you’re modelling these values with other values (think e.g. of readable enums): Application Domain: - EINZUG = Einzug/Auszug - WECHSEL = Wechsel - STILLLEGUNG = Auszug/Stilllegung … You need to transform the value pool from edifact to something that matches the domain from your application. To do so, provide this method with an edifact_to_domain_mapping: {“E01”: “EINZUG”,”E02”: “WECHSEL”,”E03”: “STILLLEGUNG”} where each entry represents the mapping of an edifact qualifier to your application domain.

This method replaces the keys of the ValuePoolEntries if they are found in the mapping. By doing so, it allows transforming checks against value pools from the edifact to your application domain. This problem does not occur for free text data elements.

Provide an optional meaning_qualifier_merger that allows you to ‘store’ the old qualifier in the meaning. This is an _unstructured_ way to save the information that is lost with the replacement.

value_pool: List[ValuePoolEntry]: The value pool contains at least one value ValuePoolEntry

value_type: DataElementDataType | None: type of the value, if known

Bases: DataElementSchema

A Schema to serialize DataElementValuePool

deserialize(data, **kwargs) → DataElementValuePool[source]: Converts the barely typed data dictionary into an actual DataElementValuePool

opts: SchemaOpts = <marshmallow.schema.SchemaOpts object>

class maus.models.edifact_components.EdifactStack(*, levels: List[EdifactStackLevel])[source]

Bases: object

The EdifactStack describes where inside an EDIFACT message data are found. The stack is independent of the actual implementation used to create the EDIFACT (be it XML, JSON whatever).

static from_json_path(json_path: str) → EdifactStack[source]: reads a json path as it is created by “to_json_path” and returns the corresponding edifact stack

is_parent_of(other: EdifactStack) → bool[source]: Returns true iff this other stack is a sub stack of self.

is_sub_stack_of(other: EdifactStack) → bool[source]: Returns true iff this (self) stack is a sub stack of the other provided stack. ([Foo][0][Bar]).is_sub_stack_of([Foo][0]) is true.

levels: List[EdifactStackLevel]: levels describe the nesting inside an edifact message

to_json_path() → str[source]: Transforms this instance into a JSON Path.

class maus.models.edifact_components.EdifactStackLevel(*, name: str, is_groupable: bool, index: int | None = None)[source]

Bases: object

The EDIFACT stack level describes the hierarchy level of information inside an EDIFACT message.

index: int | None: the index if present (e.g. 0)

is_groupable: bool: describes if this level is groupable / if there are multiple instances of this level within the same message

name: str: the name of the level, e.g. ‘Dokument’ or ‘Nachricht’ or ‘Meldepunkt’

class maus.models.edifact_components.Segment(*, discriminator: str, ahb_expression: str, ahb_line_index: int | None = None, data_elements: List[DataElement], section_name: str | None = None)[source]

Bases: SegmentLevel

A Segment contains multiple data elements.

data_elements: List[DataElement]

get_all_value_pools() → List[DataElementValuePool][source]: find all value pools in this segment :return: a list of all value pools

section_name: str | None: For the MIG matching it might be necessary to know the section in which the data element occured in the AHB. This might be necessary to e.g. distinguish gas and electricity fields which look the same otherwise. See e.g. UTILMD ‘Geplante Turnusablesung des MSB (Strom)’ vs. ‘Geplante Turnusablesung des NB (Gas)’

class maus.models.edifact_components.SegmentGroup(*, discriminator: str, ahb_expression: str, ahb_line_index: int | None = None, segments: List[Segment] | None = None, segment_groups: List[SegmentGroup] | None = None)[source]

Bases: SegmentLevel

A segment group that contains segments and nested groups. On root level of a message there might be a “virtual” segment group of all segments that formally don’t have a group This group has the key “root”.

find_segments(predicate: Callable[[Segment], bool], search_recursively: bool = True) → List[Segment][source]: Search for a segment that matches the predicate (in this group and subgroups if ‘search_recursively’ is set), Return results, if found. Return empty list otherwise.

get_all_value_pools() → List[DataElementValuePool][source]: recursively find all value pools in this segmentgroup :return: a list of all value pools

reset_ahb_line_index() → None[source]: sets the ahb_line_index to None. This is to allow comparisons regardless of the index

segment_groups: List[SegmentGroup] | None: groups that are nested into this group

segments: List[Segment] | None: the segments inside this very group

Bases: SegmentLevelSchema

A Schema to serialize SegmentGroups.

deserialize(data, **kwargs) → SegmentGroup[source]: Converts the barely typed data dictionary into an actual SegmentGroup

opts: SchemaOpts = <marshmallow.schema.SchemaOpts object>

class maus.models.edifact_components.SegmentLevel(*, discriminator: str, ahb_expression: str, ahb_line_index: int | None = None)[source]

Bases: ABC

SegmentLevel describes @annika: what does it describe?

ahb_expression: str

ahb_line_index: int | None: Allows sorting the segments depending on where they occured in the FlatAnwendungshandbuch. It won’t be serialized though.

discriminator: str

reset_ahb_line_index() → None[source]: sets the ahb_line_index to None. This is to allow comparisons regardless of the index

Bases: Schema

A Schema to serialize SegmentLevels

opts: SchemaOpts = <marshmallow.schema.SchemaOpts object>

Bases: SegmentLevelSchema

A Schema to serialize Segments

deserialize(data, **kwargs) → Segment[source]: Converts the barely typed data dictionary into an actual Segment

opts: SchemaOpts = <marshmallow.schema.SchemaOpts object>

class maus.models.edifact_components.ValuePoolEntry(*, qualifier: str, meaning: str, ahb_expression: str)[source]

Bases: object

A value pool entry contains the EDIFACT qualifier, a meaning (German text) and an ahb expression. A value pool consists of 1 to n ValuePoolEntries. The data element 3055 in UTILMD is a good example for a value pool. It is used in the segments NAD+MS and NAD+MR. Its ValuePoolEntries are: - (key: “9”, meaning: “GS1”, ahb_expression: “X”) - (key: “293”, meaning: “DE, BDEW”, ahb_expression: “X”) - (key: “332”, meaning: “DE, DVGW”, ahb_expression: “X”)

ahb_expression: str: the ahb expression, in most cases this is a simple “X”; it must not be empty

meaning: str: the meaning as it is written in the AHB (e.g. “Einzug”, “Entwurfs-Version”, “GS1”, “Codeliste Gas G_0057”

qualifier: str: the qualifier in edifact, might be e.g. “E01”, “D”, “9”, “1.1a”, “G_0057”

Bases: Schema

A Schema to serialize ValuePoolEntries

deserialize(data, **kwargs) → ValuePoolEntry[source]: Converts the barely typed data dictionary into an actual ValuePoolEntry

opts: SchemaOpts = <marshmallow.schema.SchemaOpts object>

maus.models.edifact_components.derive_data_type_from_segment_code(segment_code: str) → DataElementDataType | None[source]: derives the expected data type from the segment code, e.g. DATETIME for DTM segments :return: The DataType if it can be derived without any doubt, None otherwise

maus.models.edifact_components.gabi_edifact_qualifier_pattern = re.compile('^(?!MP-ID)(GABi)?[A-Z\\d\\-]+(RLM(o|m)T)?$')

They contain with “-” and lower case “i”/”o”/”n”

Type:: a pattern that matches the GABi qualifiers

maus.models.message_implementation_guide module

Contains classes to Model Message Implementation Guides (MIG)

class maus.models.message_implementation_guide.SegmentGroupHierarchy(*, segment_group: str | None, opening_segment: str, sub_hierarchy: List[SegmentGroupHierarchy] | None)[source]

Bases: object

Models the hierarchy of segment groups within an EDIFACT format. Therefore each segment group is modeled together with its respective first segment inside the group. This is not a full representation of the EDIFACT formats structure but enough to understand the basic hierarchy. The Segment Group Hierarchy (SGH) is what allows us to transform a flat AHB into a NestedAhb that has a structure. This information / the SGH can be (manually) extracted form the diagrams in the MIG PDFs or generated from another source (which the BDEW does not provide, sadly). See the unit tests for working examples for UTILMD and MSCONS. It’ll make sense then.

flattened() → List[Tuple[str | None, str]][source]: Returns the segment groups and opening tuples in the order in which they are expected to occur in an AHB. This is useful to match this deep structure with a flat AHB.

is_hierarchically_below(segment_group_key: str | None) → bool[source]: returns true iff the segment_group provided is a (direct or indirect) sub group of self.

opening_segment: str: first segment in group, e.g. ‘IDE’ or ‘LOC’

segment_group: str | None: segment group name, e.g. “SG4” or “SG5”

sg_is_hierarchically_below(segment_group_key_x: str | None, segment_group_key_y: str | None) → bool[source]: returns true iff segment group with segment_group_key_x is hierarchically below the segment_group with key_y :param segment_group_key_x: :param segment_group_key_y: :return:

sub_hierarchy: List[SegmentGroupHierarchy] | None: segment groups below this level, e.g.[(SG5, LOG), (SG6,RFF), (SG8, SEQ), …]

Bases: Schema

A schema to (de-)serialize SegmentGroupHierarchy

deserialize(data, **kwargs) → SegmentGroupHierarchy[source]: Converts the barely typed data dictionary into an actual SegmentGroupHierarchy

opts: SchemaOpts = <marshmallow.schema.SchemaOpts object>

Module contents

models are the data classes used to store information from e.g. ahbs. This module contains methods available to all methods in the package.