22This module contains logic to convert EbdTable data to EbdGraph data.
33"""
44
5- from typing import Dict , List , Optional
5+ from typing import Dict , List , Literal , Optional , overload
66
77from networkx import DiGraph # type:ignore[import-untyped]
88
2222 ToNoEdge ,
2323 ToYesEdge ,
2424)
25- from rebdhuhn .models .ebd_graph import EmptyNode
25+ from rebdhuhn .models .ebd_graph import EmptyNode , TransitionEdge , TransitionNode
2626from rebdhuhn .models .errors import (
2727 EbdCrossReferenceNotSupportedError ,
2828 EndeInWrongColumnError ,
@@ -69,17 +69,29 @@ def _convert_sub_row_to_outcome_node(sub_row: EbdTableSubRow) -> Optional[Outcom
6969 return None
7070
7171
72- def _convert_row_to_decision_node (row : EbdTableRow ) -> DecisionNode :
72+ def _convert_row_to_decision_or_transition_node (row : EbdTableRow ) -> DecisionNode | TransitionNode :
7373 """
7474 converts a row into a decision node
7575 """
76+ if len (row .sub_rows ) == 1 and row .sub_rows [0 ].check_result .result is None :
77+ # this is a transition node
78+ return TransitionNode (step_number = row .step_number , question = row .description , note = row .sub_rows [0 ].note )
7679 return DecisionNode (step_number = row .step_number , question = row .description )
7780
7881
79- def _yes_no_transition_edge (decision : Optional [bool ], source : DecisionNode , target : EbdGraphNode ) -> EbdGraphEdge :
80- if decision is None :
81- # happens in another PR
82- raise NotImplementedError ("None not supported yet; https://github.com/Hochfrequenz/rebdhuhn/issues/380" )
82+ @overload
83+ def _yes_no_transition_edge (
84+ decision : Literal [None ], source : TransitionNode , target : EbdGraphNode
85+ ) -> TransitionEdge : ...
86+ @overload
87+ def _yes_no_transition_edge (decision : bool , source : DecisionNode , target : EbdGraphNode ) -> EbdGraphEdge : ...
88+ def _yes_no_transition_edge (
89+ decision : Optional [bool ], source : DecisionNode | TransitionNode , target : EbdGraphNode
90+ ) -> EbdGraphEdge :
91+ if decision is None and isinstance (source , TransitionNode ):
92+ return TransitionEdge (source = source , target = target , note = None )
93+ assert not isinstance (source , TransitionNode ), "Iff the decision is None, source must be a TransitionNode"
94+ assert isinstance (source , DecisionNode )
8395 if decision is True :
8496 return ToYesEdge (source = source , target = target , note = None )
8597 if decision is False :
@@ -95,8 +107,10 @@ def get_all_nodes(table: EbdTable) -> List[EbdGraphNode]:
95107 result : List [EbdGraphNode ] = [StartNode ()]
96108 contains_ende = False
97109 for row in table .rows :
98- decision_node = _convert_row_to_decision_node (row )
99- result .append (decision_node )
110+ decision_or_transition_node = _convert_row_to_decision_or_transition_node (row )
111+ result .append (decision_or_transition_node )
112+ if isinstance (decision_or_transition_node , TransitionNode ):
113+ continue
100114 for sub_row in row .sub_rows :
101115 outcome_node = _convert_sub_row_to_outcome_node (sub_row )
102116 if outcome_node is not None :
@@ -134,12 +148,24 @@ def get_all_edges(table: EbdTable) -> List[EbdGraphEdge]:
134148 outcome_nodes_duplicates : dict [str , OutcomeNode ] = {} # map to check for duplicate outcome nodes
135149
136150 for row in table .rows :
137- decision_node = _convert_row_to_decision_node (row )
151+ row_node = _convert_row_to_decision_or_transition_node (row )
152+ if isinstance (row_node , TransitionNode ):
153+ assert row .sub_rows [0 ].check_result .subsequent_step_number is not None
154+ result .append (
155+ TransitionEdge (
156+ source = row_node ,
157+ target = nodes [row .sub_rows [0 ].check_result .subsequent_step_number ],
158+ note = row_node .note ,
159+ )
160+ )
161+ continue
162+ assert isinstance (row_node , DecisionNode )
138163 for sub_row in row .sub_rows :
164+ assert isinstance (sub_row .check_result .result , bool )
139165 if sub_row .check_result .subsequent_step_number is not None and not _is_ende_with_no_code_but_note (sub_row ):
140166 edge = _yes_no_transition_edge (
141167 sub_row .check_result .result ,
142- source = decision_node ,
168+ source = row_node ,
143169 target = nodes [sub_row .check_result .subsequent_step_number ],
144170 )
145171 else :
@@ -149,8 +175,8 @@ def get_all_edges(table: EbdTable) -> List[EbdGraphEdge]:
149175 if all (sr .result_code is None for sr in row .sub_rows ) and any (
150176 sr .note is not None and sr .note .startswith ("EBD " ) for sr in row .sub_rows
151177 ):
152- raise EbdCrossReferenceNotSupportedError (row = row , decision_node = decision_node )
153- raise OutcomeNodeCreationError (decision_node = decision_node , sub_row = sub_row )
178+ raise EbdCrossReferenceNotSupportedError (row = row , decision_node = row_node )
179+ raise OutcomeNodeCreationError (decision_node = row_node , sub_row = sub_row )
154180
155181 # check for ambiguous outcome nodes, i.e. A** with different notes
156182 is_ambiguous_outcome_node = (
@@ -167,7 +193,7 @@ def get_all_edges(table: EbdTable) -> List[EbdGraphEdge]:
167193
168194 edge = _yes_no_transition_edge (
169195 sub_row .check_result .result ,
170- source = decision_node ,
196+ source = row_node ,
171197 target = nodes [outcome_node .get_key ()],
172198 )
173199 result .append (edge )
0 commit comments