From 063d80a037041d7892eaa2e59156d1198531b432 Mon Sep 17 00:00:00 2001 From: Daniel <139119540+hf-ddernbach@users.noreply.github.com> Date: Mon, 18 Sep 2023 14:24:09 +0200 Subject: [PATCH] Added error handling for ambiguous outcome_codes -> A** (#127) cf. https://github.com/Hochfrequenz/ebdtable2graph/issues/38 --- src/ebdtable2graph/graph_conversion.py | 18 ++++++++++++++++++ src/ebdtable2graph/models/errors/__init__.py | 12 +++++++++++- 2 files changed, 29 insertions(+), 1 deletion(-) diff --git a/src/ebdtable2graph/graph_conversion.py b/src/ebdtable2graph/graph_conversion.py index 9ee7e3f..57cfba9 100644 --- a/src/ebdtable2graph/graph_conversion.py +++ b/src/ebdtable2graph/graph_conversion.py @@ -23,6 +23,7 @@ from ebdtable2graph.models.errors import ( EbdCrossReferenceNotSupportedError, EndeInWrongColumnError, + OutcomeCodeAmbiguousError, OutcomeNodeCreationError, ) @@ -89,6 +90,8 @@ def get_all_edges(table: EbdTable) -> List[EbdGraphEdge]: first_node_after_start = _get_key_and_node_with_lowest_step_number(table)[1] result: List[EbdGraphEdge] = [EbdGraphEdge(source=nodes["Start"], target=first_node_after_start, note=None)] + outcome_nodes_duplicates: dict[str, OutcomeNode] = {} # map to check for duplicate outcome nodes + for row in table.rows: decision_node = _convert_row_to_decision_node(row) for sub_row in row.sub_rows: @@ -100,6 +103,7 @@ def get_all_edges(table: EbdTable) -> List[EbdGraphEdge]: ) else: outcome_node: Optional[OutcomeNode] = _convert_sub_row_to_outcome_node(sub_row) + if outcome_node is None: if all(sr.result_code is None for sr in row.sub_rows) and any( sr.note is not None and sr.note.startswith("EBD ") for sr in row.sub_rows @@ -110,6 +114,20 @@ def get_all_edges(table: EbdTable) -> List[EbdGraphEdge]: ): raise EndeInWrongColumnError(row=row) raise OutcomeNodeCreationError(decision_node=decision_node, sub_row=sub_row) + + # check for ambiguous outcome nodes, i.e. A** with different notes + is_ambiguous_outcome_node = ( + outcome_node.result_code in outcome_nodes_duplicates + and outcome_nodes_duplicates[outcome_node.result_code].note != outcome_node.note + ) + + if not is_ambiguous_outcome_node: + outcome_nodes_duplicates[outcome_node.result_code] = outcome_node + else: + raise OutcomeCodeAmbiguousError( + outcome_node1=outcome_nodes_duplicates[outcome_node.result_code], outcome_node2=outcome_node + ) + edge = _yes_no_edge( sub_row.check_result.result, source=decision_node, diff --git a/src/ebdtable2graph/models/errors/__init__.py b/src/ebdtable2graph/models/errors/__init__.py index ce4de07..6100425 100644 --- a/src/ebdtable2graph/models/errors/__init__.py +++ b/src/ebdtable2graph/models/errors/__init__.py @@ -4,7 +4,7 @@ """ from typing import Optional -from ebdtable2graph.models import DecisionNode, EbdTableRow, EbdTableSubRow +from ebdtable2graph.models import DecisionNode, EbdTableRow, EbdTableSubRow, OutcomeNode class NotExactlyTwoOutgoingEdgesError(NotImplementedError): @@ -109,3 +109,13 @@ def __init__(self, decision_node: DecisionNode, sub_row: EbdTableSubRow): super().__init__(f"Cannot create outcome node from sub row {sub_row} for DecisionNode {decision_node}.") self.sub_row = sub_row self.decision_node = decision_node + + +class OutcomeCodeAmbiguousError(ValueError): + """ + Raised when the result nodes are ambiguous. This can be the case for "A**" results. + """ + + def __init__(self, outcome_node1: OutcomeNode, outcome_node2: OutcomeNode): + super().__init__(f"Ambiguous result codes: for [{outcome_node1, outcome_node2}].") + self.outcome_nodes = [outcome_node1, outcome_node2]