Add Sugiyama-lite layout optimization for inter-module diagrams
Layered graph drawing approach to minimize cable crossings: - Layer assignment via BFS from external connectors (J*, TP*, P*) - Barycenter ordering (forward + backward sweep) within each layer - Connection orientation: earlier connector always on the left - Pin reordering: weighted average neighbor position groups pins by destination, reducing within-connector crossings - Connection sorting: shorter (adjacent-pair) connections first Fixes from Apollo safety review: - Explicit ValueError on pin remapping failures (was silent KeyError) - Weighted pin averaging for star topology (GND shared across modules) - Fully deterministic sort keys for reproducible output - Documented closure capture pattern in loop sort keys
This commit is contained in:
parent
b9154e851b
commit
b8ff2d19da
5 changed files with 886 additions and 1 deletions
|
|
@ -6,6 +6,12 @@ Parallel wires between the same pair of modules are grouped into
|
|||
multi-wire cables for cleaner diagrams.
|
||||
|
||||
Also includes top-level J*/TP*/P* connectors directly.
|
||||
|
||||
Layout optimization uses a Sugiyama-lite approach:
|
||||
1. Layer assignment (externals at layer 0, BFS outward for modules)
|
||||
2. Barycenter ordering within each layer
|
||||
3. Connection orientation (left = earlier in ordering)
|
||||
4. Pin reordering (group by neighbor, order by neighbor position)
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
|
@ -36,6 +42,343 @@ def _make_connector_name(ref: str) -> str:
|
|||
return ref.replace(" ", "_")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Layout optimization helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _build_adjacency(
|
||||
connections: list[list[dict[str, Any]]],
|
||||
) -> dict[str, dict[str, int]]:
|
||||
"""Build weighted adjacency graph from connection list.
|
||||
|
||||
Weight = number of wires between each connector pair.
|
||||
"""
|
||||
adj: dict[str, dict[str, int]] = defaultdict(lambda: defaultdict(int))
|
||||
for conn in connections:
|
||||
left_name = next(iter(conn[0]))
|
||||
right_name = next(iter(conn[2]))
|
||||
pins = conn[0][left_name]
|
||||
wire_count = 1 if isinstance(pins, int) else len(pins)
|
||||
adj[left_name][right_name] += wire_count
|
||||
adj[right_name][left_name] += wire_count
|
||||
return adj
|
||||
|
||||
|
||||
def _compute_layer_ordering(
|
||||
names: list[str],
|
||||
adj: dict[str, dict[str, int]],
|
||||
) -> list[str]:
|
||||
"""Compute optimal left-to-right ordering using layered barycenter method.
|
||||
|
||||
External connectors (J*, TP*, P*) are placed in the leftmost layer,
|
||||
then modules are assigned to deeper layers by BFS distance.
|
||||
Within each layer, nodes are ordered by the barycenter of their
|
||||
connections to adjacent layers, minimizing edge crossings.
|
||||
"""
|
||||
if len(names) <= 1:
|
||||
return list(names)
|
||||
|
||||
name_set = set(names)
|
||||
|
||||
# --- Phase 1: Assign layers via BFS from external connectors ---
|
||||
external = sorted(n for n in names if not n.startswith("X"))
|
||||
|
||||
layers: dict[str, int] = {}
|
||||
|
||||
if external:
|
||||
queue = list(external)
|
||||
for n in queue:
|
||||
layers[n] = 0
|
||||
visited = set(queue)
|
||||
else:
|
||||
# No external connectors; start from lowest-degree node
|
||||
start = min(names, key=lambda n: sum(adj.get(n, {}).values()) if adj.get(n) else 0)
|
||||
queue = [start]
|
||||
layers[start] = 0
|
||||
visited = {start}
|
||||
|
||||
while queue:
|
||||
next_queue = []
|
||||
for node in queue:
|
||||
for neighbor in sorted(adj.get(node, {})):
|
||||
if neighbor in name_set and neighbor not in visited:
|
||||
visited.add(neighbor)
|
||||
layers[neighbor] = layers[node] + 1
|
||||
next_queue.append(neighbor)
|
||||
queue = next_queue
|
||||
|
||||
# Handle disconnected nodes
|
||||
max_layer = max(layers.values()) if layers else 0
|
||||
for n in names:
|
||||
if n not in layers:
|
||||
layers[n] = max_layer + 1
|
||||
|
||||
# --- Phase 2: Group by layer ---
|
||||
max_layer = max(layers.values())
|
||||
layer_groups: list[list[str]] = [[] for _ in range(max_layer + 1)]
|
||||
for n in names:
|
||||
layer_groups[layers[n]].append(n)
|
||||
|
||||
# Initial sort: alphabetical within each layer (stable base)
|
||||
for group in layer_groups:
|
||||
group.sort()
|
||||
|
||||
# --- Phase 3: Forward sweep — order layers 1+ by connections to previous ---
|
||||
for layer_idx in range(1, len(layer_groups)):
|
||||
prev_positions = {n: i for i, n in enumerate(layer_groups[layer_idx - 1])}
|
||||
|
||||
# Default param _pp captures THIS iteration's prev_positions, not the
|
||||
# final value after the loop completes (classic Python closure gotcha).
|
||||
def _bc_fwd(node: str, _pp: dict[str, int] = prev_positions) -> float:
|
||||
"""Weighted average position of neighbors in previous layer.
|
||||
Returns inf for nodes with no previous-layer neighbors (sorts last)."""
|
||||
neighbors = {n: w for n, w in adj.get(node, {}).items() if n in _pp}
|
||||
if not neighbors:
|
||||
return float("inf")
|
||||
total_w = sum(neighbors.values())
|
||||
return sum(_pp[n] * w for n, w in neighbors.items()) / total_w
|
||||
|
||||
layer_groups[layer_idx].sort(key=_bc_fwd)
|
||||
|
||||
# --- Phase 4: Backward sweep — refine earlier layers by next layer ---
|
||||
for layer_idx in range(len(layer_groups) - 2, -1, -1):
|
||||
next_positions = {n: i for i, n in enumerate(layer_groups[layer_idx + 1])}
|
||||
|
||||
# Same default-param capture pattern as _bc_fwd above.
|
||||
def _bc_bwd(node: str, _np: dict[str, int] = next_positions) -> float:
|
||||
"""Weighted average position of neighbors in next layer.
|
||||
Returns inf for nodes with no next-layer neighbors (sorts last)."""
|
||||
neighbors = {n: w for n, w in adj.get(node, {}).items() if n in _np}
|
||||
if not neighbors:
|
||||
return float("inf")
|
||||
total_w = sum(neighbors.values())
|
||||
return sum(_np[n] * w for n, w in neighbors.items()) / total_w
|
||||
|
||||
layer_groups[layer_idx].sort(key=_bc_bwd)
|
||||
|
||||
# Flatten layers into single linear order
|
||||
return [n for group in layer_groups for n in group]
|
||||
|
||||
|
||||
def _orient_connections(
|
||||
connections: list[list[dict[str, Any]]],
|
||||
positions: dict[str, int],
|
||||
) -> list[list[dict[str, Any]]]:
|
||||
"""Orient each connection so the earlier connector (in ordering) is on the left."""
|
||||
oriented = []
|
||||
for conn in connections:
|
||||
left_name = next(iter(conn[0]))
|
||||
right_name = next(iter(conn[2]))
|
||||
|
||||
left_pos = positions.get(left_name, 0)
|
||||
right_pos = positions.get(right_name, 0)
|
||||
|
||||
if left_pos > right_pos:
|
||||
# Swap: put the earlier connector on the left
|
||||
oriented.append([
|
||||
{right_name: conn[2][right_name]},
|
||||
conn[1], # cable is symmetric
|
||||
{left_name: conn[0][left_name]},
|
||||
])
|
||||
else:
|
||||
oriented.append(conn)
|
||||
|
||||
return oriented
|
||||
|
||||
|
||||
def _optimize_pin_order(
|
||||
connector_name: str,
|
||||
pinlabels: list[str],
|
||||
connections: list[list[dict[str, Any]]],
|
||||
positions: dict[str, int],
|
||||
) -> tuple[list[str], dict[int, int] | None]:
|
||||
"""Reorder pins so connections to the same neighbor are grouped,
|
||||
and groups are ordered by neighbor position (left neighbors at top,
|
||||
right neighbors at bottom).
|
||||
|
||||
Returns (new_pinlabels, old_to_new_mapping) or (pinlabels, None) if no change.
|
||||
"""
|
||||
my_pos = positions.get(connector_name, 0)
|
||||
|
||||
# Collect ALL neighbor (position, wire_count) per pin. A pin may appear in
|
||||
# multiple connections when it participates in a star topology (e.g. GND
|
||||
# shared across many modules). We weight by wire count so heavier
|
||||
# connections pull the pin's sort position more strongly.
|
||||
pin_neighbor_data: dict[int, list[tuple[float, int]]] = defaultdict(list)
|
||||
|
||||
for conn in connections:
|
||||
left_name = next(iter(conn[0]))
|
||||
right_name = next(iter(conn[2]))
|
||||
|
||||
if left_name == connector_name:
|
||||
neighbor_pos = positions.get(right_name, my_pos)
|
||||
pins = conn[0][connector_name]
|
||||
pin_list = pins if isinstance(pins, list) else [pins]
|
||||
wire_count = len(pin_list)
|
||||
for p in pin_list:
|
||||
pin_neighbor_data[p].append((neighbor_pos, wire_count))
|
||||
|
||||
elif right_name == connector_name:
|
||||
neighbor_pos = positions.get(left_name, my_pos)
|
||||
pins = conn[2][connector_name]
|
||||
pin_list = pins if isinstance(pins, list) else [pins]
|
||||
wire_count = len(pin_list)
|
||||
for p in pin_list:
|
||||
pin_neighbor_data[p].append((neighbor_pos, wire_count))
|
||||
|
||||
if not pin_neighbor_data:
|
||||
return pinlabels, None
|
||||
|
||||
# Weighted average neighbor position per pin
|
||||
pin_avg_pos: dict[int, float] = {}
|
||||
for p, data_list in pin_neighbor_data.items():
|
||||
total_w = sum(w for _, w in data_list)
|
||||
pin_avg_pos[p] = sum(pos * w for pos, w in data_list) / total_w
|
||||
|
||||
pin_indices = list(range(1, len(pinlabels) + 1))
|
||||
|
||||
# Sort by average neighbor position (ascending), then by original index as tiebreaker
|
||||
sorted_pins = sorted(
|
||||
pin_indices,
|
||||
key=lambda p: (pin_avg_pos.get(p, my_pos), p),
|
||||
)
|
||||
|
||||
if sorted_pins == pin_indices:
|
||||
return pinlabels, None
|
||||
|
||||
# Build mapping and new labels
|
||||
old_to_new: dict[int, int] = {}
|
||||
new_labels: list[str] = []
|
||||
for new_idx, old_idx in enumerate(sorted_pins, 1):
|
||||
old_to_new[old_idx] = new_idx
|
||||
new_labels.append(pinlabels[old_idx - 1])
|
||||
|
||||
return new_labels, old_to_new
|
||||
|
||||
|
||||
def _remap_pins(
|
||||
name: str, pins: int | list[int], mapping: dict[int, int]
|
||||
) -> int | list[int]:
|
||||
"""Apply pin index remapping, failing explicitly if a pin is missing."""
|
||||
try:
|
||||
if isinstance(pins, int):
|
||||
return mapping[pins]
|
||||
return [mapping[p] for p in pins]
|
||||
except KeyError as e:
|
||||
raise ValueError(
|
||||
f"Pin remapping failed for {name}: pin {e.args[0]} not in "
|
||||
f"connector pinlabels. Available: {sorted(mapping.keys())}, "
|
||||
f"Requested: {pins if isinstance(pins, int) else list(pins)}"
|
||||
) from e
|
||||
|
||||
|
||||
def _apply_pin_mappings(
|
||||
connections: list[list[dict[str, Any]]],
|
||||
pin_mappings: dict[str, dict[int, int]],
|
||||
) -> list[list[dict[str, Any]]]:
|
||||
"""Update connection pin indices after pin reordering."""
|
||||
if not pin_mappings:
|
||||
return connections
|
||||
|
||||
updated = []
|
||||
for conn in connections:
|
||||
new_conn = list(conn) # shallow copy
|
||||
|
||||
# Left side (index 0)
|
||||
left_name = next(iter(conn[0]))
|
||||
if left_name in pin_mappings:
|
||||
new_conn[0] = {
|
||||
left_name: _remap_pins(left_name, conn[0][left_name], pin_mappings[left_name])
|
||||
}
|
||||
|
||||
# Right side (index 2)
|
||||
right_name = next(iter(conn[2]))
|
||||
if right_name in pin_mappings:
|
||||
new_conn[2] = {
|
||||
right_name: _remap_pins(right_name, conn[2][right_name], pin_mappings[right_name])
|
||||
}
|
||||
|
||||
updated.append(new_conn)
|
||||
|
||||
return updated
|
||||
|
||||
|
||||
def _optimize_layout(
|
||||
connectors: dict[str, dict[str, Any]],
|
||||
cables: dict[str, dict[str, Any]],
|
||||
connections: list[list[dict[str, Any]]],
|
||||
) -> tuple[dict[str, dict[str, Any]], dict[str, dict[str, Any]], list[list[dict[str, Any]]]]:
|
||||
"""Optimize connector ordering and pin sequencing for cleaner diagrams.
|
||||
|
||||
Uses a Sugiyama-lite layered graph drawing approach to minimize
|
||||
cable crossings in the rendered WireViz diagram.
|
||||
"""
|
||||
if len(connectors) < 2 or not connections:
|
||||
return connectors, cables, connections
|
||||
|
||||
# Step 1: Build adjacency and compute ordering
|
||||
adj = _build_adjacency(connections)
|
||||
order = _compute_layer_ordering(list(connectors.keys()), adj)
|
||||
positions = {name: i for i, name in enumerate(order)}
|
||||
|
||||
# Step 2: Reorder connectors dict (insertion order drives GraphViz placement)
|
||||
ordered_connectors: dict[str, dict[str, Any]] = {}
|
||||
for name in order:
|
||||
if name in connectors:
|
||||
ordered_connectors[name] = connectors[name]
|
||||
# Append any disconnected connectors not in the order
|
||||
for name, conn in connectors.items():
|
||||
if name not in ordered_connectors:
|
||||
ordered_connectors[name] = conn
|
||||
|
||||
# Step 3: Orient connections (earlier connector on the left)
|
||||
oriented = _orient_connections(connections, positions)
|
||||
|
||||
# Step 4: Optimize pin ordering within each connector
|
||||
pin_mappings: dict[str, dict[int, int]] = {}
|
||||
for name in order:
|
||||
if name not in ordered_connectors:
|
||||
continue
|
||||
conn = ordered_connectors[name]
|
||||
pinlabels = conn.get("pinlabels")
|
||||
if not pinlabels or len(pinlabels) <= 1:
|
||||
continue
|
||||
new_labels, mapping = _optimize_pin_order(name, pinlabels, oriented, positions)
|
||||
if mapping:
|
||||
# Rebuild connector dict with new pinlabels (preserve key order)
|
||||
new_conn: dict[str, Any] = {}
|
||||
for k, v in conn.items():
|
||||
if k == "pinlabels":
|
||||
new_conn[k] = new_labels
|
||||
else:
|
||||
new_conn[k] = v
|
||||
ordered_connectors[name] = new_conn
|
||||
pin_mappings[name] = mapping
|
||||
|
||||
# Step 5: Apply pin index remapping to all connections
|
||||
if pin_mappings:
|
||||
oriented = _apply_pin_mappings(oriented, pin_mappings)
|
||||
|
||||
# Step 6: Sort connections by distance (shorter connections first gives
|
||||
# GraphViz better routing hints — adjacent pairs get priority).
|
||||
# Fully deterministic key: (distance, left_pos, right_pos).
|
||||
def _conn_distance(conn: list[dict[str, Any]]) -> tuple[int, int, int]:
|
||||
left = next(iter(conn[0]))
|
||||
right = next(iter(conn[2]))
|
||||
dist = abs(positions.get(right, 0) - positions.get(left, 0))
|
||||
return (dist, positions.get(left, 0), positions.get(right, 0))
|
||||
|
||||
oriented.sort(key=_conn_distance)
|
||||
|
||||
return ordered_connectors, cables, oriented
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Main mapper
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def map_inter_module(
|
||||
netlist: ParsedNetlist,
|
||||
config: FilterConfig | None = None,
|
||||
|
|
@ -199,4 +542,7 @@ def map_inter_module(
|
|||
)
|
||||
cable_counter += 1
|
||||
|
||||
# --- Optimize layout for cleaner diagrams ---
|
||||
connectors, cables, connections = _optimize_layout(connectors, cables, connections)
|
||||
|
||||
return assemble_wireviz_doc(connectors, cables, connections, metadata)
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue