vibefpmgui.control_flow

  1"""A Fortran procedure's control-flow graph, and a layout for drawing it (no Qt).
  2
  3`build` turns a procedure's statements (read as `vibefpmgui.complexity`
  4reads them) into nodes and edges:
  5
  6- a run of statements without a branch is one *block*;
  7- ``if (…) then`` / ``else if``, one-line ``if``, ``select case`` (and
  8  ``select type``, ``select rank``) and ``where`` are *decisions*, their
  9  edges labelled ("true", "false", "case (1)"…);
 10- a loop (``do``, ``do while``, ``forall`` block…) is a *loop* node: "loop"
 11  into its body, "done" out of it, the body's end back to it; ``cycle`` goes
 12  back to it, ``exit`` out of it;
 13- ``return`` goes to the *end* node, ``stop`` / ``error stop`` to a *stop*
 14  node; ``goto`` (computed too) and an arithmetic ``if`` go to their labels.
 15
 16Included files (``#include``, ``include``) are read where they're included
 17(see `vibefpmgui.complexity.expand_includes`); a node's lines are in one
 18file, its ``path``.
 19
 20For a graph with one way in, ``edges - nodes + 2`` is the number of its
 21decisions plus 1: its cyclomatic complexity (without the ``.and.`` /
 22``.or.`` that `vibefpmgui.complexity`'s CCN also counts).
 23
 24`layout` places the nodes in layers, top to bottom (loops' back edges set
 25aside), ordered to cross few edges.
 26"""
 27
 28from __future__ import annotations
 29
 30import re
 31from dataclasses import dataclass, field
 32from pathlib import Path
 33
 34from . import complexity as cx
 35
 36START, END, STOP, BLOCK, DECISION, LOOP = "start", "end", "stop", "block", "decision", "loop"
 37MAX_STATEMENTS_SHOWN = 4  # a block's lines shown (then "…")
 38
 39
 40@dataclass
 41class Node:
 42    id: int
 43    kind: str                     # START, END, STOP, BLOCK, DECISION, LOOP
 44    first: int = 0                # its lines (1-based; 0 for start/end)
 45    last: int = 0
 46    text: list[str] = field(default_factory=list)  # what's shown: its source lines (or "start", "end"…)
 47    path: Path | None = None      # the file its lines are in (an included file's, or the procedure's)
 48
 49
 50@dataclass
 51class Edge:
 52    source: int
 53    target: int
 54    label: str = ""               # "true", "false", "loop", "done", "case (1)"…
 55    back: bool = False            # (set by layout: a loop's way back)
 56
 57
 58@dataclass
 59class Graph:
 60    name: str
 61    path: Path | None = None      # the procedure's file
 62    nodes: list[Node] = field(default_factory=list)
 63    edges: list[Edge] = field(default_factory=list)
 64
 65    @property
 66    def complexity(self) -> int:
 67        """edges - nodes + 2 (with an edge from the stop node, if there's one,
 68        to the end: the way out of the procedure is one)."""
 69        stops = sum(1 for n in self.nodes if n.kind == STOP)
 70        return len(self.edges) + stops - len(self.nodes) + 2
 71
 72
 73# ------------------------------------------------------------------ statements
 74@dataclass
 75class _Statement:
 76    kind: str           # see _classify
 77    text: str           # (lowercase, strings blanked)
 78    label: str
 79    first: int
 80    last: int
 81    detail: str = ""    # a decision's condition, a case's selector, a goto's labels…
 82    shown: str = ""     # what's shown for it ("": its source line)
 83
 84
 85# The specification part: declarations and the like, before the first statement that runs.
 86SPECIFICATION_RE = re.compile(r"^(use|implicit|import|integer|real|double\s*precision|complex|logical|character|"
 87                              r"type\s*\(|class\s*\(|procedure\s*\(|dimension|parameter|external|intrinsic|save|"
 88                              r"data|common|equivalence|namelist|allocatable|pointer|target|optional|intent|private|"
 89                              r"public|protected|contiguous|asynchronous|volatile|value|bind|format|entry|include)\b")
 90GOTO_RE = re.compile(r"^go\s*to\s*(\d+)\s*$")
 91COMPUTED_GOTO_RE = re.compile(r"^go\s*to\s*\(([\d\s,]+)\)")
 92ARITHMETIC_IF_RE = re.compile(r"^(\d+)\s*,\s*(\d+)\s*,\s*(\d+)$")
 93
 94
 95def _classify(s: str) -> tuple[str, str]:
 96    """(kind, detail) of a statement."""
 97    if cx.END_RE.match(s):
 98        word = (cx.END_RE.match(s).group(1) or "").replace(" ", "")
 99        return {"if": "end_if", "do": "end_do", "select": "end_select", "where": "end_where",
100                "forall": "end_do", "associate": "end_block", "block": "end_block", "critical": "end_block",
101                "team": "end_block", "interface": "end_declaration", "type": "end_declaration",
102                "enum": "end_declaration"}.get(word, "end_unit"), ""
103    if s == "contains":
104        return "contains", ""
105    if cx.INTERFACE_RE.match(s) or cx.TYPE_DEF_RE.match(s) or cx.ENUM_RE.match(s):
106        return "declaration", ""  # (an interface block, a type's or an enum's definition: up to its end)
107    if cx._is_assignment(s):
108        return "simple", ""
109    if re.match(r"^if\s*\(", s):
110        close = cx._close_paren(s, s.index("("))
111        condition, after = s[:close], s[close:].strip()
112        if after == "then":
113            return "if_then", condition
114        if ARITHMETIC_IF_RE.match(after):
115            return "arithmetic_if", after
116        return "if_one", condition + "\0" + after
117    if re.match(r"^else\s*if\s*\(", s):
118        return "else_if", s[:cx._close_paren(s, s.index("("))]
119    if re.match(r"^else\s*$", s):
120        return "else", ""
121    if re.match(r"^else\s*where\s*\(", s):
122        return "else_where_mask", s
123    if re.match(r"^else\s*where\s*$", s):
124        return "else", ""
125    if re.match(r"^where\s*\(", s):
126        rest = s[cx._close_paren(s, s.index("(")):].strip()
127        return ("where_block", s) if not rest else ("if_one", s[:len(s) - len(rest)].strip() + "\0" + rest)
128    if re.match(r"^forall\s*\(", s):
129        rest = s[cx._close_paren(s, s.index("(")):].strip()
130        return ("do", s) if not rest else ("simple", "")
131    found = cx.DO_RE.match(s)
132    if found:
133        return "do", found.group(1) or ""
134    if cx.SELECT_RE.match(s):
135        return "select", s
136    if re.match(r"^(case\s+default|class\s+default|rank\s+default)\b", s):
137        return "case_default", s
138    if cx.GUARD_RE.match(s):
139        return "case", s
140    if re.match(r"^cycle\b", s):
141        return "cycle", ""
142    if re.match(r"^exit\b", s):
143        return "exit", ""
144    if re.match(r"^return\b", s):
145        return "return", ""
146    if re.match(r"^(error\s+)?stop\b", s):
147        return "stop", ""
148    found = GOTO_RE.match(s)
149    if found:
150        return "goto", found.group(1)
151    found = COMPUTED_GOTO_RE.match(s)
152    if found:
153        return "computed_goto", found.group(1)
154    if re.match(r"^(associate\s*\(|block\s*$|critical\b|change\s+team\s*\()", s):
155        return "begin_block", ""
156    return "simple", ""
157
158
159def procedure_statements(text: str, path: Path, start_line: int) -> list[_Statement]:
160    """The statements of the procedure that starts at ``start_line``: its
161    executable part, after its header, up to its ``end`` or ``contains``
162    (interface blocks, type and enum definitions in it left out: declarations).
163    Its included files are read in: their lines numbered as in ``text`` with
164    them read in (see `vibefpmgui.complexity.Source`)."""
165    path = Path(path)
166    return _statements_of(cx.expand_includes(text, path), path, start_line)
167
168
169def _statements_of(source: cx.Source, path: Path, start_line: int) -> list[_Statement]:
170    found = cx.statements(source.text, path.suffix.lower() in cx.FIXED_FORM)
171    out: list[_Statement] = []
172    started = False
173    declaring = 0  # (inside interface blocks, type or enum definitions: declarations)
174    for st in found:
175        if not started:
176            started = source.origin(st.lines[0], path) == (path, start_line)
177            continue
178        kind, detail = _classify(st.text)
179        if kind == "declaration":
180            declaring += 1
181            continue
182        if declaring:
183            declaring -= kind == "end_declaration"
184            continue
185        if kind in ("contains", "end_unit"):
186            break
187        if not out and (SPECIFICATION_RE.match(st.text) or "::" in st.text):
188            continue  # (its declarations: nothing runs there)
189        out.append(_Statement(kind, st.text, st.label, st.lines[0], st.lines[-1], detail))
190    return out
191
192
193# ------------------------------------------------------------------ the graph
194class _Builder:
195    def __init__(self, name: str, source_lines: list[str], fixed: bool = False,
196                 file_of=lambda line: None):
197        self.graph = Graph(name)
198        self.source = source_lines
199        self.fixed = fixed
200        self.file_of = file_of  # (a line's file: a block's lines are all in one)
201        self.start = self._node(START, text=["start"])
202        self.end = self._node(END, text=["end"])
203        self.stop_node: int | None = None
204        self.labels: dict[str, int] = {}          # statement label -> the node it starts
205        self.gotos: list[tuple[int, str, str]] = []  # (node, label it goes to, edge label)
206        self.loops: list[tuple[int, list]] = []   # (loop node, the "exit"s out of it)
207        self.block: int | None = None              # the block statements are being added to
208
209    def _node(self, kind: str, first: int = 0, last: int = 0, text: list[str] | None = None) -> int:
210        node = Node(len(self.graph.nodes), kind, first, last, text or [])
211        self.graph.nodes.append(node)
212        return node.id
213
214    def _edge(self, source: int, target: int, label: str = ""):
215        self.graph.edges.append(Edge(source, target, label))
216
217    def _connect(self, preds: list, target: int):
218        for source, label in preds:
219            self._edge(source, target, label)
220
221    def _line(self, first: int, last: int = 0, whole: bool = False) -> str:
222        """A statement's lines ``first``–``last`` as shown: their code, without
223        comments, continuations joined into one line ("…" if it's too long,
224        unless ``whole``)."""
225        text = ""
226        for number in range(first, max(first, last) + 1):
227            if not 0 < number <= len(self.source):
228                break
229            line = self.source[number - 1]
230            if number > first:
231                if self.fixed:
232                    if line[:1] in ("c", "C", "*", "!", "d", "D") or len(line) < 6:
233                        continue  # (a comment line between)
234                    line = line[6:]
235                elif not _without_comment(line).strip() or line.lstrip().startswith("#"):
236                    continue
237            code = _without_comment(line).strip()
238            if number > first and code.startswith("&"):
239                text = text + code[1:].lstrip() if text.endswith(("(", "[")) else text + " " + code[1:].lstrip()
240            elif number > first:
241                text = text + " " + code
242            else:
243                text = code
244            if not self.fixed and text.endswith("&"):
245                text = text[:-1].rstrip()
246        return text if whole else _shorten(text)
247
248    def _decision(self, st: _Statement, preds: list, kind: str = DECISION, shown: str = "") -> int:
249        self.block = None
250        node = self._node(kind, st.first, st.last, [shown or self._line(st.first, st.last)])
251        self._connect(preds, node)
252        if st.label:
253            self.labels[st.label] = node
254        return node
255
256    def _simple(self, st: _Statement, preds: list) -> list:
257        """A statement without a branch: into the open block (a new one if
258        something else leads in, or it's a jump's target)."""
259        node = self.graph.nodes[self.block] if self.block is not None else None
260        if node is None or st.label or preds != [(node.id, "")] or self.file_of(st.first) != self.file_of(node.first):
261            self.block = self._node(BLOCK, st.first, st.last)
262            node = self.graph.nodes[self.block]
263            self._connect(preds, node.id)
264            if st.label:
265                self.labels[st.label] = node.id
266        node.last = st.last
267        if len(node.text) < MAX_STATEMENTS_SHOWN:
268            node.text.append(st.shown or self._line(st.first, st.last))
269        elif node.text[-1] != "…":
270            node.text.append("…")
271        return [(node.id, "")]
272
273    def _stop(self) -> int:
274        if self.stop_node is None:
275            self.stop_node = self._node(STOP, text=["stop"])
276        return self.stop_node
277
278    def sequence(self, stmts: list[_Statement], i: int, preds: list, until: set[str],
279                 loop_label: str = "") -> tuple[int, list]:
280        """Statements from ``i`` up to one of the kinds in ``until`` (not taken),
281        or a labelled loop's last statement (taken). (index after, the ways
282        out at the end)."""
283        while i < len(stmts):
284            st = stmts[i]
285            if st.kind in until:
286                return i, preds
287            i, preds = self.statement(stmts, i, preds)
288            if loop_label and stmts[i - 1].label == loop_label:
289                return i, preds  # (also when a loop inside ended on the same label: "do 10 … do 10 …")
290        return i, preds
291
292    def statement(self, stmts: list[_Statement], i: int, preds: list) -> tuple[int, list]:
293        st = stmts[i]
294        kind = st.kind
295        if kind in ("simple", "begin_block", "end_block") or kind in ("case", "case_default", "else", "else_if",
296                                                                     "else_where_mask", "end_if", "end_do",
297                                                                     "end_select", "end_where"):
298            if kind in ("simple", "begin_block", "end_block"):
299                return i + 1, self._simple(st, preds) if preds or st.label else preds
300            return i + 1, preds  # (out of place: read as nothing)
301        if kind == "if_one":
302            _condition, inner = st.detail.split("\0", 1)
303            condition_shown, inner_shown = map(_shorten, _split_one_line(self._line(st.first, st.last, whole=True)))
304            node = self._decision(st, preds, shown=condition_shown)
305            inner_kind, inner_detail = _classify(inner)
306            inner_st = _Statement(inner_kind, inner, "", st.first, st.last, inner_detail, inner_shown)
307            _j, inner_out = self.statement([inner_st], 0, [(node, "true")])
308            self.block = None
309            return i + 1, inner_out + [(node, "false")]
310        if kind in ("if_then", "where_block"):
311            ends = {"if_then": ("else_if", "else", "end_if"), "where_block": ("else_where_mask", "else", "end_where")}
312            branch_kinds, end_kind = set(ends[kind][:2]), ends[kind][2]
313            node = self._decision(st, preds)
314            out = []
315            i, branch = self.sequence(stmts, i + 1, [(node, "true")], branch_kinds | {end_kind})
316            out += branch
317            has_else = False
318            while i < len(stmts) and stmts[i].kind in branch_kinds:
319                if stmts[i].kind == "else":
320                    self.block = None
321                    has_else = True
322                    i, branch = self.sequence(stmts, i + 1, [(node, "false")], {end_kind})
323                else:
324                    node = self._decision(stmts[i], [(node, "false")])
325                    i, branch = self.sequence(stmts, i + 1, [(node, "true")], branch_kinds | {end_kind})
326                out += branch
327            if not has_else:
328                out.append((node, "false"))
329            self.block = None
330            return i + 1, out
331        if kind == "select":
332            node = self._decision(st, preds)
333            out, has_default = [], False
334            i += 1
335            while i < len(stmts) and stmts[i].kind in ("simple",) and not stmts[i].label:
336                i += 1  # (nothing runs between "select" and its first case)
337            while i < len(stmts) and stmts[i].kind in ("case", "case_default"):
338                guard = stmts[i]
339                has_default |= guard.kind == "case_default"
340                label = self._line(guard.first, guard.last)
341                self.block = None
342                i, branch = self.sequence(stmts, i + 1, [(node, label)], {"case", "case_default", "end_select"})
343                out += branch
344            if not has_default:
345                out.append((node, "default"))
346            self.block = None
347            return i + 1, out
348        if kind == "do":
349            node = self._decision(st, preds, LOOP)
350            exits: list = []
351            self.loops.append((node, exits))
352            if st.detail:  # ("do 10 i = …": up to the statement labelled 10)
353                i, body = self.sequence(stmts, i + 1, [(node, "loop")], set(), loop_label=st.detail)
354            else:
355                i, body = self.sequence(stmts, i + 1, [(node, "loop")], {"end_do"})
356                i += 1
357            self.loops.pop()
358            self._connect(body, node)
359            self.block = None
360            endless = st.text.strip() == "do"
361            return i, exits + ([] if endless else [(node, "done")])
362        if kind in ("cycle", "exit"):
363            preds = self._simple(st, preds)
364            if self.loops:
365                loop, exits = self.loops[-1]
366                if kind == "cycle":
367                    self._connect(preds, loop)
368                else:
369                    exits.extend(preds)
370            self.block = None
371            return i + 1, []
372        if kind == "return":
373            preds = self._simple(st, preds)
374            self._connect(preds, self.end)
375            self.block = None
376            return i + 1, []
377        if kind == "stop":
378            preds = self._simple(st, preds)
379            self._connect(preds, self._stop())
380            self.block = None
381            return i + 1, []
382        if kind == "goto":
383            preds = self._simple(st, preds)
384            for source, _label in preds:
385                self.gotos.append((source, st.detail, ""))
386            self.block = None
387            return i + 1, []
388        if kind in ("computed_goto", "arithmetic_if"):
389            node = self._decision(st, preds)
390            targets = re.findall(r"\d+", st.detail)
391            names = ["<0", "=0", ">0"] if kind == "arithmetic_if" else [f"{n + 1}" for n in range(len(targets))]
392            ways: dict[str, list[str]] = {}  # (a label given twice: one edge, both names)
393            for target, name in zip(targets, names, strict=False):
394                ways.setdefault(target, []).append(name)
395            for target, labels in ways.items():
396                self.gotos.append((node, target, ", ".join(labels)))
397            return i + 1, [] if kind == "arithmetic_if" else [(node, "else")]
398        return i + 1, self._simple(st, preds)
399
400    def finish(self, exits: list) -> Graph:
401        self._connect(exits, self.end)
402        for source, label, name in self.gotos:
403            if label in self.labels:
404                self._edge(source, self.labels[label], name)
405        return self.graph
406
407
408def _shorten(text: str, width: int = 60) -> str:
409    """``text`` cut to ``width`` characters, the last one "…" if it was cut."""
410    return text if len(text) <= width else text[:width - 1] + "…"
411
412
413def _without_comment(line: str) -> str:
414    """A line without its "!" comment (a "!" in a string isn't one)."""
415    quote = None
416    for i, ch in enumerate(line):
417        if quote:
418            if ch == quote:
419                quote = None
420        elif ch in "'\"":
421            quote = ch
422        elif ch == "!":
423            return line[:i].rstrip()
424    return line
425
426
427def _split_one_line(line: str) -> tuple[str, str]:
428    """A one-line "if (c) statement" (or "where (m) …") as written: ("if (c)",
429    "statement"); ("", "") if it isn't one on a line of its own."""
430    found = re.match(r"\s*(if|where)\s*\(", line, re.IGNORECASE)
431    if not found:
432        return "", ""
433    close = cx._close_paren(line, found.end() - 1)
434    return line[:close].strip(), line[close:].strip()
435
436
437def build(text: str, path: Path | str, procedure: cx.Procedure) -> Graph:
438    """The control-flow graph of a procedure of ``text`` (the file it's in)."""
439    path = Path(path)
440    source = cx.expand_includes(text, path)
441    stmts = _statements_of(source, path, procedure.start)
442    builder = _Builder(procedure.name, source.text.split("\n"), path.suffix.lower() in cx.FIXED_FORM,
443                       lambda line: source.origin(line, path)[0])
444    _i, exits = builder.sequence(stmts, 0, [(builder.start, "")], set())
445    graph = builder.finish(exits)
446    graph.path = path
447    for node in graph.nodes:  # (the lines as read: back to their own files')
448        if node.first:
449            node.path, node.first = source.origin(node.first, path)
450            node.last = source.origin(node.last, path)[1]
451    return graph
452
453
454# ------------------------------------------------------------------ the layout
455@dataclass
456class Layout:
457    positions: dict[int, tuple[float, float]]   # node -> its center
458    layers: list[list[int]]
459    routes: dict[int, list[tuple[float, float]]] = field(default_factory=dict)
460    # ^ an edge (its index) that crosses layers -> the points it goes through, one per layer crossed
461
462
463WAYPOINT = (14.0, 10.0)  # (a long edge's place in a layer it crosses)
464
465
466def layout(graph: Graph, sizes: dict[int, tuple[float, float]], h_gap: float = 40, v_gap: float = 46) -> Layout:
467    """Layers top to bottom from the start node (a loop's back edges set
468    aside: they're marked ``back``), ordered by their neighbours' places. An
469    edge that crosses layers gets a waypoint in each, ordered with the nodes:
470    it goes around them, not through them."""
471    succ: dict[int, list[int]] = {n.id: [] for n in graph.nodes}
472    for edge in graph.edges:
473        succ[edge.source].append(edge.target)
474    # Back edges: to a node still being explored (depth first from the start).
475    state: dict[int, int] = {}  # 1: being explored, 2: done
476    back: set[tuple[int, int]] = set()
477    stack = [(graph.nodes[0].id, iter(succ[graph.nodes[0].id]))] if graph.nodes else []
478    if stack:
479        state[stack[0][0]] = 1
480    while stack:
481        node, children = stack[-1]
482        child = next(children, None)
483        if child is None:
484            state[node] = 2
485            stack.pop()
486        elif state.get(child) == 1:
487            back.add((node, child))
488        elif child not in state:
489            state[child] = 1
490            stack.append((child, iter(succ[child])))
491    for edge in graph.edges:
492        edge.back = (edge.source, edge.target) in back
493    forward = [e for e in graph.edges if not e.back]
494    # Longest path from the start (in the edges left: no cycles).
495    depth = {graph.nodes[0].id: 0} if graph.nodes else {}
496    order = _topological(graph, forward)
497    outgoing: dict[int, list[Edge]] = {n.id: [] for n in graph.nodes}
498    for e in forward:
499        outgoing[e.source].append(e)
500    for node in order:
501        if node in depth:
502            for e in outgoing[node]:
503                depth[e.target] = max(depth.get(e.target, 0), depth[node] + 1)
504    for n in graph.nodes:  # (not reached from the start: a label nothing jumps to)
505        depth.setdefault(n.id, max(depth.values(), default=0) + 1)
506    bottom = max(depth.values(), default=0)
507    for n in graph.nodes:  # (the end and stop nodes at the bottom)
508        if n.kind in (END, STOP):
509            depth[n.id] = bottom
510    sizes = dict(sizes)
511    layers: list[list[int]] = [[] for _ in range(bottom + 1)]
512    for node in order + [n.id for n in graph.nodes if n.id not in order]:
513        if node not in layers[depth[node]]:
514            layers[depth[node]].append(node)
515    parents: dict[int, list[int]] = {n.id: [] for n in graph.nodes}
516    children: dict[int, list[int]] = {n.id: [] for n in graph.nodes}
517    chains: dict[int, list[int]] = {}  # an edge's index -> its waypoints, top to bottom
518    next_id = max((n.id for n in graph.nodes), default=0) + 1
519    for index, e in enumerate(graph.edges):
520        if e.back:
521            continue
522        previous = e.source
523        chain = []
524        for layer in range(depth[e.source] + 1, depth[e.target]):
525            waypoint, next_id = next_id, next_id + 1
526            sizes[waypoint] = WAYPOINT
527            layers[layer].append(waypoint)
528            parents[waypoint], children[waypoint] = [previous], []
529            children[previous].append(waypoint)
530            chain.append(waypoint)
531            previous = waypoint
532        children[previous].append(e.target)
533        parents[e.target].append(previous)
534        if chain:
535            chains[index] = chain
536
537    def barycenter(node, neighbours, index):
538        places = [index[n] for n in neighbours[node] if n in index]
539        return sum(places) / len(places) if places else index.get(node, 0)
540
541    for _ in range(4):
542        for k in range(1, len(layers)):
543            index = {n: j for j, n in enumerate(layers[k - 1])}
544            layers[k].sort(key=lambda n, ix=index: barycenter(n, parents, ix))
545        for k in range(len(layers) - 2, -1, -1):
546            index = {n: j for j, n in enumerate(layers[k + 1])}
547            layers[k].sort(key=lambda n, ix=index: barycenter(n, children, ix))
548    positions: dict[int, tuple[float, float]] = {}
549    y = 0.0
550    for layer in layers:
551        if not layer:
552            continue
553        width = sum(sizes[n][0] for n in layer) + h_gap * (len(layer) - 1)
554        height = max(sizes[n][1] for n in layer)
555        x = -width / 2
556        for n in layer:
557            positions[n] = (x + sizes[n][0] / 2, y + height / 2)
558            x += sizes[n][0] + h_gap
559        y += height + v_gap
560    routes = {index: [positions[w] for w in chain] for index, chain in chains.items()}
561    nodes = {n.id for n in graph.nodes}
562    return Layout({n: p for n, p in positions.items() if n in nodes},
563                  [[n for n in layer if n in nodes] for layer in layers if any(n in nodes for n in layer)], routes)
564
565
566def _topological(graph: Graph, edges: list[Edge]) -> list[int]:
567    incoming = {n.id: 0 for n in graph.nodes}
568    succ: dict[int, list[int]] = {n.id: [] for n in graph.nodes}
569    for e in edges:
570        incoming[e.target] += 1
571        succ[e.source].append(e.target)
572    ready = [n.id for n in graph.nodes if incoming[n.id] == 0]
573    order = []
574    while ready:
575        node = ready.pop(0)
576        order.append(node)
577        for child in succ[node]:
578            incoming[child] -= 1
579            if incoming[child] == 0:
580                ready.append(child)
581    return order