Coverage for collectra/gui/pipeline_order.py: 95.83%

1"""A reading order for the pipeline's DAG labels. 

2 

3The DAG is drawn in layers from the top down, and left to right within a 

4layer. This gives each label a rank that follows that order, so lists of 

5matches can be walked the way the pipeline reads. 

6""" 

7 

8 

9def pipeline_label_ranks(graph: dict) -> dict[str, int]: 

10 """Rank every node of a ``gui_graph()`` dictionary, 0 first. 

11 

12 A node's layer is the longest chain of parents above it. Within a layer, 

13 nodes keep the order in which the edges first reach them (and, for nodes 

14 with no parent, their order in the node list). 

15 """ 

16 ids = [node["id"] for node in graph.get("nodes", [])] 

17 edges = [(e["source"], e["target"]) for e in graph.get("edges", [])] 

18 for source, target in edges: 

19 for node_id in (source, target): 

20 if node_id not in ids: 

21 ids.append(node_id) 

22 node_index = {node_id: i for i, node_id in enumerate(ids)} 

23 

24 parents = {node_id: [] for node_id in ids} 

25 first_reached: dict[str, int] = {} 

26 for position, (source, target) in enumerate(edges): 

27 parents[target].append(source) 

28 first_reached.setdefault(target, position) 

29 

30 depth: dict[str, int] = {} 

31 

32 def layer(node_id: str, path: frozenset) -> int: 

33 if node_id in depth: 

34 return depth[node_id] 

35 above = [p for p in parents[node_id] if p not in path] # ignore cycles 

36 depth[node_id] = ( 

37 1 + max(layer(p, path | {node_id}) for p in above) if above else 0 

38 ) 

39 return depth[node_id] 

40 

41 for node_id in ids: 

42 layer(node_id, frozenset()) 

43 

44 ordered = sorted( 

45 ids, 

46 key=lambda n: (depth[n], first_reached.get(n, -1), node_index[n]), 

47 ) 

48 return {node_id: rank for rank, node_id in enumerate(ordered)}