File size: 3,246 Bytes
921d377 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 | """
Path simulator — enumerate every viewer-path through a graph.
Used by:
- QA walker (``qa/path_walker.py`` in batch 8) to verify every
path terminates and has all required assets
- Analytics preview to estimate maximum viewer variety
Safety caps prevent runaway enumeration on malformed graphs:
- max_paths: hard stop after N complete paths
- max_steps: hard stop after N total edge-traversals
These defaults are generous but finite.
"""
from __future__ import annotations
from typing import Iterator, List, Optional
from .graph import BranchGraph
def walk_paths(
graph: BranchGraph,
*,
max_paths: int = 500,
max_steps: int = 10_000,
) -> Iterator[List[str]]:
"""Yield every simple path from entry to an ending, as a list
of node ids (including both endpoints).
'Simple' = no repeated nodes within a single path. Since the
graph is validated acyclic, simple paths exhaust reachable
endings without infinite loops.
Stops cleanly at either cap and silently returns the partial
enumeration — use ``enumerate_paths`` when you need the count
with a cap indicator.
"""
entry = graph.entry()
if not entry:
return
# Adjacency map keyed by from_id; ordered by (ordinal, then insertion).
adj: dict = {}
for e in graph.edges:
adj.setdefault(e.from_id, []).append(e)
for from_id, es in adj.items():
es.sort(key=lambda x: (x.ordinal, 0))
steps = 0
paths_yielded = 0
stack: List[List[str]] = [[entry.id]]
while stack:
if paths_yielded >= max_paths:
return
if steps >= max_steps:
return
path = stack.pop()
tail_id = path[-1]
tail = graph.node(tail_id)
if tail is None:
continue
# Ending nodes → yield the path and don't expand.
if tail.kind == "ending":
yield list(path)
paths_yielded += 1
continue
outs = adj.get(tail_id, [])
if not outs:
# Dead end (shouldn't happen on a validated graph, but
# don't crash — treat as a path terminus).
yield list(path)
paths_yielded += 1
continue
# Iterate in reverse so leftmost ordinal is processed first
# when popped off the stack.
for e in reversed(outs):
steps += 1
if e.to_id in path:
# Simple-path invariant — skip revisits.
continue
stack.append(list(path) + [e.to_id])
def enumerate_paths(
graph: BranchGraph, *, max_paths: int = 500, max_steps: int = 10_000,
) -> dict:
"""Count enumeration with cap indicators.
Returns a dict: ``{'paths': [...], 'count': int, 'truncated': bool}``
where ``truncated=True`` means the walker hit one of the caps.
"""
collected: List[List[str]] = []
count = 0
for p in walk_paths(graph, max_paths=max_paths, max_steps=max_steps):
collected.append(p)
count += 1
# Heuristic for 'truncated': if we returned exactly the cap,
# there were probably more paths.
truncated = count >= max_paths
return {"paths": collected, "count": count, "truncated": truncated}
|