Coverage for python/lsst/images/diagram.py: 97%

374 statements  

« prev     ^ index     » next       coverage.py v7.14.3, created at 2026-06-29 22:40 +0000

1# This file is part of lsst-images. 

2# 

3# Developed for the LSST Data Management System. 

4# This product includes software developed by the LSST Project 

5# (https://www.lsst.org). 

6# See the COPYRIGHT file at the top-level directory of this distribution 

7# for details of code ownership. 

8# 

9# Use of this source code is governed by a 3-clause BSD-style 

10# license that can be found in the LICENSE file. 

11"""Turn `pydantic` serialization models into composition diagrams.""" 

12 

13from __future__ import annotations 

14 

15__all__ = ( 

16 "DEFAULT_LEAF_TYPES", 

17 "Attribute", 

18 "Graph", 

19 "Node", 

20 "Policy", 

21 "Reference", 

22 "build_graph", 

23 "build_instance_graph", 

24 "graph_from_file", 

25 "make_policy", 

26 "render", 

27) 

28 

29import dataclasses 

30import re 

31import types 

32from collections.abc import Iterable, Mapping, Sequence, Set 

33from typing import TypeAliasType, TypeGuard, Union, get_args, get_origin 

34 

35import pydantic 

36 

37from .serialization import open as open_archive 

38 

39#: Serialization-plumbing helper models collapsed to leaves by default; these 

40#: carry array/table payloads rather than meaningful model composition. 

41DEFAULT_LEAF_TYPES: frozenset[str] = frozenset( 

42 { 

43 "ArrayReferenceModel", 

44 "ArrayReferenceQuantityModel", 

45 "InlineArrayModel", 

46 "InlineArrayQuantityModel", 

47 "QuantityModel", 

48 "TimeModel", 

49 "TableModel", 

50 "TableColumnModel", 

51 # Per-backend array-payload pointers seen when diagramming a file. 

52 "JsonRef", 

53 "PointerModel", 

54 "NdfPointerModel", 

55 } 

56) 

57 

58_SEQUENCE_ORIGINS: frozenset[object] = frozenset({list, set, frozenset, tuple, Sequence, Set}) 

59_MAPPING_ORIGINS: frozenset[object] = frozenset({dict, Mapping}) 

60_UNION_ORIGINS: frozenset[object] = frozenset({Union, types.UnionType}) 

61 

62# Compact, format-safe cardinality markers appended to a field name in edge 

63# labels: ``*`` a list, ``+`` a mapping, ``?`` optional. (Brackets and braces 

64# are avoided because they are structural in dot and mermaid.) 

65_CARDINALITY_MARKER: dict[str, str] = {"one": "", "optional": "?", "list": "*", "dict": "+"} 

66 

67 

68@dataclasses.dataclass 

69class Attribute: 

70 """A scalar field rendered as an attribute inside a model's node.""" 

71 

72 name: str 

73 type_str: str 

74 

75 

76@dataclasses.dataclass 

77class Reference: 

78 """A model-valued field rendered as a composition edge to other nodes.""" 

79 

80 name: str 

81 targets: list[str] 

82 cardinality: str 

83 has_other: bool = False 

84 

85 

86@dataclasses.dataclass 

87class Node: 

88 """A model class in the diagram.""" 

89 

90 key: str 

91 label: str 

92 attributes: list[Attribute] 

93 references: list[Reference] 

94 collapsed: bool = False 

95 

96 

97@dataclasses.dataclass 

98class Graph: 

99 """A composition graph: a root model and all models reachable from it.""" 

100 

101 root: str 

102 nodes: dict[str, Node] 

103 

104 

105@dataclasses.dataclass(frozen=True) 

106class Policy: 

107 """Controls how much of each model the walk records. 

108 

109 Models whose type name (the unparameterized class name) is in ``leaves`` 

110 are rendered as leaf nodes without expanding their fields. When 

111 ``include_attributes`` is false, scalar (non-model) fields are dropped, so 

112 the diagram shows only model composition and all-scalar models such as 

113 ``ObservationInfo`` become leaves. 

114 """ 

115 

