Add star-topology layout optimization for single-module diagrams

Three-pass optimization eliminates cable crossings:
1. Order boundary components by average header pin position
2. Regroup header pins by boundary component (reduces inter-cable crossings)
3. Reorder boundary component pins to parallel header (eliminates within-cable
   crossings)

Safety hardening from Apollo review:
- Duplicate header pin deduplication (prevents silent mapping corruption)
- Connection structure validation at entry
- Fan-out averaging for component pins connected to multiple header pins
- Explicit ValueError on pin remapping failures with diagnostic context

145 tests passing (was 130).
This commit is contained in:
Ryan Malloy 2026-02-13 03:07:50 -07:00
parent b8ff2d19da
commit c2197f6fe6
3 changed files with 482 additions and 3 deletions

View file

@ -8,11 +8,18 @@ defined inside it) and port interface. Generates:
- Cables connecting the module header to boundary components via shared nets
This shows the external interface of a single board/module.
Layout optimization for the star topology:
1. Order boundary components by average header pin position
2. Group header pins by boundary component (reduces inter-cable crossings)
3. Reorder boundary component pins to be parallel with header (eliminates
within-cable crossings)
"""
from __future__ import annotations
import sys
from collections import defaultdict
from typing import Any
from ..emitter.yaml_emitter import (
@ -34,6 +41,257 @@ def _net_wire_color(net: str, netlist: ParsedNetlist) -> str:
return ""
# ---------------------------------------------------------------------------
# Layout optimization helpers
# ---------------------------------------------------------------------------
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"mapping. Available: {sorted(mapping.keys())}, "
f"Requested: {pins if isinstance(pins, int) else list(pins)}"
) from e
def _optimize_single_layout(
header_name: str,
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 star-topology layout for single-module diagrams.
Three-pass optimization:
1. Order boundary components by average header pin position
2. Regroup header pins so pins connecting to the same boundary
component are adjacent (reduces inter-cable crossings)
3. Reorder boundary component pins to be parallel with header
(eliminates within-cable crossings)
"""
if len(connectors) < 2 or not connections:
return connectors, cables, connections
header = connectors.get(header_name)
if not header or "pinlabels" not in header:
return connectors, cables, connections
header_labels = header["pinlabels"]
n_header = len(header_labels)
# --- Parse connections to extract (header_pin, comp_pin) pairs ---
comp_pin_pairs: dict[str, list[tuple[int, int]]] = defaultdict(list)
for i, conn in enumerate(connections):
if len(conn) < 3:
raise ValueError(
f"Connection {i} has {len(conn)} elements, expected at least 3 "
f"(connector, cable, connector): {conn}"
)
left_name = next(iter(conn[0]))
right_name = next(iter(conn[2]))
if left_name == header_name:
comp_name = right_name
h_pins = conn[0][header_name]
c_pins = conn[2][comp_name]
elif right_name == header_name:
comp_name = left_name
h_pins = conn[2][header_name]
c_pins = conn[0][comp_name]
else:
continue
h_list = [h_pins] if isinstance(h_pins, int) else list(h_pins)
c_list = [c_pins] if isinstance(c_pins, int) else list(c_pins)
for h, c in zip(h_list, c_list, strict=True):
comp_pin_pairs[comp_name].append((h, c))
if not comp_pin_pairs:
return connectors, cables, connections
# --- Step 1: Order boundary components by average header pin ---
comp_avg: dict[str, float] = {}
for comp_name, pairs in comp_pin_pairs.items():
comp_avg[comp_name] = sum(h for h, _ in pairs) / len(pairs)
boundary_order = sorted(comp_avg.keys(), key=lambda c: (comp_avg[c], c))
# --- Step 2: Regroup header pins by boundary component ---
# Pins connecting to the first boundary component come first, then
# the second, etc. Unconnected header pins go at the end.
# Duplicate detection: a header pin shared by multiple boundary
# components (e.g., GND routed to both J1 and J2) is assigned to
# the first component in boundary_order — subsequent duplicates
# are skipped to avoid corrupting the pin mapping.
new_header_order: list[int] = []
seen_pins: set[int] = set()
for comp_name in boundary_order:
comp_h_pins = sorted(h for h, _ in comp_pin_pairs[comp_name])
for pin in comp_h_pins:
if pin not in seen_pins:
seen_pins.add(pin)
new_header_order.append(pin)
connected = set(new_header_order)
for p in range(1, n_header + 1):
if p not in connected:
new_header_order.append(p)
# Build header pin mapping (old position -> new position)
header_mapping: dict[int, int] | None = None
if new_header_order != list(range(1, n_header + 1)):
header_mapping = {
old: new for new, old in enumerate(new_header_order, 1)
}
new_header_labels = [header_labels[old - 1] for old in new_header_order]
else:
new_header_labels = header_labels
# --- Step 3: Reorder boundary component pins to parallel header ---
# For each component, sort its pins by the corresponding (remapped)
# header pin position so wires run parallel without crossing.
comp_mappings: dict[str, dict[int, int]] = {}
for comp_name, pairs in comp_pin_pairs.items():
comp_conn = connectors.get(comp_name, {})
comp_labels = comp_conn.get("pinlabels", [])
n_comp = len(comp_labels) if comp_labels else comp_conn.get("pincount", 0)
if n_comp <= 1:
continue
# Map each component pin to its effective header position.
# A pin may connect to multiple header pins (fan-out); use
# the average position rather than last-wins.
comp_pin_h_positions: dict[int, list[float]] = defaultdict(list)
for h, c in pairs:
h_pos = header_mapping[h] if header_mapping else h
comp_pin_h_positions[c].append(float(h_pos))
comp_pin_to_h_pos: dict[int, float] = {
pin: sum(positions) / len(positions)
for pin, positions in comp_pin_h_positions.items()
}
# Sort: connected pins by header position, unconnected at end
all_comp_pins = list(range(1, n_comp + 1))
sorted_comp = sorted(
all_comp_pins,
key=lambda p: (comp_pin_to_h_pos.get(p, float("inf")), p),
)
if sorted_comp != all_comp_pins:
mapping = {old: new for new, old in enumerate(sorted_comp, 1)}
comp_mappings[comp_name] = mapping
# --- Build new ordered connectors dict (no in-place mutation) ---
ordered: dict[str, dict[str, Any]] = {}
# Header first (with possible pinlabel update)
if header_mapping:
new_h: dict[str, Any] = {}
for k, v in header.items():
new_h[k] = new_header_labels if k == "pinlabels" else v
ordered[header_name] = new_h
else:
ordered[header_name] = connectors[header_name]
# Boundary components in order (with possible pinlabel update)
for comp_name in boundary_order:
if comp_name not in connectors:
continue
comp_conn = connectors[comp_name]
if comp_name in comp_mappings:
mapping = comp_mappings[comp_name]
comp_labels = comp_conn.get("pinlabels", [])
if comp_labels:
sorted_pins = sorted(mapping.keys(), key=lambda p: mapping[p])
new_labels = [comp_labels[old - 1] for old in sorted_pins]
new_comp: dict[str, Any] = {}
for k, v in comp_conn.items():
new_comp[k] = new_labels if k == "pinlabels" else v
ordered[comp_name] = new_comp
else:
ordered[comp_name] = comp_conn
else:
ordered[comp_name] = comp_conn
# Remaining connectors (e.g., disconnected test points)
for name, conn in connectors.items():
if name not in ordered:
ordered[name] = conn
# --- Apply all pin mappings to connections ---
all_mappings: dict[str, dict[int, int]] = {}
if header_mapping:
all_mappings[header_name] = header_mapping
all_mappings.update(comp_mappings)
if all_mappings:
updated: list[list[dict[str, Any]]] = []
for conn in connections:
# Shallow copy: only conn[0] and conn[2] are replaced (new dicts).
# conn[1] (cable) is shared by reference — safe because cables
# are not remapped in this pass.
new_conn = list(conn)
left_name = next(iter(conn[0]))
if left_name in all_mappings:
new_conn[0] = {
left_name: _remap_pins(
left_name, conn[0][left_name], all_mappings[left_name]
)
}
right_name = next(iter(conn[2]))
if right_name in all_mappings:
new_conn[2] = {
right_name: _remap_pins(
right_name, conn[2][right_name], all_mappings[right_name]
)
}
updated.append(new_conn)
connections = updated
# --- Sort connections by minimum header pin (top-to-bottom order) ---
def _sort_key(conn: list[dict[str, Any]]) -> tuple[float, ...]:
left_name = next(iter(conn[0]))
right_name = next(iter(conn[2]))
if left_name == header_name:
h_pins = conn[0][header_name]
elif right_name == header_name:
h_pins = conn[2][header_name]
else:
return (float("inf"),)
pin_list = [h_pins] if isinstance(h_pins, int) else h_pins
return (min(pin_list),)
connections.sort(key=_sort_key)
return ordered, cables, connections
# ---------------------------------------------------------------------------
# Main mapper
# ---------------------------------------------------------------------------
def map_single_module(
netlist: ParsedNetlist,
subcircuit_name: str,
@ -143,6 +401,11 @@ def map_single_module(
comp_name, mapped_comp,
))
# --- Optimize layout for cleaner diagrams ---
connectors, cables, connections = _optimize_single_layout(
header_name, connectors, cables, connections
)
return assemble_wireviz_doc(connectors, cables, connections, metadata)