70 lines
2.9 KiB
Python
70 lines
2.9 KiB
Python
|
"""The flow graph of a runbook-pack/v0.2: stages, steps, alternatives, gates, canonical order."""
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
|
||
|
|
class Flow:
|
||
|
|
def __init__(self, data: dict):
|
||
|
|
self.data = data
|
||
|
|
self.stages: dict[str, dict] = {}
|
||
|
|
self.steps: dict[str, dict] = {}
|
||
|
|
self.gates: dict[str, dict] = {}
|
||
|
|
self.alts: dict[str, str] = {} # alternative id -> parent step id
|
||
|
|
self.parent: dict[str, str | None] = {} # flow id -> enclosing stage id
|
||
|
|
self.order: list[str] = [] # canonical order of steps and gates
|
||
|
|
self.datapoints = {d["id"]: d for d in data.get("datapoints", [])}
|
||
|
|
self.actors = {a["id"]: a for a in data.get("actors", [])}
|
||
|
|
self._walk(data["stages"], None)
|
||
|
|
|
||
|
|
def _walk(self, stages: list[dict], parent: str | None) -> None:
|
||
|
|
for st in stages:
|
||
|
|
self.stages[st["id"]] = st
|
||
|
|
self.parent[st["id"]] = parent
|
||
|
|
for s in st.get("steps", []):
|
||
|
|
self.steps[s["id"]] = s
|
||
|
|
self.parent[s["id"]] = st["id"]
|
||
|
|
self.order.append(s["id"])
|
||
|
|
for a in s.get("one_of", []):
|
||
|
|
self.alts[a["id"]] = s["id"]
|
||
|
|
self._walk(st.get("stages", []), st["id"])
|
||
|
|
for g in st.get("gates", []):
|
||
|
|
self.gates[g["id"]] = g
|
||
|
|
self.parent[g["id"]] = st["id"]
|
||
|
|
self.order.append(g["id"])
|
||
|
|
|
||
|
|
# -- structure ------------------------------------------------------------------------
|
||
|
|
def kind(self, id_: str) -> str:
|
||
|
|
return "stage" if id_ in self.stages else "step" if id_ in self.steps else "gate" if id_ in self.gates else "?"
|
||
|
|
|
||
|
|
def node(self, id_: str) -> dict:
|
||
|
|
return self.stages.get(id_) or self.steps.get(id_) or self.gates[id_]
|
||
|
|
|
||
|
|
def resolve(self, ref: str) -> str | None:
|
||
|
|
if ref.startswith("gate:"):
|
||
|
|
return ref[5:] if ref[5:] in self.gates else None
|
||
|
|
return ref if ref in self.stages or ref in self.steps else None
|
||
|
|
|
||
|
|
def ancestors(self, id_: str) -> list[str]:
|
||
|
|
out, p = [], self.parent.get(id_)
|
||
|
|
while p:
|
||
|
|
out.append(p)
|
||
|
|
p = self.parent.get(p)
|
||
|
|
return out
|
||
|
|
|
||
|
|
def members(self, stage_id: str, include_gates: bool = True) -> list[str]:
|
||
|
|
"""Steps (and gates) inside a stage, recursively."""
|
||
|
|
out = []
|
||
|
|
st = self.stages[stage_id]
|
||
|
|
out += [s["id"] for s in st.get("steps", [])]
|
||
|
|
for child in st.get("stages", []):
|
||
|
|
out += self.members(child["id"], include_gates)
|
||
|
|
if include_gates:
|
||
|
|
out += [g["id"] for g in st.get("gates", [])]
|
||
|
|
return out
|
||
|
|
|
||
|
|
def requires(self, id_: str) -> list[str]:
|
||
|
|
"""Own requires plus those of every enclosing stage, as resolved flow ids."""
|
||
|
|
refs = list(self.node(id_).get("requires", []))
|
||
|
|
for a in self.ancestors(id_):
|
||
|
|
refs += self.stages[a].get("requires", [])
|
||
|
|
return [r for r in (self.resolve(x) for x in refs) if r]
|