116 leaves: frozenset[str] = DEFAULT_LEAF_TYPES 

117 include_attributes: bool = True 

118 hide_fields: frozenset[str] = frozenset() 

119 hide_types: frozenset[str] = frozenset() 

120 public_names: bool = False 

121 

122 

123def make_policy( 

124 expand_leaves: bool = False, 

125 expand: Iterable[str] = (), 

126 collapse: Iterable[str] = (), 

127 include_attributes: bool = True, 

128 hide_fields: Iterable[str] = (), 

129 hide_types: Iterable[str] = (), 

130 public_names: bool = False, 

131) -> Policy: 

132 """Build a `Policy` from the default leaf set and CLI-style overrides. 

133 

134 Parameters 

135 ---------- 

136 expand_leaves 

137 Start from an empty leaf set instead of `DEFAULT_LEAF_TYPES`. 

138 expand 

139 Type names to remove from the leaf set (force expansion). 

140 collapse 

141 Type names to add to the leaf set (force collapse). 

142 include_attributes 

143 Record scalar (non-model) fields as node attributes. When false the 

144 diagram shows only model composition. 

145 hide_fields 

146 Field names to drop entirely, along with any sub-tree reachable only 

147 through them. 

148 hide_types 

149 Type names (public or serialization) to drop entirely, removing every 

150 edge that points at them. 

151 public_names 

152 Label nodes with the public in-memory class name (e.g. 

153 ``SkyProjection``) instead of the serialization model name, where one 

154 is available. 

155 """ 

156 leaves = set() if expand_leaves else set(DEFAULT_LEAF_TYPES) 

157 leaves -= set(expand) 

158 leaves |= set(collapse) 

159 return Policy( 

160 leaves=frozenset(leaves), 

161 include_attributes=include_attributes, 

162 hide_fields=frozenset(hide_fields), 

163 hide_types=frozenset(hide_types), 

164 public_names=public_names, 

165 ) 

166 

167 

168def _key(cls: type) -> str: 

169 """Return a stable unique key for ``cls``.""" 

170 return f"{cls.__module__}.{cls.__qualname__}" 

171 

172 

173def _label(cls: type) -> str: 

174 """Return the display label for ``cls`` (with any generic parameters).""" 

175 return cls.__name__ 

176 

177 

178def _type_name(cls: type) -> str: 

179 """Return the unparameterized class name used for policy matching.""" 

180 metadata = getattr(cls, "__pydantic_generic_metadata__", None) 

181 if metadata and metadata.get("origin") is not None: 

182 return metadata["origin"].__name__ 

183 return cls.__name__ 

184 

185 

186def _public_name(cls: type) -> str | None: 

187 """Return the public in-memory class name for ``cls``, if any. 

188 

189 Serialization models expose a ``PUBLIC_TYPE`` ClassVar naming the class 

190 their data deserializes to. Builtins (e.g. ``dict``) are ignored so the 

191 model name is kept instead. 

192 """ 

193 public = getattr(cls, "PUBLIC_TYPE", None) 

194 if isinstance(public, type) and public.__module__ != "builtins": 

195 return public.__name__ 

196 return None 

197 

198 

199def _node_label(cls: type, policy: Policy) -> str: 

200 """Return the node label, preferring the public class name if requested.""" 

201 if policy.public_names and (public := _public_name(cls)) is not None: 

202 return public 

203 return _label(cls) 

204 

205 

206def _type_names(cls: type) -> set[str]: 

207 """Return the names that match a leaf/collapse directive for ``cls``. 

208 

209 Both the serialization model name and the public class name match, so 

210 ``--collapse Image`` works whether or not public names are displayed. 

211 """ 

212 names = {_type_name(cls)} 

213 if (public := _public_name(cls)) is not None: 

214 names.add(public) 

215 return names 

216 

217 

218def _is_hidden_type(cls: type, policy: Policy) -> bool: 

219 """Return whether ``cls`` should be dropped from the diagram entirely.""" 

220 return bool(_type_names(cls) & policy.hide_types) 

221 

222 

223def _resolve(annotation: object) -> object: 

224 """Unwrap ``TypeAliasType`` and ``Annotated`` layers to the real type. 

225 

226 PEP 695 aliases (e.g. ``FieldSerializationModel``) and ``Annotated`` 

227 wrappers (discriminated unions, validators) hide the underlying union or 

228 model from naive introspection. 

229 """ 

230 while True: 

231 if isinstance(annotation, TypeAliasType): 

232 annotation = annotation.__value__ 

233 elif hasattr(annotation, "__metadata__"): 

234 annotation = getattr(annotation, "__origin__") 

235 else: 

236 return annotation 

237 

238 

239def _strip_annotated(annotation: object) -> object: 

240 """Unwrap ``Annotated`` layers while preserving type aliases.""" 

241 while hasattr(annotation, "__metadata__"): 

242 annotation = getattr(annotation, "__origin__") 

243 return annotation 

244 

245 

246def _is_model(annotation: object) -> TypeGuard[type[pydantic.BaseModel]]: 

247 """Return whether ``annotation`` is a `pydantic.BaseModel` subclass.""" 

248 return isinstance(annotation, type) and issubclass(annotation, pydantic.BaseModel) 

249 

250 

251def _type_str(annotation: object) -> str: 

252 """Return a readable, module-stripped string for a scalar annotation.""" 

253 # Use an alias's own name rather than expanding it: aliases such as the 

254 # JSON-like ``MetadataValue`` are recursive and would not terminate. 

255 if isinstance(annotation, TypeAliasType): 

256 return annotation.__name__ 

257 if hasattr(annotation, "__metadata__"): 

258 return _type_str(getattr(annotation, "__origin__")) 

259 if annotation is type(None): 

260 return "None" 

261 if annotation is Ellipsis: 

262 return "..." 

263 origin = get_origin(annotation) 

264 if origin in _UNION_ORIGINS: 

265 return " | ".join(_type_str(arg) for arg in get_args(annotation)) 

266 if origin is not None: 

267 args = get_args(annotation) 

268 origin_name = _type_str(origin) 

269 if args: 269 ↛ 271line 269 didn't jump to line 271 because the condition on line 269 was always true

270 return f"{origin_name}[{', '.join(_type_str(arg) for arg in args)}]" 

271 return origin_name 

272 name = getattr(annotation, "__name__", None) 

273 if isinstance(name, str): 

274 return name 

275 return str(annotation).replace("typing.", "") 

276 

277 

278def _models_in( 

279 args: tuple[object, ...], seen_aliases: frozenset[int] = frozenset() 

280) -> tuple[list[type], bool]: 

281 """Split union members into model types and a "something else" flag. 

282 

283 Nested unions and containers are flattened; `None` is ignored (the 

284 optionality is captured by the caller); anything else sets the "other" 

285 flag. 

286 """ 

287 models: list[type] = [] 

288 has_other = False 

289 for arg in args: 

290 arg = _strip_annotated(arg) 

291 if isinstance(arg, TypeAliasType): 

292 alias_id = id(arg) 

293 if alias_id in seen_aliases: 

294 has_other = True 

295 continue 

296 sub_models, sub_other = _models_in((arg.__value__,), seen_aliases | {alias_id}) 

297 models.extend(sub_models) 

298 has_other = has_other or sub_other 

299 continue 

300 if arg is type(None): 

301 continue 

302 if _is_model(arg): 

303 models.append(arg) 

304 elif get_origin(arg) in _UNION_ORIGINS: 

305 sub_models, sub_other = _models_in(get_args(arg), seen_aliases) 

306 models.extend(sub_models) 

307 has_other = has_other or sub_other 

308 elif get_origin(arg) in _SEQUENCE_ORIGINS: 

309 sub_models, sub_other = _models_in(get_args(arg), seen_aliases) 

310 models.extend(sub_models) 

311 has_other = has_other or sub_other 

312 elif get_origin(arg) in _MAPPING_ORIGINS: 

313 mapping_args = get_args(arg) 

314 value = (mapping_args[1],) if len(mapping_args) == 2 else () 

315 sub_models, sub_other = _models_in(value, seen_aliases) 

316 models.extend(sub_models) 

