Coverage for vibefpmgui/control

1"""A Fortran procedure's control-flow graph, and a layout for drawing it (no Qt).

3`build` turns a procedure's statements (read as `vibefpmgui.complexity`

4reads them) into nodes and edges:

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.

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``.

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).

24`layout` places the nodes in layers, top to bottom (loops' back edges set

25aside), ordered to cross few edges.

26"""

28from __future__ import annotations

30import re

31from dataclasses import dataclass, field

32from pathlib import Path

34from . import complexity as cx

36START, END, STOP, BLOCK, DECISION, LOOP = "start", "end", "stop", "block", "decision", "loop"

37MAX_STATEMENTS_SHOWN = 4 # a block's lines shown (then "…")

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)

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)

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)

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

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)

85# The specification part: declarations and the like, before the first statement that runs.

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+)$")

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): 121 ↛ 122line 121 didn't jump to line 122 because the condition on line 121 was never true

122 return "else_where_mask", s

123 if re.match(r"^else\s*where\s*$", s): 123 ↛ 124line 123 didn't jump to line 124 because the condition on line 123 was never true

124 return "else", ""

125 if re.match(r"^where\s*\(", s): 125 ↛ 126line 125 didn't jump to line 126 because the condition on line 125 was never true

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): 128 ↛ 129line 128 didn't jump to line 129 because the condition on line 128 was never true

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: 149 ↛ 150line 149 didn't jump to line 150 because the condition on line 149 was never true

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: 174 ↛ 190line 174 didn't jump to line 190 because the loop on line 174 didn't complete

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): 227 ↛ 228line 227 didn't jump to line 228 because the condition on line 227 was never true

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("#"): 235 ↛ 236line 235 didn't jump to line 236 because the condition on line 235 was never true

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: 274 ↛ 276line 274 didn't jump to line 276 because the condition on line 274 was always true

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"): 298 ↛ 300line 298 didn't jump to line 300 because the condition on line 298 was always true

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: 335 ↛ 336line 335 didn't jump to line 336 because the condition on line 335 was never true

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: 364 ↛ 370line 364 didn't jump to line 370 because the condition on line 364 was always true

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": 382 ↛ 383line 382 didn't jump to line 383 because the condition on line 382 was never true

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"): 388 ↛ 398line 388 didn't jump to line 398 because the condition on line 388 was always true

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: 403 ↛ 402line 403 didn't jump to line 402 because the condition on line 403 was always true

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: 431 ↛ 432line 431 didn't jump to line 432 because the condition on line 431 was never true

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: 478 ↛ 480line 478 didn't jump to line 480 because the condition on line 478 was always true

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: 501 ↛ 500line 501 didn't jump to line 500 because the condition on line 501 was always true

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]]: 513 ↛ 512line 513 didn't jump to line 512 because the condition on line 513 was always true

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: 551 ↛ 552line 551 didn't jump to line 552 because the condition on line 551 was never true

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

Coverage for vibefpmgui/control_flow.py: 94%

456 statements