From 8e4e734ba43b9ddda7fc7984232421f55a4cfe4e Mon Sep 17 00:00:00 2001 From: subzero Date: Wed, 20 May 2026 19:09:54 +0200 Subject: [PATCH 1/3] =?UTF-8?q?framework:=20error=20message=20ergonomics?= =?UTF-8?q?=20=E2=80=94=20physical=20justifications=20+=20source-line=20at?= =?UTF-8?q?tribution=20(Phase=202b.1)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Every framework exception now teaches the user *why* the rule exists and *where* the offending wire() call lives, turning the diagnostic from "tells you what's wrong" into "teaches you why the rule exists." Two additions per exception class, both append-only: - Every WirebenchError subclass declares a PHYSICAL_JUSTIFICATION ClassVar — one or two sentences anchored to a physical-world referent. The justification renders as a "Why: …" line in str(exception). - Every wire() call captures its source location (skipping pydantic's validate_call wrappers via a filename-walk) and accumulates it on the resulting Node. Errors raised from inside wire() or Circuit._validate surface those locations as a "Wired at: …" line. The WireRecord format round-trips source_location through .wirebench so reconstructed designs still attribute back to user source rather than the loader. The shape: ShortCircuitError: wire() has multiple drivers ('y_1', 'y_2') — short circuit Why: Two OUT-direction ports on one net fight each other on the copper — current sinks through the losing output stage until the FETs overheat; one driver per shared conductor. Wired at: hello_led.py:14 Rules are unchanged; only the messages got richer. Existing pytest.raises(..., match='...') patterns continue to match the head of the original message; CLI JSON extractors operate on the first line only. Suite: 4726 passed (95 new tests), mypy clean. --- src/cli/validate_extractors.py | 32 +- src/framework/circuit.py | 35 ++- src/framework/errors.py | 273 ++++++++++++++++++ src/framework/format.py | 23 +- src/framework/format_records.py | 10 +- src/framework/node.py | 22 +- src/framework/wire.py | 68 ++++- tests/framework/test_dynamically_driven.py | 5 +- tests/framework/test_error_messages.py | 122 ++++++++ .../test_error_with_source_location.py | 140 +++++++++ .../test_source_location_roundtrip.py | 102 +++++++ tests/framework/test_wire_source_location.py | 74 +++++ 12 files changed, 879 insertions(+), 27 deletions(-) create mode 100644 tests/framework/test_error_messages.py create mode 100644 tests/framework/test_error_with_source_location.py create mode 100644 tests/framework/test_source_location_roundtrip.py create mode 100644 tests/framework/test_wire_source_location.py diff --git a/src/cli/validate_extractors.py b/src/cli/validate_extractors.py index a16b69a..a7ff382 100644 --- a/src/cli/validate_extractors.py +++ b/src/cli/validate_extractors.py @@ -42,6 +42,18 @@ def empty_details() -> Details: _Extractor = Callable[[str], Details] +def _head_line(message: str) -> str: + """Return the first line of an exception message. + + `WirebenchError.__str__` appends optional `Why:` and `Wired at:` + lines after the base message; the per-class patterns below match + against the base message only. Bullet-style payloads (used by + `BreadboardIncompatibleError`) handle their multi-line shape via + `splitlines()` directly. + """ + return message.splitlines()[0] if message else '' + + def _strip_quotes(token: str) -> str: token = token.strip() if len(token) >= 2 and token[0] == token[-1] and token[0] in ("'", '"'): @@ -109,11 +121,12 @@ def _split_part_pin_list(payload: str) -> tuple[list[str], list[str]]: def extract_short_circuit(message: str) -> Details: out = empty_details() - m = _SHORT_WIRE.match(message) + head = _head_line(message) + m = _SHORT_WIRE.match(head) if m: out['pins'] = _split_quoted_list(m.group(1)) return out - m = _SHORT_NET.match(message) + m = _SHORT_NET.match(head) if m: parts, pins = _split_part_pin_list(m.group(1)) out['parts'] = parts @@ -123,7 +136,7 @@ def extract_short_circuit(message: str) -> Details: def extract_floating_net(message: str) -> Details: out = empty_details() - m = _FLOATING_NET.match(message) + m = _FLOATING_NET.match(_head_line(message)) if m: parts, pins = _split_part_pin_list(m.group(1)) out['parts'] = parts @@ -133,7 +146,7 @@ def extract_floating_net(message: str) -> Details: def extract_incompatible_mate(message: str) -> Details: out = empty_details() - m = _INCOMPATIBLE_MATE.match(message) + m = _INCOMPATIBLE_MATE.match(_head_line(message)) if m: out['parts'] = [m.group(1), m.group(3)] return out @@ -141,18 +154,19 @@ def extract_incompatible_mate(message: str) -> Details: def extract_part_configuration(message: str) -> Details: out = empty_details() - m = _PARTCONFIG_OUT_NO_CELL.match(message) + head = _head_line(message) + m = _PARTCONFIG_OUT_NO_CELL.match(head) if m: out['parts'] = [m.group(1)] out['pin'] = m.group(2) out['pin_number'] = int(m.group(3)) return out - m = _PARTCONFIG_DRIVE_WRONG_DIR.match(message) + m = _PARTCONFIG_DRIVE_WRONG_DIR.match(head) if m: out['parts'] = [m.group(1)] out['pin'] = m.group(2) return out - m = _PARTCONFIG_TYPO_ENTRY.match(message) + m = _PARTCONFIG_TYPO_ENTRY.match(head) if m: out['parts'] = [m.group(1)] out['pin'] = m.group(2) @@ -161,7 +175,7 @@ def extract_part_configuration(message: str) -> Details: def extract_part_parameter(message: str) -> Details: out = empty_details() - m = _PARTPARAM_PIN_FUNCTION.match(message) + m = _PARTPARAM_PIN_FUNCTION.match(_head_line(message)) if m: out['pin_number'] = int(m.group(1)) out['pin'] = m.group(2) @@ -170,7 +184,7 @@ def extract_part_parameter(message: str) -> Details: def extract_domain_crossing(message: str) -> Details: out = empty_details() - m = _DOMAIN_CROSSING.match(message) + m = _DOMAIN_CROSSING.match(_head_line(message)) if not m: return out pins: list[str] = [] diff --git a/src/framework/circuit.py b/src/framework/circuit.py index d7917f2..3478e6f 100644 --- a/src/framework/circuit.py +++ b/src/framework/circuit.py @@ -87,6 +87,24 @@ def _empty_circuit_message(self) -> str: f"`parts=[]` explicitly." ) + @staticmethod + def _collect_net_source_locations( + net_ports: Sequence[tuple[Part, Port]], + ) -> list[tuple[str, int]]: + """Deduped source locations of every `wire()` call that touched + any port on this net, preserving first-appearance order.""" + collected: list[tuple[str, int]] = [] + seen: set[tuple[str, int]] = set() + for _, port in net_ports: + if port.node is None: + continue + for loc in port.node.source_locations: + if loc in seen: + continue + seen.add(loc) + collected.append(loc) + return collected + def _validate_no_orphan_ports(self, parts: list[Part]) -> None: """Rule 2: every port wired to a port we know about must itself belong to a part we know about. @@ -129,7 +147,8 @@ def _validate_no_orphan_ports(self, parts: list[Part]) -> None: f"the orphan as `self. = …` so the " f"framework auto-collects it, or pass " f"`parts=[…]` explicitly listing every " - f"part." + f"part.", + source_locations=port.node.source_locations, ) def _auto_collect_parts(self) -> list[Part]: @@ -202,7 +221,9 @@ def _validate(self, parts: list[Part]) -> None: # boundaries. from framework.export.nets import compute_logical_nets shorts: list[str] = [] + short_locations: list[tuple[str, int]] = [] floats: list[str] = [] + float_locations: list[tuple[str, int]] = [] for net in compute_logical_nets(self): outs = [(o, p) for (o, p) in net.ports if p.direction is Direction.OUT] bidirs = [(o, p) for (o, p) in net.ports if p.direction is Direction.BIDIR] @@ -210,6 +231,9 @@ def _validate(self, parts: list[Part]) -> None: if len(outs) > 1: shorts.append(', '.join( f"'{type(o).__name__}.{p.name}'" for o, p in outs)) + for loc in self._collect_net_source_locations(net.ports): + if loc not in short_locations: + short_locations.append(loc) elif (len(outs) == 0 and len(bidirs) > 1 and not net.dynamically_driven): # `dynamically_driven` is the designer's explicit @@ -219,16 +243,21 @@ def _validate(self, parts: list[Part]) -> None: # detection above stays strict regardless. floats.append(', '.join( f"'{type(o).__name__}.{p.name}'" for o, p in bidirs)) + for loc in self._collect_net_source_locations(net.ports): + if loc not in float_locations: + float_locations.append(loc) if shorts: raise ShortCircuitError( "Short circuit on logical net — multiple drivers: " - + '; '.join(shorts) + + '; '.join(shorts), + source_locations=short_locations, ) if floats: raise FloatingNetError( "Floating logical net — multiple passive BIDIRs with no driver: " - + '; '.join(floats) + + '; '.join(floats), + source_locations=float_locations, ) # Duplicate-refdes detection. Walks only refdes-bearing children diff --git a/src/framework/errors.py b/src/framework/errors.py index 4e3c46b..8dfe1cc 100644 --- a/src/framework/errors.py +++ b/src/framework/errors.py @@ -16,15 +16,67 @@ so existing `pytest.raises(ValueError, …)` checks keep working without modification. The family classes themselves inherit from `Exception` only — they're abstract groupings. + +Every class carries a `PHYSICAL_JUSTIFICATION` ClassVar — one or two +short sentences anchored to a physical-world referent that explain +*why this rule exists*. The justification renders as a `Why:` line +after the base message whenever the exception is stringified, turning +the diagnostic from *"tells you what's wrong"* into *"teaches you why +the rule exists."* When the framework knows which `wire()` call(s) +introduced the defect, those source locations render as an additional +`Wired at:` line. Both additions are append-only — existing +`pytest.raises(..., match='...')` patterns continue to find the head of +the original message. """ from __future__ import annotations +from collections.abc import Sequence +from typing import Any, ClassVar + # ----------------------------------------------------------------- root class WirebenchError(Exception): """Root of every framework-raised exception.""" + PHYSICAL_JUSTIFICATION: ClassVar[str] = ( + "Real breadboards punish topology mistakes at runtime — overheated " + "outputs, undefined logic levels, parts that won't seat. " + "wirebench catches them at construction time instead." + ) + + def __init__( + self, + *args: Any, + source_locations: Sequence[tuple[str, int]] | None = None, + ) -> None: + super().__init__(*args) + self._source_locations: tuple[tuple[str, int], ...] = ( + tuple(source_locations) if source_locations else () + ) + + @property + def source_locations(self) -> tuple[tuple[str, int], ...]: + """File-and-line of each `wire()` call that contributed to the + defect, in attachment order. Empty when the framework couldn't + attribute the error to a specific call site (e.g. errors raised + outside the `wire()` path).""" + return self._source_locations + + def __str__(self) -> str: + base = super().__str__() + parts = [base] + justification = type(self).PHYSICAL_JUSTIFICATION + if justification: + parts.append(f" Why: {justification}") + if self._source_locations: + locs = ", ".join( + f"{filename}:{lineno}" + for filename, lineno in self._source_locations + ) + parts.append(f" Wired at: {locs}") + return "\n".join(parts) + # ============================================================ Circuit === # Something is wrong with the circuit you described. @@ -33,6 +85,12 @@ class CircuitError(WirebenchError): """Anything wrong with the circuit being modelled — its topology, its signals, its parts, its connectors, or its allowed states.""" + PHYSICAL_JUSTIFICATION: ClassVar[str] = ( + "The circuit as described has a real-world defect: wires routed " + "wrong, signals that can't share copper, parts in impossible " + "configurations, or connectors that don't physically fit." + ) + # ----------------------------------------------------------- Wiring --- @@ -41,31 +99,65 @@ class WiringError(CircuitError): wrong — too many drivers, no drivers, the same node touched twice, a mandatory pin left floating.""" + PHYSICAL_JUSTIFICATION: ClassVar[str] = ( + "Copper traces only carry one signal at one potential at a time; " + "wiring defects break that physical assumption." + ) + class ShortCircuitError(WiringError, ValueError): """Multiple drivers on a single logical net — two outputs tied together, or an output and a rail tied to the same node.""" + PHYSICAL_JUSTIFICATION: ClassVar[str] = ( + "Two OUT-direction ports on one net fight each other on the " + "copper — current sinks through the losing output stage until " + "the FETs overheat; one driver per shared conductor." + ) + class FloatingNetError(WiringError, ValueError): """A net touched only by passive BIDIR ports with no driver anywhere. Nothing can determine the net's value.""" + PHYSICAL_JUSTIFICATION: ClassVar[str] = ( + "A net with no driver has no defined voltage — CMOS inputs in " + "particular drift to mid-rail and oscillate, picking up nearby " + "switching noise as if the trace were an antenna." + ) + class UnconnectedPinError(WiringError, ValueError): """A pin declared `mandatory=True` was not connected to any wire inside the enclosing circuit.""" + PHYSICAL_JUSTIFICATION: ClassVar[str] = ( + "A mandatory pin (regulator IN, op-amp V+, MCU VDD) left in air " + "leaves the silicon stage tied to nothing — the part either " + "doesn't power up or behaves unpredictably." + ) + class NodeMergeError(WiringError, ValueError): """`wire()` was asked to join ports already on two distinct existing nodes — the framework refuses to merge them.""" + PHYSICAL_JUSTIFICATION: ClassVar[str] = ( + "Two distinct existing nets fused at construction time would " + "silently combine independent design intent; the framework " + "requires the join to be made explicit." + ) + class EmptyWireError(WiringError, ValueError): """`wire()` was called with no ports (or with a single port — which couldn't connect anything to anything).""" + PHYSICAL_JUSTIFICATION: ClassVar[str] = ( + "A wire with no endpoints connects nothing to nothing — it has " + "no physical analogue on a real breadboard." + ) + # ----------------------------------------------------------- Signal --- @@ -73,22 +165,46 @@ class SignalError(CircuitError): """Signal-type / direction / ground-domain mismatch — the wires are joined, but the signals they carry don't fit together.""" + PHYSICAL_JUSTIFICATION: ClassVar[str] = ( + "A piece of copper carries one kind of signal at one reference; " + "mismatched signal-types or ground domains can't coexist on the " + "same conductor without an isolator or level-shifter." + ) + class SignalTypeMismatchError(SignalError, TypeError): """An Analog signal landed on a Digital port, or vice versa.""" + PHYSICAL_JUSTIFICATION: ClassVar[str] = ( + "An Analog continuous voltage and a Digital logic-level on one " + "net are incompatible — converting between them needs an " + "explicit interface (comparator, ADC, level-shifter)." + ) + class DomainCrossingError(SignalError, ValueError): """A `wire()` or `Port↔Node` attachment crosses a `GroundDomain` boundary. Only `IsolatedChannel`-style cells with ports in different domains may bridge.""" + PHYSICAL_JUSTIFICATION: ClassVar[str] = ( + "Two ground domains share a net only through an isolator " + "(optocoupler, digital isolator, transformer); a direct wire " + "would tie the references together and defeat the isolation." + ) + class PortContentionError(SignalError, ValueError): """A BIDIR pin's external and internal faces hold conflicting values during evaluation — two simultaneous drivers on the bond wire, one from outside the chip and one from inside.""" + PHYSICAL_JUSTIFICATION: ClassVar[str] = ( + "A BIDIR pin is one piece of metal at one potential at a time — " + "the external and internal faces can't simultaneously drive " + "opposite values." + ) + # -------------------------------------------------- ForbiddenState --- @@ -100,6 +216,12 @@ class ForbiddenStateError(CircuitError, ValueError): *wirings* whose evaluation produces an undefined or destructive real-world result.""" + PHYSICAL_JUSTIFICATION: ClassVar[str] = ( + "Some logical states real silicon refuses to enter — SR latch " + "with S=R=1, half-bridge shoot-through, three-phase Hall " + "(1,1,1); evaluation produces undefined or destructive output." + ) + # ------------------------------------------------------------ Part --- @@ -107,35 +229,71 @@ class PartError(CircuitError): """Anything wrong with a part itself — its refdes, its parameters, its registration, its configuration.""" + PHYSICAL_JUSTIFICATION: ClassVar[str] = ( + "A part on a real board has fixed properties — label, pinout, " + "rated parameter ranges; a wirebench Part with broken " + "properties describes something the parts catalogue can't sell." + ) + class RefdesError(PartError, ValueError): """Reference-designator format violation: unknown prefix, non-positive number, duplicate refdes in a circuit, duplicate surface-port name.""" + PHYSICAL_JUSTIFICATION: ClassVar[str] = ( + "Every part on a real board needs one unique label so the " + "schematic, the layout, the BOM, and the assembly drawing all " + "refer to the same physical chip." + ) + class DuplicateRegistrationError(PartError, ValueError): """The component registry rejected a `@register(name)` because another class is already registered under that name.""" + PHYSICAL_JUSTIFICATION: ClassVar[str] = ( + "Two classes claiming the same registered name would make " + "`.wirebench` load ambiguous — the loader couldn't know which " + "class to instantiate." + ) + class UnknownPartError(PartError, KeyError): """The component registry was asked for a class name that hasn't been registered — typically when a saved `.wirebench` file references a class whose module hasn't been imported.""" + PHYSICAL_JUSTIFICATION: ClassVar[str] = ( + "The registry only knows classes whose module has been " + "imported; a name with no entry refers to no part the " + "framework can build." + ) + class PartConfigurationError(PartError, TypeError): """A component's constructor was passed bad arguments or the class is missing required class-level configuration (PIN_COUNT, PITCH_MM, PINOUT, distinct ground domains, etc.).""" + PHYSICAL_JUSTIFICATION: ClassVar[str] = ( + "A component's class-level configuration (pin count, pinout, " + "ground domains) is what makes it that specific part; missing " + "or inconsistent values describe nothing real." + ) + class PartParameterError(PartError, ValueError): """A component's constructor accepted the kwarg by signature but rejected the *value* — out-of-range state of charge, trip-high below trip-low, duplicate diode-OR input names, etc.""" + PHYSICAL_JUSTIFICATION: ClassVar[str] = ( + "Physical parameters (state of charge, trip levels, voltage " + "ratings) live within ranges set by the underlying physics; " + "values outside the range describe a part that can't exist." + ) + # ---------------------------------------------------------- Mating --- @@ -144,12 +302,25 @@ class MatingError(CircuitError): requested — wrong family, wrong dimensions, no in-model partner declared.""" + PHYSICAL_JUSTIFICATION: ClassVar[str] = ( + "Two connectors mate only when their shells, pin counts, and " + "pitches match the same mechanical drawing; mismatches don't " + "physically seat together." + ) + class IncompatibleMateError(MatingError, TypeError): """The two connectors are from fundamentally different families (e.g. a USB-A receptacle and a TRRS audio jack). The partner class doesn't match what `MATES_WITH` declared.""" + PHYSICAL_JUSTIFICATION: ClassVar[str] = ( + "Two connector families don't fit together physically — USB-A " + "and TRRS audio have different shells, pin counts, and " + "pitches; calling them mated is asserting a fact contradicted " + "by the mechanical drawings." + ) + class UnmateableError(MatingError, TypeError): """The connector has no in-model partner type declared @@ -157,16 +328,32 @@ class UnmateableError(MatingError, TypeError): mating partner lives outside the design (a barrel jack, an audio plug, a USB cable end).""" + PHYSICAL_JUSTIFICATION: ClassVar[str] = ( + "Some user-facing receptacles (barrel jacks, audio sockets) " + "mate with cables that live outside the design — there is no " + "in-model partner to wire to." + ) + class PinCountMismatchError(MatingError, ValueError): """Two connectors of compatible families have different pin counts — a 4-pin plug into a 5-pin receptacle.""" + PHYSICAL_JUSTIFICATION: ClassVar[str] = ( + "A 4-pin plug into a 5-pin receptacle leaves one pin unmated; " + "the physical connection is incomplete." + ) + class PitchMismatchError(MatingError, ValueError): """Two connectors of compatible families have different pitches — a 2.54 mm plug into a 2.00 mm receptacle.""" + PHYSICAL_JUSTIFICATION: ClassVar[str] = ( + "A 2.54 mm plug into a 2.00 mm receptacle won't seat — the " + "pins land between contacts, not on them." + ) + # ============================================================= Format === # Something is wrong with saving / loading / rendering a circuit. @@ -175,18 +362,36 @@ class FormatError(WirebenchError): """Anything wrong with the file format / persistence / rendering layer. The circuit itself may be fine; the I/O around it isn't.""" + PHYSICAL_JUSTIFICATION: ClassVar[str] = ( + "The serialised form of a design is the contract every " + "downstream tool (SPICE, KiCad, schematic visualisers) reads; " + "a broken format leaves no downstream tool able to consume it." + ) + class SaveError(FormatError, ValueError): """`save_wirebench` couldn't serialise the design — a component has no refdes and no synthesised id, a kwarg's type isn't encodable, etc.""" + PHYSICAL_JUSTIFICATION: ClassVar[str] = ( + "A design's serialised form must round-trip back to an " + "equivalent in-memory model; a save that loses information " + "leaves the file unable to reconstruct what it describes." + ) + class LoadError(FormatError, ValueError): """`load_wirebench` couldn't reconstruct a design from its serialised form — malformed port reference, unknown component id, duplicate id, unknown class, missing required field.""" + PHYSICAL_JUSTIFICATION: ClassVar[str] = ( + "A `.wirebench` file is a contract about which parts and wires " + "the design contains; a file that can't be parsed into that " + "contract isn't a wirebench design." + ) + class UnknownFormatError(FormatError, ValueError): """`export()` / `export_to_string()` was asked for a format name @@ -195,6 +400,12 @@ class UnknownFormatError(FormatError, ValueError): or a name the caller carried over from before the adapter was renamed.""" + PHYSICAL_JUSTIFICATION: ClassVar[str] = ( + "Each export adapter (KiCad, SPICE, SVG) speaks a specific " + "downstream tool's dialect; an unknown format name doesn't " + "correspond to any real tool." + ) + class BreadboardIncompatibleError(FormatError, ValueError): """The `assembly_guide` export was asked to render a design that @@ -207,6 +418,12 @@ class BreadboardIncompatibleError(FormatError, ValueError): DIP-28 instead of an ATmega2560 TQFP-100; a sensor breakout module instead of a bare BGA chip).""" + PHYSICAL_JUSTIFICATION: ClassVar[str] = ( + "Solderless breadboards accept only through-hole leads on " + "0.1\" centres; SMD chips and BGA packages need a PCB to " + "mount on." + ) + class RendererRegistryError(FormatError): """Anything wrong with the export-adapter renderer / net-namer @@ -214,18 +431,36 @@ class RendererRegistryError(FormatError): based on whether the operation was a duplicate registration (ValueError) or a missing lookup (KeyError).""" + PHYSICAL_JUSTIFICATION: ClassVar[str] = ( + "The export pipeline routes each component to a registered " + "renderer; a broken registry leaves the pipeline with no way " + "to translate the design into the downstream format." + ) + class DuplicateRendererError(RendererRegistryError, ValueError): """A renderer or net-namer is already registered for the requested (class, format) pair — the export adapter tried to register a second one.""" + PHYSICAL_JUSTIFICATION: ClassVar[str] = ( + "Each (component class, export format) pair has one canonical " + "renderer; two registered renderers would make the export " + "order-dependent and non-deterministic." + ) + class RendererNotFoundError(RendererRegistryError, KeyError): """No renderer (or net-namer) is registered for the requested (class, format) pair — typically because the adapter module wasn't imported before `export_to_string` was called.""" + PHYSICAL_JUSTIFICATION: ClassVar[str] = ( + "The export adapter registers renderers when its module is " + "imported; a missing entry means the module wasn't loaded " + "before export was attempted." + ) + # ============================================================== Usage === # The framework's API was called incorrectly. The circuit is fine; @@ -234,12 +469,25 @@ class RendererNotFoundError(RendererRegistryError, KeyError): class UsageError(WirebenchError): """The framework's API was called incorrectly.""" + PHYSICAL_JUSTIFICATION: ClassVar[str] = ( + "The framework's API is the boundary between the designer's " + "intent and the model; calling it incorrectly leaves the " + "model unable to represent what the designer meant." + ) + class WiredChipCallError(UsageError, RuntimeError): """A chip's standalone `__call__` was invoked while one or more of its ports were wired by an enclosing circuit. Drive via the parent's `evaluate()` instead.""" + PHYSICAL_JUSTIFICATION: ClassVar[str] = ( + "A chip wired into an enclosing circuit must be driven by " + "that circuit's `evaluate()`; calling its standalone " + "`__call__` bypasses the wired ports and produces " + "inconsistent output." + ) + class AmbiguousPinNameError(UsageError, KeyError): """`chip.ports['NAME']` was indexed by a name that resolves to @@ -247,6 +495,13 @@ class AmbiguousPinNameError(UsageError, KeyError): similar). Use the disambiguated name (`'VSS_1'`, `'VSS_2'`) or look up by pin number.""" + PHYSICAL_JUSTIFICATION: ClassVar[str] = ( + "Some chips have multiple pins sharing a common name (two VSS " + "pads, paired power inputs); a bare lookup can't pick one — " + "the disambiguated name (`VSS_1`, `VSS_2`) identifies the " + "specific solder pad." + ) + class CompositeShapeError(UsageError, TypeError): """A composite `Circuit` subclass is missing the `__dict__` the @@ -254,12 +509,24 @@ class CompositeShapeError(UsageError, TypeError): `__slots__`), or it constructed itself with an empty part list.""" + PHYSICAL_JUSTIFICATION: ClassVar[str] = ( + "A Circuit composite needs parts and a way to find them — " + "either auto-collection through `__dict__` or an explicit " + "`parts=` list; an empty composite describes nothing real." + ) + class UnknownPortError(UsageError, ValueError): """An internal API was given a port that doesn't belong to the object it was passed to — e.g. `pin.other_face(stranger_port)` where `stranger_port` is not one of this pin's two faces.""" + PHYSICAL_JUSTIFICATION: ClassVar[str] = ( + "An internal API was handed a port that doesn't belong to the " + "object it was passed to — the port reference is dangling and " + "names nothing on the recipient." + ) + class OrphanWireError(UsageError, ValueError): """A wire crosses the boundary of the enclosing circuit — joining @@ -269,3 +536,9 @@ class OrphanWireError(UsageError, ValueError): explicit `parts=[...]` list that omitted it. The framework can validate or evaluate the orphan's contribution to the circuit, so it refuses to build the composite.""" + + PHYSICAL_JUSTIFICATION: ClassVar[str] = ( + "A wire whose endpoint isn't in the composite's `parts` list " + "joins a phantom — a component the framework can't validate, " + "evaluate, or serialise." + ) diff --git a/src/framework/format.py b/src/framework/format.py index c713428..f377081 100644 --- a/src/framework/format.py +++ b/src/framework/format.py @@ -206,11 +206,16 @@ def _collect_wires( continue # Sort port refs lexicographically for determinism. sorted_refs = sorted(_port_ref(c, p) for c, p in refs) + node = nodes_by_id[nid] + # A Node touched by multiple wire() calls accumulates a + # list of source locations; round-trip preserves the first + # — sufficient for the loader to attribute the reconstructed + # WireRecord to the user's original call site. + node_locs = getattr(node, 'source_locations', ()) wires.append(WireRecord( ports=sorted_refs, - dynamically_driven=getattr( - nodes_by_id[nid], 'dynamically_driven', False, - ), + dynamically_driven=getattr(node, 'dynamically_driven', False), + source_location=node_locs[0] if node_locs else None, )) # Sort wires by their first port-ref for determinism. return sorted(wires, key=lambda w: w.ports[0]) @@ -516,7 +521,11 @@ def _from_circuit_record(record: CircuitRecord) -> Circuit: components, by_id = _rebuild_circuit_components(record.components) for w in record.wires: ports = [_resolve_port(r, by_id) for r in w.ports] - wire(*ports, dynamically_driven=w.dynamically_driven) + wire( + *ports, + dynamically_driven=w.dynamically_driven, + source_location=w.source_location, + ) surface_ports = { name: _resolve_port(ref, by_id) for name, ref in record.surface_ports.items() @@ -528,7 +537,11 @@ def _from_board_record(record: BoardRecord) -> Board: components, by_id = _rebuild_circuit_components(record.components) for w in record.wires: ports = [_resolve_port(r, by_id) for r in w.ports] - wire(*ports, dynamically_driven=w.dynamically_driven) + wire( + *ports, + dynamically_driven=w.dynamically_driven, + source_location=w.source_location, + ) refdes_number = _refdes_number_from_refdes(record.refdes) return Board( name=record.name, diff --git a/src/framework/format_records.py b/src/framework/format_records.py index 30a1a8d..3deb070 100644 --- a/src/framework/format_records.py +++ b/src/framework/format_records.py @@ -496,12 +496,20 @@ class WireRecord(_Record): # net only. Omitted from emitted JSON when False to keep the # format compact for the common case. dynamically_driven: bool = False + # `(filename, lineno)` of the user's original `wire()` call that + # introduced this net, preserved across save / load so that + # framework errors raised against the reconstructed circuit point + # back at the user-source line rather than the loader. Omitted + # from emitted JSON when absent. + source_location: tuple[str, int] | None = None @model_serializer(mode='wrap') - def _omit_default_dyn(self, handler: Any) -> dict[str, Any]: + def _omit_defaults(self, handler: Any) -> dict[str, Any]: data: dict[str, Any] = handler(self) if not self.dynamically_driven: data.pop('dynamically_driven', None) + if self.source_location is None: + data.pop('source_location', None) return data diff --git a/src/framework/node.py b/src/framework/node.py index b396d9b..1aaa57e 100644 --- a/src/framework/node.py +++ b/src/framework/node.py @@ -11,7 +11,10 @@ class Node: """A Kirchhoff junction node. Carries a potential value within one ground domain.""" - __slots__ = ('name', 'domain', '_value', '_ports', 'dynamically_driven') + __slots__ = ( + 'name', 'domain', '_value', '_ports', + 'dynamically_driven', '_source_locations', + ) def __init__(self, name: str, domain: GroundDomain) -> None: self.name = name @@ -28,6 +31,11 @@ def __init__(self, name: str, domain: GroundDomain) -> None: # multi-BIDIR-no-driver rule should not apply. Promoted to # True by `wire(..., dynamically_driven=True)`; never cleared. self.dynamically_driven: bool = False + # (filename, lineno) of every `wire()` call that touched this + # node, in call order. Populated by `wire()` via + # `_add_source_location`; surfaced in error messages that + # implicate the node and in `.wirebench` save / load. + self._source_locations: list[tuple[str, int]] = [] @property def value(self) -> Any: @@ -47,5 +55,17 @@ def ports(self) -> tuple[Port, ...]: """All ports attached to this node, in attachment order.""" return tuple(self._ports) + @property + def source_locations(self) -> tuple[tuple[str, int], ...]: + """`(filename, lineno)` of every `wire()` call that touched this + node, in call order. Empty when the node was created outside + the `wire()` path (e.g. by direct framework code).""" + return tuple(self._source_locations) + + def _add_source_location(self, location: tuple[str, int]) -> None: + """Called by `wire()` when a call touches this node. Underscore + prefix because external callers should never invoke this.""" + self._source_locations.append(location) + def __repr__(self) -> str: return f"Node('{self.name}', domain='{self.domain.name}', value={self._value!r})" diff --git a/src/framework/wire.py b/src/framework/wire.py index 859266d..e2e299a 100644 --- a/src/framework/wire.py +++ b/src/framework/wire.py @@ -1,4 +1,6 @@ +import sys import warnings +from types import FrameType from pydantic import validate_call @@ -10,8 +12,30 @@ from framework.port import Port, Direction +def _capture_user_call_site() -> tuple[str, int] | None: + """Walk up the call stack from inside `wire()` to find the first + frame that isn't inside pydantic's `validate_call` wrapper. + + Returns `(filename, lineno)` of the user's call site, or `None` if + no caller frame is available. Pydantic frames are identified by + 'pydantic' appearing in the filename — robust across pydantic + versions, where the number of wrapper frames varies. + """ + frame: 'FrameType | None' = sys._getframe(1) # skip wire() itself + while frame is not None: + filename = frame.f_code.co_filename + if 'pydantic' not in filename and filename != __file__: + return (filename, frame.f_lineno) + frame = frame.f_back + return None + + @validate_call(config={"arbitrary_types_allowed": True}) -def wire(*ports: Port, dynamically_driven: bool = False) -> None: +def wire( + *ports: Port, + dynamically_driven: bool = False, + source_location: tuple[str, int] | None = None, +) -> None: """Connect ports together, creating the junction node at their meeting point. Enforces: @@ -26,27 +50,45 @@ def wire(*ports: Port, dynamically_driven: bool = False) -> None: `Circuit._validate`'s multi-BIDIR-no-driver check on this net only. Promotion is one-way: once a node is annotated, subsequent wire() calls that omit the kwarg do not clear it. + + `source_location` is the `(filename, lineno)` to attribute this + `wire()` call to. Captured automatically from the caller's frame + when omitted; the `.wirebench` loader overrides with the + serialised location so reconstructed nodes carry the original + user-source attribution. """ + if source_location is None: + source_location = _capture_user_call_site() + call_locs: list[tuple[str, int]] = ( + [source_location] if source_location is not None else [] + ) + if not ports: - raise EmptyWireError("wire() requires at least one port") + raise EmptyWireError( + "wire() requires at least one port", + source_locations=call_locs, + ) domains = {p.domain for p in ports} if len(domains) > 1: names = ', '.join(f"'{p.name}' ({p.domain.name})" for p in ports) raise DomainCrossingError( - f"Cannot wire ports across ground domains: {names}" + f"Cannot wire ports across ground domains: {names}", + source_locations=call_locs, ) out_ports = [p for p in ports if p.direction is Direction.OUT] bidir_ports = [p for p in ports if p.direction is Direction.BIDIR] if len(out_ports) == 0 and len(bidir_ports) == 0: raise FloatingNetError( - "wire() has no driver: all ports are IN — nothing drives the node" + "wire() has no driver: all ports are IN — nothing drives the node", + source_locations=call_locs, ) if len(out_ports) > 1: names = ', '.join(f"'{p.name}'" for p in out_ports) raise ShortCircuitError( - f"wire() has multiple drivers ({names}) — short circuit" + f"wire() has multiple drivers ({names}) — short circuit", + source_locations=call_locs, ) from framework.signals import Analog, Digital @@ -81,7 +123,8 @@ def _base(t: type) -> type: if len(specific) > 1 or (Digital in specific and len(base_types) > 1): details = ', '.join(f"'{p.name}': {p.signal_type.__name__}" for p in ports) raise SignalTypeMismatchError( - f"Signal type mismatch in wire(): {details}" + f"Signal type mismatch in wire(): {details}", + source_locations=call_locs, ) # If any port is already on a node, extend that node (Kirchhoff junction). @@ -92,8 +135,17 @@ def _base(t: type) -> type: existing = {id(p.node): p.node for p in ports if p.node is not None} if len(existing) > 1: names = ', '.join(f"'{p.name}' on {p.node.name}" for p in ports if p.node is not None) + # Both pre-existing nodes' source-locations help the designer + # locate the two wire() calls whose nets the merge would join, + # plus this call site itself. + merge_locs: list[tuple[str, int]] = list(call_locs) + for node in existing.values(): + for loc in node.source_locations: + if loc not in merge_locs: + merge_locs.append(loc) raise NodeMergeError( - f"wire() would merge two existing nodes — not supported: {names}" + f"wire() would merge two existing nodes — not supported: {names}", + source_locations=merge_locs, ) domain = next(iter(domains)) @@ -102,6 +154,8 @@ def _base(t: type) -> type: else: name = '—'.join(p.name for p in ports) node = Node(name, domain) + if source_location is not None: + node._add_source_location(source_location) for p in ports: if p.node is None: p.connect(node) # connect() registers the port with the node diff --git a/tests/framework/test_dynamically_driven.py b/tests/framework/test_dynamically_driven.py index 0a5f316..00bb37b 100644 --- a/tests/framework/test_dynamically_driven.py +++ b/tests/framework/test_dynamically_driven.py @@ -206,7 +206,10 @@ def __init__(self) -> None: path = tmp_path / 'plain.wirebench' save_wirebench(PlainDivider(), path) text = path.read_text() - assert 'dynamically_driven' not in text + # Match the JSON key, not the bare substring — the round-tripped + # source_location field carries the filename of this test module, + # which happens to contain "dynamically_driven". + assert '"dynamically_driven"' not in text def test_save_includes_dynamically_driven_when_true(tmp_path): diff --git a/tests/framework/test_error_messages.py b/tests/framework/test_error_messages.py new file mode 100644 index 0000000..502d077 --- /dev/null +++ b/tests/framework/test_error_messages.py @@ -0,0 +1,122 @@ +"""Every framework exception class carries a `PHYSICAL_JUSTIFICATION` +ClassVar — a one-or-two-sentence physical-world referent that explains +*why this rule exists*. The justification renders as a `Why:` line in +the exception's `str(...)` output, turning the diagnostic from +*"tells you what's wrong"* into *"teaches you why the rule exists."* +""" +from __future__ import annotations + +import inspect + +import pytest + +from framework import errors as E + + +def _every_framework_exception_class() -> list[type[E.WirebenchError]]: + """Discover every WirebenchError subclass declared in framework.errors.""" + classes: list[type[E.WirebenchError]] = [] + for name, obj in vars(E).items(): + if not inspect.isclass(obj): + continue + if not issubclass(obj, E.WirebenchError): + continue + if obj.__module__ != E.__name__: + continue + classes.append(obj) + return classes + + +EVERY_EXCEPTION = _every_framework_exception_class() + + +@pytest.mark.parametrize( + 'cls', EVERY_EXCEPTION, ids=lambda c: c.__name__, +) +def test_every_class_declares_physical_justification( + cls: type[E.WirebenchError], +) -> None: + """Each framework exception class carries a non-empty + `PHYSICAL_JUSTIFICATION` string ending in a period.""" + justification = cls.PHYSICAL_JUSTIFICATION + assert isinstance(justification, str), ( + f"{cls.__name__}.PHYSICAL_JUSTIFICATION must be a str; " + f"got {type(justification).__name__}" + ) + assert justification, ( + f"{cls.__name__}.PHYSICAL_JUSTIFICATION is empty — every class " + f"must surface a physical justification." + ) + assert justification.rstrip().endswith('.'), ( + f"{cls.__name__}.PHYSICAL_JUSTIFICATION must end with '.'; " + f"got {justification!r}" + ) + + +@pytest.mark.parametrize( + 'cls', EVERY_EXCEPTION, ids=lambda c: c.__name__, +) +def test_justification_appears_in_str_output( + cls: type[E.WirebenchError], +) -> None: + """The class's `PHYSICAL_JUSTIFICATION` is included in `str(...)`.""" + instance = cls('test message') + rendered = str(instance) + assert 'Why: ' in rendered, ( + f"{cls.__name__}'s str(...) must include a 'Why:' line. Got:\n" + f"{rendered!r}" + ) + assert cls.PHYSICAL_JUSTIFICATION in rendered, ( + f"{cls.__name__}'s str(...) must include the full " + f"PHYSICAL_JUSTIFICATION text. Got:\n{rendered!r}" + ) + + +def test_str_keeps_base_message_as_first_line() -> None: + """The append-only design: the original message remains the head + of the rendered output so existing `pytest.raises(..., match=...)` + patterns continue to find it.""" + e = E.ShortCircuitError("multi-driver short") + assert str(e).splitlines()[0] == "multi-driver short" + + +def test_source_locations_render_when_provided() -> None: + """When the exception carries source locations, they render as a + `Wired at:` line after the justification.""" + e = E.ShortCircuitError( + "multi-driver short", + source_locations=[('hello_led.py', 14), ('hello_led.py', 18)], + ) + rendered = str(e) + assert 'Wired at: ' in rendered + assert 'hello_led.py:14' in rendered + assert 'hello_led.py:18' in rendered + + +def test_source_locations_absent_when_not_provided() -> None: + """No source-location line when the exception was raised without + attribution — keeps the message tight for non-wiring defects.""" + e = E.RefdesError("bad refdes") + assert 'Wired at:' not in str(e) + + +def test_source_locations_property_is_tuple() -> None: + """`source_locations` exposes an immutable tuple regardless of how + the constructor was called.""" + e = E.ShortCircuitError( + "x", source_locations=[('a.py', 1), ('b.py', 2)], + ) + assert e.source_locations == (('a.py', 1), ('b.py', 2)) + assert isinstance(e.source_locations, tuple) + + +def test_existing_match_patterns_still_hit() -> None: + """Append-only design: regexes anchored on the base message still + find their target after the new `Why:` / `Wired at:` lines.""" + import re + e = E.ShortCircuitError( + "Short circuit on logical net — multiple drivers: foo, bar", + source_locations=[('x.py', 1)], + ) + assert re.search(r'^Short circuit on logical net', str(e), + re.MULTILINE) diff --git a/tests/framework/test_error_with_source_location.py b/tests/framework/test_error_with_source_location.py new file mode 100644 index 0000000..498cc1c --- /dev/null +++ b/tests/framework/test_error_with_source_location.py @@ -0,0 +1,140 @@ +"""Framework errors that fire from `wire()` or `Circuit._validate` +carry source locations attributing the defect to the user's `wire()` +call site(s). The locations render as a `Wired at:` line in the +exception's `str(...)` output. +""" +from __future__ import annotations + +import inspect + +import pytest + +from framework.ground import ELECTRICAL, GroundDomain +from framework.port import Direction, Port +from framework.signals import Analog, Digital +from framework.wire import wire +from framework.errors import ( + DomainCrossingError, EmptyWireError, FloatingNetError, NodeMergeError, + ShortCircuitError, SignalTypeMismatchError, +) + + +def _next_line() -> int: + """Return the line number that immediately follows the caller's + call to this helper — i.e. the line the next statement occupies.""" + frame = inspect.currentframe() + assert frame is not None and frame.f_back is not None + return frame.f_back.f_lineno + 1 + + +# --------------------------------------------------- wire-time errors + + +def test_short_circuit_error_records_call_site() -> None: + out1 = Port('o1', Direction.OUT, ELECTRICAL, signal_type=Digital) + out2 = Port('o2', Direction.OUT, ELECTRICAL, signal_type=Digital) + with pytest.raises(ShortCircuitError) as exc_info: + expected_line = _next_line() + wire(out1, out2) + locs = exc_info.value.source_locations + assert len(locs) == 1 + filename, lineno = locs[0] + assert filename.endswith('test_error_with_source_location.py') + assert lineno == expected_line + assert 'Wired at:' in str(exc_info.value) + + +def test_empty_wire_error_records_call_site() -> None: + with pytest.raises(EmptyWireError) as exc_info: + expected_line = _next_line() + wire() + locs = exc_info.value.source_locations + assert len(locs) == 1 + _, lineno = locs[0] + assert lineno == expected_line + + +def test_domain_crossing_error_records_call_site() -> None: + iso = GroundDomain('test_iso_xc_dom') + a = Port('a', Direction.OUT, ELECTRICAL, signal_type=Digital) + b = Port('b', Direction.IN, iso, signal_type=Digital) + with pytest.raises(DomainCrossingError) as exc_info: + expected_line = _next_line() + wire(a, b) + _, lineno = exc_info.value.source_locations[0] + assert lineno == expected_line + + +def test_floating_net_error_records_call_site() -> None: + """All-IN ports → wire() refuses immediately with source attribution.""" + a = Port('a', Direction.IN, ELECTRICAL, signal_type=Digital) + b = Port('b', Direction.IN, ELECTRICAL, signal_type=Digital) + with pytest.raises(FloatingNetError) as exc_info: + expected_line = _next_line() + wire(a, b) + _, lineno = exc_info.value.source_locations[0] + assert lineno == expected_line + + +def test_signal_type_mismatch_records_call_site() -> None: + a = Port('a', Direction.OUT, ELECTRICAL, signal_type=Analog) + b = Port('b', Direction.IN, ELECTRICAL, signal_type=Digital) + with pytest.raises(SignalTypeMismatchError) as exc_info: + expected_line = _next_line() + wire(a, b) + _, lineno = exc_info.value.source_locations[0] + assert lineno == expected_line + + +def test_node_merge_error_records_call_sites_for_each_prior_wire() -> None: + """Merging two pre-existing nets surfaces all involved call sites: + the current `wire()` plus the two earlier calls whose nodes the + merge would join.""" + a_out = Port('ao', Direction.OUT, ELECTRICAL, signal_type=Digital) + a_bid = Port('ab', Direction.BIDIR, ELECTRICAL, signal_type=Digital) + b_out = Port('bo', Direction.OUT, ELECTRICAL, signal_type=Digital) + b_bid = Port('bb', Direction.BIDIR, ELECTRICAL, signal_type=Digital) + + line_a = _next_line() + wire(a_out, a_bid) + line_b = _next_line() + wire(b_out, b_bid) + with pytest.raises(NodeMergeError) as exc_info: + line_merge = _next_line() + wire(a_bid, b_bid) + + lines = [n for _, n in exc_info.value.source_locations] + assert line_merge in lines + assert line_a in lines + assert line_b in lines + + +# --------------------------------------------- _validate-time errors + + +def test_circuit_validate_short_records_source_locations() -> None: + """`Circuit._validate` walks logical nets; when it raises + `ShortCircuitError`, the offending `wire()` calls are attributed.""" + from framework.circuit import Circuit + from components.passives.rail import Rail + from components.passives.resistor import Resistor + + class Shorted(Circuit): + def __init__(self) -> None: + self.hi = Rail(True) + self.lo = Rail(False) + self.r = Resistor(330, refdes_number=1) + wire(self.hi.out, self.r.t1) + wire(self.lo.out, self.r.t1) + wire(self.r.t2, self.lo.out) + super().__init__() + + with pytest.raises(ShortCircuitError) as exc_info: + Shorted() + files = [f for f, _ in exc_info.value.source_locations] + assert any( + f.endswith('test_error_with_source_location.py') for f in files + ) + # Both shorting wire() calls contribute their source location to + # the offending net, so the attribution includes ≥2 entries. + assert len(exc_info.value.source_locations) >= 2 diff --git a/tests/framework/test_source_location_roundtrip.py b/tests/framework/test_source_location_roundtrip.py new file mode 100644 index 0000000..3788fc2 --- /dev/null +++ b/tests/framework/test_source_location_roundtrip.py @@ -0,0 +1,102 @@ +"""`.wirebench` save / load preserves the `source_location` recorded +on each WireRecord, so a reconstructed design still attributes errors +back to the user's original `wire()` call site rather than to the +loader. When no source was captured, the field is omitted from the +emitted JSON. +""" +from __future__ import annotations + +import json + +from framework.circuit import Circuit +from framework.format import load_wirebench, save_wirebench +from framework.signals import Analog +from framework.wire import wire +from components.passives.rail import Rail +from components.passives.resistor import Resistor + + +class PlainDivider(Circuit): + """Minimal 2-wire divider for round-trip tests.""" + + def __init__(self) -> None: + self.vcc = Rail(True, signal_type=Analog) + self.gnd = Rail(False, signal_type=Analog) + self.r1 = Resistor(10_000, refdes_number=1) + wire(self.vcc.out, self.r1.t1) + wire(self.r1.t2, self.gnd.out) + super().__init__() + + +def _node_for_port(circuit: Circuit, refdes: str, port_name: str): + for part in circuit.parts: + if getattr(part, 'refdes', None) == refdes: + return part.ports[port_name].node + raise AssertionError(f"No port {refdes}.{port_name} found") + + +def test_source_location_round_trips_through_wirebench(tmp_path) -> None: + """Save a circuit then load it back; each reconstructed node carries + the same `source_location` it had when saved.""" + original = PlainDivider() + saved_locs = { + ('R1', 't1'): _node_for_port(original, 'R1', 't1').source_locations, + ('R1', 't2'): _node_for_port(original, 'R1', 't2').source_locations, + } + # Sanity: capture actually happened for both wires. + assert all(saved_locs.values()) + + path = tmp_path / 'plain.wirebench' + save_wirebench(original, path) + + reloaded = load_wirebench(path) + assert isinstance(reloaded, Circuit) + reloaded_locs = { + ('R1', 't1'): _node_for_port(reloaded, 'R1', 't1').source_locations, + ('R1', 't2'): _node_for_port(reloaded, 'R1', 't2').source_locations, + } + for key, original_value in saved_locs.items(): + # The loader preserves at least the first saved location. + assert reloaded_locs[key], ( + f"source_locations vanished after round-trip for {key}" + ) + assert original_value[0] in reloaded_locs[key], ( + f"original location {original_value[0]} missing after " + f"round-trip for {key}; got {reloaded_locs[key]}" + ) + + +def test_source_location_omitted_when_none(tmp_path) -> None: + """When no source_location is attached to a node, the field stays + out of the emitted JSON — keeps the format compact for inputs that + were constructed outside the user-facing wire() path.""" + # Build a divider, then manually clear source locations to simulate + # nodes that the framework couldn't attribute. + circuit = PlainDivider() + for part in circuit.parts: + for port in part.ports.values(): + if port.node is not None: + port.node._source_locations.clear() + path = tmp_path / 'unattributed.wirebench' + save_wirebench(circuit, path) + text = path.read_text() + assert '"source_location"' not in text + + +def test_source_location_appears_in_emitted_json_when_present( + tmp_path, +) -> None: + """The field shows up as a JSON list of [filename, lineno] for every + wire that has an attributed call site.""" + path = tmp_path / 'attributed.wirebench' + save_wirebench(PlainDivider(), path) + obj = json.loads(path.read_text()) + wires = obj['root']['wires'] + assert wires + for w in wires: + assert 'source_location' in w + loc = w['source_location'] + assert isinstance(loc, list) and len(loc) == 2 + filename, lineno = loc + assert filename.endswith('test_source_location_roundtrip.py') + assert isinstance(lineno, int) diff --git a/tests/framework/test_wire_source_location.py b/tests/framework/test_wire_source_location.py new file mode 100644 index 0000000..3042b95 --- /dev/null +++ b/tests/framework/test_wire_source_location.py @@ -0,0 +1,74 @@ +"""Every `wire()` call captures its source location and records it on +the resulting `Node`. The capture is automatic (from the caller's +frame), survives `validate_call`'s pydantic wrapper, and accumulates +across calls that extend an existing node. +""" +from __future__ import annotations + +import inspect + +from framework.ground import ELECTRICAL +from framework.port import Direction, Port +from framework.signals import Digital +from framework.wire import wire + + +def _make_pair() -> tuple[Port, Port]: + out = Port('o', Direction.OUT, ELECTRICAL, signal_type=Digital) + inp = Port('i', Direction.IN, ELECTRICAL, signal_type=Digital) + return out, inp + + +def _here() -> int: + """Return the line number of the *caller's* call to this helper.""" + frame = inspect.currentframe() + assert frame is not None and frame.f_back is not None + return frame.f_back.f_lineno + + +def test_node_records_source_location_from_user_frame() -> None: + """A bare `wire()` captures the test file and the exact line.""" + out, inp = _make_pair() + wire(out, inp); expected_line = _here() + assert out.node is not None + locs = out.node.source_locations + assert len(locs) == 1 + filename, lineno = locs[0] + assert filename.endswith('test_wire_source_location.py'), filename + assert lineno == expected_line + + +def test_multiple_wire_calls_accumulate_on_same_node() -> None: + """When a second `wire()` extends an existing node, both source + locations are recorded in call order.""" + a_out = Port('ao', Direction.OUT, ELECTRICAL, signal_type=Digital) + a_bid = Port('ab', Direction.BIDIR, ELECTRICAL, signal_type=Digital) + a_in = Port('ai', Direction.IN, ELECTRICAL, signal_type=Digital) + + wire(a_out, a_bid); first_line = _here() + wire(a_bid, a_in); second_line = _here() + + locs = a_out.node.source_locations + assert len(locs) == 2 + files = [f for f, _ in locs] + lines = [n for _, n in locs] + assert all(f.endswith('test_wire_source_location.py') for f in files) + assert lines == [first_line, second_line] + + +def test_explicit_source_location_overrides_capture() -> None: + """The loader passes `source_location` explicitly to attribute a + reconstructed wire to the user's original code rather than to the + loader itself.""" + out, inp = _make_pair() + wire(out, inp, source_location=('hello_led.py', 14)) + assert out.node.source_locations == (('hello_led.py', 14),) + + +def test_source_locations_is_immutable_tuple() -> None: + """The public `source_locations` property hands out a tuple — the + internal list isn't directly exposed.""" + out, inp = _make_pair() + wire(out, inp) + locs = out.node.source_locations + assert isinstance(locs, tuple) From 6e4574d7c28e50c7cc3f541f265da6dfc6c3355c Mon Sep 17 00:00:00 2001 From: subzero Date: Wed, 20 May 2026 19:19:48 +0200 Subject: [PATCH 2/3] review: preserve unattributed nodes when loading legacy .wirebench files MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Reviewer flagged that the loader passing `source_location=None` into the public `wire()` triggered auto-capture pointing at format.py frames — fabricating attribution to framework internals and corrupting legacy files on re-save. Fix: - Split `wire()` into a public auto-capturing API and an internal `_wire_with_attribution(ports, *, dynamically_driven, source_location)` that takes attribution as-is. `None` now genuinely means *no attribution* — the node carries no source_locations entry and the emitted .wirebench keeps the field omitted. - The .wirebench loader uses `_wire_with_attribution` directly so legacy records (where `source_location` was absent) reconstruct unattributed instead of inheriting loader-frame attribution. Regression coverage: - `test_legacy_file_loads_without_fabricating_source_attribution` — hand-rolled pre-Phase-2b.1 JSON (no source_location field) loads with `port.node.source_locations == ()` for every reconstructed net. - `test_legacy_file_round_trips_without_injecting_source_location` — load → save of that file produces output whose wire records still omit `source_location`, so the round-trip is field-preserving. - `test_internal_helper_with_none_leaves_node_unattributed` — explicit `source_location=None` to the internal helper bypasses auto-capture even though a valid caller frame exists. Suite: 4729 passed (3 new), mypy clean. --- src/framework/format.py | 18 +++-- src/framework/wire.py | 46 ++++++++---- .../test_source_location_roundtrip.py | 72 +++++++++++++++++++ tests/framework/test_wire_source_location.py | 29 ++++++-- 4 files changed, 141 insertions(+), 24 deletions(-) diff --git a/src/framework/format.py b/src/framework/format.py index f377081..d094673 100644 --- a/src/framework/format.py +++ b/src/framework/format.py @@ -23,7 +23,7 @@ from framework.refdes import RefdesBearing from framework.errors import LoadError, SaveError from framework.registry import lookup -from framework.wire import wire +from framework.wire import _wire_with_attribution from framework.format_extension import deserialize_kwargs, serialize_kwargs from framework.format_records import ( AssemblyRecord, BoardRecord, CircuitRecord, CircuitryFile, @@ -521,8 +521,12 @@ def _from_circuit_record(record: CircuitRecord) -> Circuit: components, by_id = _rebuild_circuit_components(record.components) for w in record.wires: ports = [_resolve_port(r, by_id) for r in w.ports] - wire( - *ports, + # Use the internal helper so a `None` `source_location` from + # legacy files stays unattributed — calling the public `wire()` + # would auto-capture the loader's frame and fabricate + # attribution to framework internals. + _wire_with_attribution( + ports, dynamically_driven=w.dynamically_driven, source_location=w.source_location, ) @@ -537,8 +541,12 @@ def _from_board_record(record: BoardRecord) -> Board: components, by_id = _rebuild_circuit_components(record.components) for w in record.wires: ports = [_resolve_port(r, by_id) for r in w.ports] - wire( - *ports, + # Use the internal helper so a `None` `source_location` from + # legacy files stays unattributed — calling the public `wire()` + # would auto-capture the loader's frame and fabricate + # attribution to framework internals. + _wire_with_attribution( + ports, dynamically_driven=w.dynamically_driven, source_location=w.source_location, ) diff --git a/src/framework/wire.py b/src/framework/wire.py index e2e299a..1f2487c 100644 --- a/src/framework/wire.py +++ b/src/framework/wire.py @@ -1,5 +1,6 @@ import sys import warnings +from collections.abc import Sequence from types import FrameType from pydantic import validate_call @@ -21,7 +22,7 @@ def _capture_user_call_site() -> tuple[str, int] | None: 'pydantic' appearing in the filename — robust across pydantic versions, where the number of wrapper frames varies. """ - frame: 'FrameType | None' = sys._getframe(1) # skip wire() itself + frame: FrameType | None = sys._getframe(1) # skip wire() itself while frame is not None: filename = frame.f_code.co_filename if 'pydantic' not in filename and filename != __file__: @@ -31,11 +32,7 @@ def _capture_user_call_site() -> tuple[str, int] | None: @validate_call(config={"arbitrary_types_allowed": True}) -def wire( - *ports: Port, - dynamically_driven: bool = False, - source_location: tuple[str, int] | None = None, -) -> None: +def wire(*ports: Port, dynamically_driven: bool = False) -> None: """Connect ports together, creating the junction node at their meeting point. Enforces: @@ -51,14 +48,37 @@ def wire( Promotion is one-way: once a node is annotated, subsequent wire() calls that omit the kwarg do not clear it. - `source_location` is the `(filename, lineno)` to attribute this - `wire()` call to. Captured automatically from the caller's frame - when omitted; the `.wirebench` loader overrides with the - serialised location so reconstructed nodes carry the original - user-source attribution. + The caller's `(filename, lineno)` is captured automatically and + recorded on the resulting `Node`, then surfaced in framework error + messages as a `Wired at:` line. The `.wirebench` loader uses + `_wire_with_attribution` directly so legacy files without a + captured source stay unattributed across save → load → save. + """ + _wire_with_attribution( + list(ports), + dynamically_driven=dynamically_driven, + source_location=_capture_user_call_site(), + ) + + +def _wire_with_attribution( + ports: Sequence[Port], + *, + dynamically_driven: bool, + source_location: tuple[str, int] | None, +) -> None: + """Internal wire-implementation that takes source attribution + explicitly. Pass `source_location=None` to mean *no attribution*: + the resulting Node carries no `source_locations` entry, so + diagnostics for that net stay free of a fabricated `Wired at:` + line and the round-tripped `.wirebench` keeps the field omitted. + + Used by the `.wirebench` loader so reconstructed nodes carry the + original user-source attribution (when present in the file) or + stay unattributed (when the file pre-dates source-location + capture). Not part of the public API — user code should call + `wire()` and let it auto-capture. """ - if source_location is None: - source_location = _capture_user_call_site() call_locs: list[tuple[str, int]] = ( [source_location] if source_location is not None else [] ) diff --git a/tests/framework/test_source_location_roundtrip.py b/tests/framework/test_source_location_roundtrip.py index 3788fc2..2124902 100644 --- a/tests/framework/test_source_location_roundtrip.py +++ b/tests/framework/test_source_location_roundtrip.py @@ -100,3 +100,75 @@ def test_source_location_appears_in_emitted_json_when_present( filename, lineno = loc assert filename.endswith('test_source_location_roundtrip.py') assert isinstance(lineno, int) + + +# ----------------------------------------- legacy-file regression + + +def _legacy_wirebench_json() -> str: + """A `.wirebench` payload predating source_location capture — every + wire record omits the field entirely. Modelled on the bytes a + pre-Phase-2b.1 `save_wirebench` would have produced.""" + return ( + '{\n' + ' "format_version": "1.0.0",\n' + ' "name": null,\n' + ' "root": {\n' + ' "type": "Circuit",\n' + ' "components": [\n' + ' {"type": "Rail", "id": "Rail_0", "level": true, "domain": null, "signal_type": "Analog"},\n' + ' {"type": "Rail", "id": "Rail_1", "level": false, "domain": null, "signal_type": "Analog"},\n' + ' {"type": "Resistor", "refdes": "R1", "ohms": 10000.0}\n' + ' ],\n' + ' "wires": [\n' + ' {"ports": ["R1.t1", "Rail_0.out"]},\n' + ' {"ports": ["R1.t2", "Rail_1.out"]}\n' + ' ],\n' + ' "surface_ports": {}\n' + ' }\n' + '}\n' + ) + + +def test_legacy_file_loads_without_fabricating_source_attribution( + tmp_path, +) -> None: + """A `.wirebench` file written before Phase 2b.1 has no + `source_location` field on its wire records. Loading must keep + each reconstructed node *unattributed* — fabricating loader-frame + attribution would mislead diagnostics and corrupt the file on + re-save.""" + path = tmp_path / 'legacy.wirebench' + path.write_text(_legacy_wirebench_json()) + loaded = load_wirebench(path) + assert isinstance(loaded, Circuit) + for part in loaded.parts: + for port in part.ports.values(): + if port.node is None: + continue + assert port.node.source_locations == (), ( + f"Loader fabricated source attribution for " + f"{type(part).__name__}.{port.name}: " + f"{port.node.source_locations}" + ) + + +def test_legacy_file_round_trips_without_injecting_source_location( + tmp_path, +) -> None: + """Re-saving a legacy file must not inject `source_location` data + that wasn't there originally — the file's wire records stay + field-for-field equivalent after load → save.""" + in_path = tmp_path / 'legacy.wirebench' + out_path = tmp_path / 'roundtripped.wirebench' + in_path.write_text(_legacy_wirebench_json()) + + loaded = load_wirebench(in_path) + save_wirebench(loaded, out_path) + + out_obj = json.loads(out_path.read_text()) + for w in out_obj['root']['wires']: + assert 'source_location' not in w, ( + f"Re-saved legacy file injected source_location into wire " + f"record: {w}" + ) diff --git a/tests/framework/test_wire_source_location.py b/tests/framework/test_wire_source_location.py index 3042b95..d8f93de 100644 --- a/tests/framework/test_wire_source_location.py +++ b/tests/framework/test_wire_source_location.py @@ -10,7 +10,7 @@ from framework.ground import ELECTRICAL from framework.port import Direction, Port from framework.signals import Digital -from framework.wire import wire +from framework.wire import _wire_with_attribution, wire def _make_pair() -> tuple[Port, Port]: @@ -56,15 +56,32 @@ def test_multiple_wire_calls_accumulate_on_same_node() -> None: assert lines == [first_line, second_line] -def test_explicit_source_location_overrides_capture() -> None: - """The loader passes `source_location` explicitly to attribute a - reconstructed wire to the user's original code rather than to the - loader itself.""" +def test_explicit_source_location_via_internal_helper() -> None: + """The `.wirebench` loader uses `_wire_with_attribution` directly + to attribute a reconstructed wire to the user's original source + line rather than to the loader's own frame.""" out, inp = _make_pair() - wire(out, inp, source_location=('hello_led.py', 14)) + _wire_with_attribution( + [out, inp], + dynamically_driven=False, + source_location=('hello_led.py', 14), + ) assert out.node.source_locations == (('hello_led.py', 14),) +def test_internal_helper_with_none_leaves_node_unattributed() -> None: + """Passing `source_location=None` to the internal helper means *no + attribution*: the resulting node carries no source location, even + though we're inside a frame the auto-capture would otherwise grab.""" + out, inp = _make_pair() + _wire_with_attribution( + [out, inp], + dynamically_driven=False, + source_location=None, + ) + assert out.node.source_locations == () + + def test_source_locations_is_immutable_tuple() -> None: """The public `source_locations` property hands out a tuple — the internal list isn't directly exposed.""" From e7ee507d638264b7fb6783d794aef2fefb9c0479 Mon Sep 17 00:00:00 2001 From: subzero Date: Wed, 20 May 2026 19:24:46 +0200 Subject: [PATCH 3/3] =?UTF-8?q?review:=20harden=20=5Fcapture=5Fuser=5Fcall?= =?UTF-8?q?=5Fsite=20=E2=80=94=20guard=20=5Fgetframe,=20store=20basename?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two inline review concerns on `_capture_user_call_site`: - `sys._getframe(1)` is documented to raise `ValueError` when the requested depth isn't available. Vanishingly rare from inside a real function, but guarding it costs nothing and makes the helper robust to bizarre invocation contexts. - `frame.f_code.co_filename` is typically an absolute path; persisting that into `.wirebench` saved on one machine and loaded on another would carry someone else's `/Users/...`/`/home/...`/`C:\\...` paths through the file. Normalise to `os.path.basename` at capture time so attribution stays portable. Aligns with the spec's own example rendering (`Wired at: hello_led.py:14`, not a full path). Added `test_captured_filename_is_basename_for_portability` pinning the basename behavior; existing `.endswith('filename.py')` assertions already match basenames so other tests stay green. Suite: 4730 passed (1 new), mypy clean. --- src/framework/wire.py | 27 +++++++++++++++----- tests/framework/test_wire_source_location.py | 14 ++++++++++ 2 files changed, 35 insertions(+), 6 deletions(-) diff --git a/src/framework/wire.py b/src/framework/wire.py index 1f2487c..3eb8081 100644 --- a/src/framework/wire.py +++ b/src/framework/wire.py @@ -1,3 +1,4 @@ +import os import sys import warnings from collections.abc import Sequence @@ -17,16 +18,30 @@ def _capture_user_call_site() -> tuple[str, int] | None: """Walk up the call stack from inside `wire()` to find the first frame that isn't inside pydantic's `validate_call` wrapper. - Returns `(filename, lineno)` of the user's call site, or `None` if - no caller frame is available. Pydantic frames are identified by - 'pydantic' appearing in the filename — robust across pydantic - versions, where the number of wrapper frames varies. + Returns `(basename, lineno)` of the user's call site, or `None` if + no caller frame is available or the stack walk fails. Pydantic + frames are identified by 'pydantic' appearing in the filename — + robust across pydantic versions, where the number of wrapper + frames varies. + + Filenames are normalised to `os.path.basename` so attribution stays + portable: `.wirebench` files saved on one machine and loaded on + another won't carry someone else's `/Users/...`/`/home/...`/`C:\\...` + absolute paths. Basename collisions across files in a project are + rare in practice (one demo per file) and the trade-off is worth + the portability. """ - frame: FrameType | None = sys._getframe(1) # skip wire() itself + try: + frame: FrameType | None = sys._getframe(1) # skip wire() itself + except ValueError: + # `sys._getframe` raises ValueError when called at a depth the + # interpreter can't satisfy — vanishingly rare from inside a + # real function, but documented behavior worth guarding. + return None while frame is not None: filename = frame.f_code.co_filename if 'pydantic' not in filename and filename != __file__: - return (filename, frame.f_lineno) + return (os.path.basename(filename), frame.f_lineno) frame = frame.f_back return None diff --git a/tests/framework/test_wire_source_location.py b/tests/framework/test_wire_source_location.py index d8f93de..996f872 100644 --- a/tests/framework/test_wire_source_location.py +++ b/tests/framework/test_wire_source_location.py @@ -69,6 +69,20 @@ def test_explicit_source_location_via_internal_helper() -> None: assert out.node.source_locations == (('hello_led.py', 14),) +def test_captured_filename_is_basename_for_portability() -> None: + """Auto-captured source locations store the file *basename*, never + an absolute path — so `.wirebench` files written on one machine + stay portable across users / filesystems / CI.""" + import os + out, inp = _make_pair() + wire(out, inp) + filename, _ = out.node.source_locations[0] + assert filename == 'test_wire_source_location.py' + assert os.sep not in filename, ( + f"Captured filename should be a basename; got {filename!r}" + ) + + def test_internal_helper_with_none_leaves_node_unattributed() -> None: """Passing `source_location=None` to the internal helper means *no attribution*: the resulting node carries no source location, even