317 has_other = has_other or sub_other 

318 else: 

319 has_other = True 

320 # Preserve order while removing duplicate model types. 

321 return list(dict.fromkeys(models)), has_other 

322 

323 

324def _classify(annotation: object) -> tuple[list[type], str, bool] | None: 

325 """Classify a field annotation as a model reference or a scalar. 

326 

327 Returns ``(model_types, cardinality, has_other)`` for a field that 

328 references one or more models, or `None` for a pure scalar field. 

329 """ 

330 annotation = _resolve(annotation) 

331 origin = get_origin(annotation) 

332 if origin in _UNION_ORIGINS: 

333 args = get_args(annotation) 

334 models, has_other = _models_in(args) 

335 if not models: 

336 return None 

337 cardinality = "optional" if type(None) in args else "one" 

338 return models, cardinality, has_other 

339 if origin in _SEQUENCE_ORIGINS: 

340 models, has_other = _models_in(get_args(annotation)) 

341 if not models: 

342 return None 

343 return models, "list", has_other 

344 if origin in _MAPPING_ORIGINS: 

345 args = get_args(annotation) 

346 value = (args[1],) if len(args) == 2 else () 

347 models, has_other = _models_in(value) 

348 if not models: 

349 return None 

350 return models, "dict", has_other 

351 if _is_model(annotation): 

352 return [annotation], "one", False 

353 return None 

354 

355 

356def build_graph(model_cls: type[pydantic.BaseModel], policy: Policy | None = None) -> Graph: 

357 """Walk ``model_cls`` and its sub-models into a composition `Graph`. 

358 

359 Parameters 

360 ---------- 

361 model_cls 

362 Pydantic model class to diagram. 

363 policy 

364 Policy controlling leaf collapsing, hidden fields, and labels; a 

365 default `Policy` is used when `None`. 

366 """ 

367 if policy is None: 

368 policy = Policy() 

369 nodes: dict[str, Node] = {} 

370 root = _walk(model_cls, nodes, policy) 

371 return Graph(root=root, nodes=nodes) 

372 

373 

374def _walk(cls: type[pydantic.BaseModel], nodes: dict[str, Node], policy: Policy) -> str: 

375 """Add ``cls`` (and its sub-models) to ``nodes``; return its key.""" 

376 key = _key(cls) 

377 if key in nodes: 

378 return key 

379 node = Node(key=key, label=_node_label(cls, policy), attributes=[], references=[]) 

380 nodes[key] = node 

381 if _type_names(cls) & policy.leaves: 

382 node.collapsed = True 

383 return key 

384 for name, field in cls.model_fields.items(): 

385 if name in policy.hide_fields: 

386 continue 

387 annotation = field.annotation 

388 classified = _classify(annotation) 

389 if classified is None: 

390 if policy.include_attributes: 

391 node.attributes.append(Attribute(name=name, type_str=_type_str(annotation))) 

392 continue 

393 model_types, cardinality, has_other = classified 

394 model_types = [model for model in model_types if not _is_hidden_type(model, policy)] 

395 if not model_types and not has_other: 

396 continue 

397 targets = [_walk(model, nodes, policy) for model in model_types] 

398 node.references.append( 

399 Reference(name=name, targets=targets, cardinality=cardinality, has_other=has_other) 

400 ) 

401 return key 

402 

403 

404def build_instance_graph(instance: pydantic.BaseModel, policy: Policy | None = None) -> Graph: 

405 """Walk a model *instance* into a composition `Graph`. 

406 

407 Unlike `build_graph`, which works from field annotations, this follows the 

408 actual values, so unions collapse to the concrete member present and 

409 lists/dicts expand only the element types that actually occur. 

410 

411 Parameters 

412 ---------- 

413 instance 

414 Model instance to diagram. 

415 policy 

416 Policy controlling leaf collapsing, hidden fields, and labels; a 

417 default `Policy` is used when `None`. 

418 """ 

419 if policy is None: 

420 policy = Policy() 

421 nodes: dict[str, Node] = {} 

422 root = _walk_instance(instance, nodes, policy, visited=set()) 

