[Pather] add route completion callback

masque/test/test_pather_route_completion.py

@@ -0,0 +1,166 @@
+from collections.abc import Mapping, Sequence
+from typing import Any, Literal
+
+import numpy
+import pytest
+
+from masque import Library, Pather, Port
+from masque.builder import PathTool, RenderStep, RouteCompletionCallback, Tool
+from masque.error import BuildError
+
+
+def test_route_completion_can_label_before_automatic_render() -> None:
+    calls = 0
+
+    def label_endpoints(pather: Pather, endpoints: Mapping[str, Port]) -> None:
+        nonlocal calls
+        calls += 1
+        assert not pather.pattern.has_shapes()
+        assert numpy.allclose(pather.ports['A'].offset, endpoints['A'].offset)
+        for name, port in endpoints.items():
+            pather.label('LABELS', string=name, offset=port.offset)
+
+    callback: RouteCompletionCallback = label_endpoints
+    p = Pather(
+        Library(),
+        tools=PathTool(layer='M1', width=1, ptype='wire'),
+        on_route_complete=callback,
+        )
+    p.ports['A'] = Port((0, 0), rotation=0, ptype='wire')
+
+    p.at('A').straight(5)
+
+    assert calls == 1
+    assert p.pattern.has_shapes()
+    assert len(p.pattern.labels['LABELS']) == 1
+    assert p.pattern.labels['LABELS'][0].string == 'A'
+    assert numpy.allclose(p.pattern.labels['LABELS'][0].offset, p.ports['A'].offset)
+
+
+def test_bundle_route_completion_is_ordered_read_only_and_isolated() -> None:
+    calls: list[Mapping[str, Port]] = []
+
+    def record(pather: Pather, endpoints: Mapping[str, Port]) -> None:
+        _ = pather
+        calls.append(endpoints)
+        with pytest.raises(TypeError):
+            endpoints['extra'] = Port((0, 0))  # type: ignore[index]
+        endpoints['B'].translate((100, 100))
+
+    p = Pather(
+        Library(),
+        tools=PathTool(layer='M1', width=1, ptype='wire'),
+        render='deferred',
+        on_route_complete=record,
+        )
+    p.ports['B'] = Port((0, 2), rotation=0, ptype='wire')
+    p.ports['A'] = Port((0, 0), rotation=0, ptype='wire')
+
+    p.trace(['A', 'B'], None, each=5)
+
+    assert len(calls) == 1
+    assert list(calls[0]) == ['A', 'B']
+    assert numpy.allclose(p.ports['A'].offset, (-5, 0))
+    assert numpy.allclose(p.ports['B'].offset, (-5, 2))
+
+
+def test_trace_into_completion_retains_consumed_source_endpoint() -> None:
+    calls: list[Mapping[str, Port]] = []
+
+    def record(pather: Pather, endpoints: Mapping[str, Port]) -> None:
+        assert set(pather.ports) == {'src'}
+        assert numpy.allclose(pather.ports['src'].offset, (20, 0))
+        calls.append(endpoints)
+
+    p = Pather(
+        Library(),
+        tools=PathTool(layer='M1', width=1, ptype='wire'),
+        render='deferred',
+        on_route_complete=record,
+        )
+    p.ports['src'] = Port((0, 0), rotation=0, ptype='wire')
+    p.ports['dst'] = Port((-10, 0), rotation=numpy.pi, ptype='wire')
+    p.ports['thru'] = Port((20, 0), rotation=0, ptype='wire')
+
+    p.trace_into('src', 'dst', thru='thru')
+
+    assert len(calls) == 1
+    assert list(calls[0]) == ['src']
+    assert numpy.allclose(calls[0]['src'].offset, (-10, 0))
+
+
+def test_route_completion_exception_propagates_before_render() -> None:
+    def fail(pather: Pather, endpoints: Mapping[str, Port]) -> None:
+        _ = pather, endpoints
+        raise RuntimeError('completion failed')
+
+    p = Pather(
+        Library(),
+        tools=PathTool(layer='M1', width=1, ptype='wire'),
+        on_route_complete=fail,
+        )
+    p.ports['A'] = Port((0, 0), rotation=0, ptype='wire')
+
+    with pytest.raises(RuntimeError, match='completion failed'):
+        p.straight('A', 5)
+
+    assert numpy.allclose(p.ports['A'].offset, (-5, 0))
+    assert p._paths['A']
+    assert not p.pattern.has_shapes()
+
+
+class NoRouteTool(Tool):
+    def primitive_offers(
+            self,
+            kind: Literal['straight', 'bend', 's', 'u'],
+            **kwargs: Any,
+            ) -> tuple[()]:
+        _ = kind, kwargs
+        return ()
+
+    def render(
+            self,
+            batch: Sequence[RenderStep],
+            *,
+            port_names: tuple[str, str] = ('A', 'B'),
+            **kwargs: Any,
+            ) -> Library:
+        _ = batch, port_names, kwargs
+        return Library()
+
+
+def test_dead_bundle_fallback_invokes_route_completion_once() -> None:
+    calls: list[Mapping[str, Port]] = []
+    p = Pather(
+        Library(),
+        tools=NoRouteTool(),
+        on_route_complete=lambda _pather, endpoints: calls.append(endpoints),
+        )
+    p.ports['A'] = Port((0, 0), rotation=0, ptype='wire')
+    p.ports['B'] = Port((0, 2), rotation=0, ptype='wire')
+    p.set_dead()
+
+    p.trace(['B', 'A'], None, each=5)
+
+    assert len(calls) == 1
+    assert list(calls[0]) == ['B', 'A']
+    assert numpy.allclose(calls[0]['B'].offset, (-5, 2))
+    assert numpy.allclose(calls[0]['A'].offset, (-5, 0))
+    assert not p.pattern.has_shapes()
+
+
+def test_failed_route_does_not_invoke_route_completion() -> None:
+    calls = 0
+
+    def record(pather: Pather, endpoints: Mapping[str, Port]) -> None:
+        nonlocal calls
+        _ = pather, endpoints
+        calls += 1
+
+    p = Pather(Library(), tools=NoRouteTool(), on_route_complete=record)
+    p.ports['A'] = Port((0, 0), rotation=0, ptype='wire')
+
+    with pytest.raises(BuildError):
+        p.straight('A', 5)
+
+    assert calls == 0