| 197 | |
| 198 | @dataclass |
| 199 | class WorkflowIR: |
| 200 | name: str |
| 201 | goal: str |
| 202 | parameters: list[TypedVar] |
| 203 | nodes: list[NodeIR] |
| 204 | edges: list[EdgeIR] |
| 205 | entry: int |
| 206 | exits: list[int] |
| 207 | submodule_map: Optional[dict[str, SubmoduleRef]] = None |
| 208 | |
| 209 | def to_dict(self) -> dict: |
| 210 | d: dict = { |
| 211 | "name": self.name, "goal": self.goal, |
| 212 | "parameters": [p.to_dict() for p in self.parameters], |
| 213 | "nodes": [n.to_dict() for n in self.nodes], |
| 214 | "edges": [e.to_dict() for e in self.edges], |
| 215 | "entry": self.entry, "exits": self.exits, |
| 216 | } |
| 217 | if self.submodule_map: |
| 218 | d["submodule_map"] = {k: v.to_dict() for k, v in self.submodule_map.items()} |
| 219 | return d |
| 220 | |
| 221 | @staticmethod |
| 222 | def from_dict(d: dict) -> "WorkflowIR": |
| 223 | return WorkflowIR( |
| 224 | name=d["name"], goal=d["goal"], |
| 225 | parameters=[TypedVar.from_dict(p) for p in d["parameters"]], |
| 226 | nodes=[NodeIR.from_dict(n) for n in d["nodes"]], |
| 227 | edges=[EdgeIR.from_dict(e) for e in d["edges"]], |
| 228 | entry=d["entry"], exits=d["exits"], |
| 229 | submodule_map={k: SubmoduleRef.from_dict(v) for k, v in d["submodule_map"].items()} |
| 230 | if d.get("submodule_map") else None, |
| 231 | ) |
| 232 | |
| 233 | def save_json(self, path: str, indent: int = 2) -> None: |
| 234 | with open(path, "w", encoding="utf-8") as f: |
| 235 | json.dump(self.to_dict(), f, indent=indent, ensure_ascii=False) |
| 236 | |
| 237 | @staticmethod |
| 238 | def load_json(path: str) -> "WorkflowIR": |
| 239 | with open(path, "r", encoding="utf-8") as f: |
| 240 | return WorkflowIR.from_dict(json.load(f)) |
| 241 | |
| 242 | |
| 243 | # ============================================================================ |
no outgoing calls
no test coverage detected