423 return Graph(root=root, nodes=nodes) 

424 

425 

426def graph_from_file(path: str, policy: Policy | None = None) -> Graph: 

427 """Build an instance `Graph` from a serialized ``lsst.images`` file. 

428 

429 Reads only the on-disk reference tree (pointers, not pixel data), so this 

430 is cheap even for large images. 

431 

432 Parameters 

433 ---------- 

434 path 

435 Path to the serialized ``lsst.images`` file. 

436 policy 

437 Policy controlling leaf collapsing, hidden fields, and labels; a 

438 default `Policy` is used when `None`. 

439 """ 

440 with open_archive(path) as reader: 

441 return build_instance_graph(reader.get_tree(), policy) 

442 

443 

444def _walk_instance( 

445 instance: pydantic.BaseModel, nodes: dict[str, Node], policy: Policy, visited: set[int] 

446) -> str: 

447 """Add ``instance`` (and the models it holds) to ``nodes``; return key.""" 

448 cls = type(instance) 

449 key = _key(cls) 

450 if id(instance) in visited: 450 ↛ 451line 450 didn't jump to line 451 because the condition on line 450 was never true

451 return key 

452 visited.add(id(instance)) 

453 if (node := nodes.get(key)) is None: 

454 node = Node(key=key, label=_node_label(cls, policy), attributes=[], references=[]) 

455 nodes[key] = node 

456 if _type_names(cls) & policy.leaves: 

457 node.collapsed = True 

458 return key 

459 for name in type(instance).model_fields: 

460 if name in policy.hide_fields: 460 ↛ 461line 460 didn't jump to line 461 because the condition on line 460 was never true

461 continue 

462 value = getattr(instance, name) 

463 _add_instance_field(node, name, value, nodes, policy, visited) 

464 return key 

465 

466 

467def _merge_reference(node: Node, reference: Reference) -> None: 

468 """Merge a newly observed instance reference into ``node``.""" 

469 node.attributes = [attribute for attribute in node.attributes if attribute.name != reference.name] 

470 for existing in node.references: 

471 if existing.name == reference.name and existing.cardinality == reference.cardinality: 

472 for target in reference.targets: 

473 if target not in existing.targets: 

474 existing.targets.append(target) 

475 existing.has_other = existing.has_other or reference.has_other 

476 return 

477 node.references.append(reference) 

478 

479 

480def _merge_attribute(node: Node, attribute: Attribute) -> None: 

481 """Merge a newly observed scalar instance attribute into ``node``.""" 

482 if any(reference.name == attribute.name for reference in node.references): 482 ↛ 483line 482 didn't jump to line 483 because the condition on line 482 was never true

483 return 

484 for existing in node.attributes: 

485 if existing.name == attribute.name: 

486 if attribute.type_str not in existing.type_str.split(" | "): 486 ↛ 487line 486 didn't jump to line 487 because the condition on line 486 was never true

487 existing.type_str += f" | {attribute.type_str}" 

488 return 

489 node.attributes.append(attribute) 

490 

491 

492def _add_instance_field( 

493 node: Node, 

494 name: str, 

495 value: object, 

496 nodes: dict[str, Node], 

497 policy: Policy, 

498 visited: set[int], 

499) -> None: 

500 """Classify a field *value* as a concrete-model reference or scalar. 

501 

502 Instance mode reports only what the file actually holds: an empty container 

503 yields no edge, not the declared element types. 

504 """ 

505 if isinstance(value, pydantic.BaseModel): 

506 if _is_hidden_type(type(value), policy): 506 ↛ 507line 506 didn't jump to line 507 because the condition on line 506 was never true

507 return 

508 target = _walk_instance(value, nodes, policy, visited) 

509 _merge_reference(node, Reference(name=name, targets=[target], cardinality="one")) 

510 return 

511 if isinstance(value, Mapping): 

512 models = [v for v in value.values() if isinstance(v, pydantic.BaseModel)] 

513 cardinality = "dict" 

514 elif isinstance(value, (list, tuple, set, frozenset)): 

515 models = [v for v in value if isinstance(v, pydantic.BaseModel)] 

516 cardinality = "list" 

517 else: 

518 models = [] 

519 cardinality = "" 

520 models = [model for model in models if not _is_hidden_type(type(model), policy)] 

521 if models: 

522 targets: list[str] = [] 

523 for model in models: 

524 target = _walk_instance(model, nodes, policy, visited) 

525 if target not in targets: 

526 targets.append(target) 

527 _merge_reference(node, Reference(name=name, targets=targets, cardinality=cardinality)) 

528 elif not policy.include_attributes: 

529 return 

530 elif value is None: 

531 _merge_attribute(node, Attribute(name=name, type_str="None")) 

532 else: 

533 _merge_attribute(node, Attribute(name=name, type_str=type(value).__name__)) 

534 

535 

536def render(graph: Graph, fmt: str) -> str: 

537 """Render ``graph`` as ``fmt`` text (``dot``, ``mermaid`` or ``tree``). 

538 

539 Parameters 

540 ---------- 

541 graph 

542 Composition graph to render. 

543 fmt 

544 Output format: ``dot``, ``mermaid`` or ``tree``. 

545 """ 

546 emitters = {"dot": _to_dot, "mermaid": _to_mermaid, "tree": _to_tree} 

547 try: 

548 emitter = emitters[fmt] 

549 except KeyError: 

550 choices = ", ".join(sorted(emitters)) 

551 raise ValueError(f"Unknown diagram format {fmt!r}; choose from {choices}.") from None 

552 return emitter(graph) 

553 

554 

555def _edge_label(reference: Reference) -> str: 

556 """Return the field-name edge label with cardinality/other markers.""" 

557 label = reference.name + _CARDINALITY_MARKER[reference.cardinality] 

558 if reference.has_other: 

559 label += " (+other)" 

560 return label 

561 

562 

563def _node_ids(graph: Graph) -> dict[str, str]: 

564 """Map node keys to dot/mermaid-safe identifiers (alphanumeric + ``_``).""" 

565 ids: dict[str, str] = {} 

566 used: set[str] = set() 

567 for key, node in graph.nodes.items(): 

568 base = re.sub(r"\W", "_", node.label) or "node" 

569 if base[0].isdigit(): 569 ↛ 570line 569 didn't jump to line 570 because the condition on line 569 was never true

570 base = "_" + base 

571 candidate = base 

572 suffix = 1 

573 while candidate in used: 573 ↛ 574line 573 didn't jump to line 574 because the condition on line 573 was never true

574 suffix += 1 

575 candidate = f"{base}_{suffix}" 

576 used.add(candidate) 

577 ids[key] = candidate 

578 return ids 

579 

580 

581def _dot_escape(text: str) -> str: 

582 """Escape characters that are structural in a dot record label.""" 

583 return "".join("\\" + ch if ch in '\\{}|<>"' else ch for ch in text) 

584 

585 

586def _to_dot(graph: Graph) -> str: 

587 ids = _node_ids(graph) 

588 lines = [ 

589 f'digraph "{_dot_escape(graph.nodes[graph.root].label)}" {{', 

590 " rankdir=LR;", 

591 " node [shape=record];", 

592 ] 

593 for key, node in graph.nodes.items(): 

594 attrs = "".join(f"{_dot_escape(a.name)} : {_dot_escape(a.type_str)}\\l" for a in node.attributes) 

595 body = "{" + _dot_escape(node.label) + ("|" + attrs if attrs else "") + "}" 

596 lines.append(f' {ids[key]} [label="{body}"];') 

597 for key, node in graph.nodes.items(): 

598 for reference in node.references: 

599 label = _dot_escape(_edge_label(reference)) 

600 for target in reference.targets: 

601 lines.append(f' {ids[key]} -> {ids[target]} [label="{label}"];') 

602 lines.append("}") 

603 return "\n".join(lines) + "\n" 

604 

605 

606def _mermaid_escape(text: str) -> str: 

607 """Make ``text`` safe for mermaid labels and class-member lines. 

608 

609 Brackets and braces are structural in mermaid; square brackets become the 

610 ``~`` generic markers and curly braces become parentheses so generic and 

611 container types survive without breaking the parser. 

612 """ 

