| 138 | |
| 139 | @dataclass |
| 140 | class EdgeIR: |
| 141 | edge_type: str |
| 142 | from_node: Optional[int] = None |
| 143 | to_node: Optional[int] = None |
| 144 | cond_node: Optional[int] = None |
| 145 | then_entry: Optional[int] = None |
| 146 | else_entry: Optional[int] = None |
| 147 | header: Optional[int] = None |
| 148 | body_entry: Optional[int] = None |
| 149 | exit_node: Optional[int] = None |
| 150 | fork_node: Optional[int] = None |
| 151 | branches: Optional[list[int]] = None |
| 152 | join_node: Optional[int] = None |
| 153 | |
| 154 | def to_lean(self, prefix: str) -> str: |
| 155 | if self.edge_type == "seq": |
| 156 | return f".seqEdge {prefix}_nodeId{self.from_node} {prefix}_nodeId{self.to_node}" |
| 157 | if self.edge_type == "branch": |
| 158 | if self.else_entry is None: |
| 159 | return "-- ERROR: branchEdge with None else_entry" |
| 160 | return (f".branchEdge {prefix}_nodeId{self.cond_node} " |
| 161 | f"{prefix}_nodeId{self.then_entry} {prefix}_nodeId{self.else_entry}") |
| 162 | if self.edge_type == "loop": |
| 163 | if self.exit_node is None: |
| 164 | return "-- ERROR: loopEdge with None exit" |
| 165 | return (f".loopEdge {prefix}_nodeId{self.header} " |
| 166 | f"{prefix}_nodeId{self.body_entry} {prefix}_nodeId{self.exit_node}") |
| 167 | if self.edge_type == "loopBack": |
| 168 | return f".loopBackEdge {prefix}_nodeId{self.from_node} {prefix}_nodeId{self.to_node}" |
| 169 | if self.edge_type == "fork" and self.fork_node is not None and self.branches: |
| 170 | bl = ", ".join(f"{prefix}_nodeId{b}" for b in self.branches) |
| 171 | return f".forkEdge {prefix}_nodeId{self.fork_node} [{bl}]" |
| 172 | if self.edge_type == "join" and self.branches and self.join_node is not None: |
| 173 | bl = ", ".join(f"{prefix}_nodeId{b}" for b in self.branches) |
| 174 | return f".joinEdge [{bl}] {prefix}_nodeId{self.join_node}" |
| 175 | return f"-- UNSUPPORTED edge: {self.edge_type}" |
| 176 | |
| 177 | def to_dict(self) -> dict: |
| 178 | d: dict = {"edge_type": self.edge_type} |
| 179 | for k in ("from_node", "to_node", "cond_node", "then_entry", |
| 180 | "else_entry", "header", "body_entry", "exit_node", |
| 181 | "fork_node", "branches", "join_node"): |
| 182 | v = getattr(self, k) |
| 183 | if v is not None: |
| 184 | d[k] = v |
| 185 | return d |
| 186 | |
| 187 | @staticmethod |
| 188 | def from_dict(d: dict) -> "EdgeIR": |
| 189 | return EdgeIR(**{k: d.get(k) for k in ( |
| 190 | "edge_type", "from_node", "to_node", "cond_node", "then_entry", |
| 191 | "else_entry", "header", "body_entry", "exit_node", |
| 192 | "fork_node", "branches", "join_node", |
| 193 | )}) |
| 194 | |
| 195 | |
| 196 | # ---------- WorkflowIR ---------- |
no outgoing calls
no test coverage detected