613 return text.replace('"', "'").replace("[", "~").replace("]", "~").replace("{", "(").replace("}", ")") 

614 

615 

616def _to_mermaid(graph: Graph) -> str: 

617 ids = _node_ids(graph) 

618 lines = ["classDiagram"] 

619 for key, node in graph.nodes.items(): 

620 lines.append(f' class {ids[key]}["{_mermaid_escape(node.label)}"] {{') 

621 for attribute in node.attributes: 

622 lines.append(f" +{_mermaid_escape(attribute.type_str)} {_mermaid_escape(attribute.name)}") 

623 lines.append(" }") 

624 for key, node in graph.nodes.items(): 

625 for reference in node.references: 

626 label = _mermaid_escape(_edge_label(reference)) 

627 for target in reference.targets: 

628 lines.append(f" {ids[key]} --> {ids[target]} : {label}") 

629 return "\n".join(lines) + "\n" 

630 

631 

632def _to_tree(graph: Graph) -> str: 

633 lines = [graph.nodes[graph.root].label] 

634 _tree_children(graph, graph.root, prefix="", path=frozenset({graph.root}), lines=lines) 

635 return "\n".join(lines) + "\n" 

636 

637 

638def _tree_children(graph: Graph, key: str, prefix: str, path: frozenset[str], lines: list[str]) -> None: 

639 """Append the tree(1)-style child lines of ``key`` to ``lines``.""" 

640 node = graph.nodes[key] 

641 items: list[tuple[str, object]] = [("attribute", a) for a in node.attributes] 

642 items += [("reference", r) for r in node.references] 

643 for index, (kind, obj) in enumerate(items): 

644 last = index == len(items) - 1 

645 branch = "└── " if last else "├── " 

646 child_prefix = prefix + (" " if last else "│ ") 

647 if kind == "attribute": 

648 assert isinstance(obj, Attribute) 

649 lines.append(f"{prefix}{branch}{obj.name}: {obj.type_str}") 

650 continue 

651 assert isinstance(obj, Reference) 

652 marker = _CARDINALITY_MARKER[obj.cardinality] 

653 if len(obj.targets) == 1 and not obj.has_other: 

654 _tree_single(graph, obj, marker, prefix, branch, child_prefix, path, lines) 

655 else: 

656 _tree_union(graph, obj, marker, prefix, branch, child_prefix, path, lines) 

657 

658 

659def _tree_single( 

660 graph: Graph, 

661 reference: Reference, 

662 marker: str, 

663 prefix: str, 

664 branch: str, 

665 child_prefix: str, 

666 path: frozenset[str], 

667 lines: list[str], 

668) -> None: 

669 target = reference.targets[0] 

670 label = graph.nodes[target].label 

671 if target in path: 

672 lines.append(f"{prefix}{branch}{reference.name}{marker}: {label} (↻)") 

673 return 

674 lines.append(f"{prefix}{branch}{reference.name}{marker}: {label}") 

675 _tree_children(graph, target, child_prefix, path | {target}, lines) 

676 

677 

678def _tree_union( 

679 graph: Graph, 

680 reference: Reference, 

681 marker: str, 

682 prefix: str, 

683 branch: str, 

684 child_prefix: str, 

685 path: frozenset[str], 

686 lines: list[str], 

687) -> None: 

688 lines.append(f"{prefix}{branch}{reference.name}{marker} (one of):") 

689 total = len(reference.targets) + (1 if reference.has_other else 0) 

690 for index, target in enumerate(reference.targets): 

691 member_last = index == total - 1 

692 member_branch = "└── " if member_last else "├── " 

693 member_prefix = child_prefix + (" " if member_last else "│ ") 

694 label = graph.nodes[target].label 

695 if target in path: 

696 lines.append(f"{child_prefix}{member_branch}{label} (↻)") 

697 else: 

698 lines.append(f"{child_prefix}{member_branch}{label}") 

699 _tree_children(graph, target, member_prefix, path | {target}, lines) 

700 if reference.has_other: 

701 lines.append(f"{child_prefix}└── …(other)")