jan/masque
1
1
Fork 0
Code Issues Pull requests Releases Wiki Activity

[planner] make solver carry state across increasing bend counts

Browse source
This commit is contained in:
Jan Petykiewicz 2026-07-09 11:09:52 -07:00
commit f864ebbeab
2 changed files with 186 additions and 101 deletions
Download patch file Download diff file Expand all files Collapse all files

153
masque/builder/planner/planner.py
View file

@ -293,16 +293,24 @@ class Solver:
tuple[PrimitiveKind, str | None, str | None, tuple[tuple[str, Any], ...]],
tuple[PrimitiveOffer, ...],
] = {}
self.seen_candidate_keys: set[tuple[Any, ...]] = set()
self.order = 0
def solve(self) -> Candidate:
"""
Enumerate, finalize, deduplicate, and rank legal candidates.
@staticmethod
def route_bend_count(steps: Sequence[SelectedPrimitive]) -> int:
"""Return the route bend budget consumed by non-adapter primitives."""
count = 0
for step in steps:
if step.role == 'adapter':
continue
if step.route_kind == 'bend':
count += 1
elif step.route_kind in ('s', 'u'):
count += 2
return count
Non-fatal candidate errors are accumulated so the failure message can
preserve useful Tool feedback. Fatal offer-contract errors stop the
solve immediately.
"""
def candidate_key(self, candidate: Candidate) -> tuple[Any, ...]:
"""Return a deterministic key for duplicate solved candidates."""
def endpoint_key(port: Port) -> tuple[float, float, float | None, str | None]:
return (
round(float(port.x), 9),
@ -311,7 +319,6 @@ class Solver:
port.ptype,
)
def candidate_key(candidate: Candidate) -> tuple[Any, ...]:
def offer_key(offer: PrimitiveOffer) -> tuple[Any, ...]:
return (
type(offer).__qualname__,
@ -335,12 +342,28 @@ class Solver:
) for step in candidate.steps),
)
def solve(
self,
*,
min_bends: int = 0,
max_bends: int | None = None,
) -> Candidate:
"""
Enumerate, finalize, deduplicate, and rank legal candidates.
Non-fatal candidate errors are accumulated so the failure message can
preserve useful Tool feedback. Fatal offer-contract errors stop the
solve immediately.
"""
if max_bends is None:
max_bends = self.request.bend_budget
candidates: list[Candidate] = []
errors: list[Exception] = []
seen: set[tuple[Any, ...]] = set()
for steps in self.enumerate_grammar():
for steps in self.enumerate_grammar(max_bends):
if not steps:
continue
if self.route_bend_count(steps) < min_bends:
continue
if any(first.role == 'adapter' and second.role == 'adapter' for first, second in zip(steps, steps[1:], strict=False)):
continue
try:
@ -349,10 +372,10 @@ class Solver:
raise_if_fatal(err)
errors.append(err)
continue
key = candidate_key(candidate)
if key in seen:
key = self.candidate_key(candidate)
if key in self.seen_candidate_keys:
continue
seen.add(key)
self.seen_candidate_keys.add(key)
candidates.append(candidate)
if not candidates:
@ -676,13 +699,13 @@ class Solver:
options.extend((turn, 2) for turn in self.su_primitive_options(steps, 'u', jog))
return tuple(options)
def enumerate_grammar(self) -> Iterable[tuple[SelectedPrimitive, ...]]:
def enumerate_grammar(self, max_bends: int) -> Iterable[tuple[SelectedPrimitive, ...]]:
"""Yield raw primitive sequences allowed by the bounded route grammar."""
residual_jog = 0.0 if self.request.jog is None else float(self.request.jog)
base: tuple[SelectedPrimitive, ...] = ()
for prefix_adapter in self.adapter_options(base, residual_jog=residual_jog):
prefix = (*base, *prefix_adapter)
yield from self.enumerate_segments(prefix, self.request.bend_budget, residual_jog=residual_jog)
yield from self.enumerate_segments(prefix, max_bends, residual_jog=residual_jog)
def enumerate_segments(
self,
@ -909,11 +932,63 @@ class RoutingPlanner:
TRACE_INTO_MAX_BENDS: int = 4
def trace_into_bend_budgets(self, family: PrimitiveKind) -> tuple[int, ...]:
"""Return staged trace_into bend budgets for the route rotation parity."""
def trace_into_bend_bands(self, family: PrimitiveKind) -> tuple[tuple[int, int], ...]:
"""Return non-overlapping trace_into bend-budget bands for staged solving."""
max_bends = self.TRACE_INTO_MAX_BENDS
if family == 'bend':
return tuple(budget for budget in (1, 3) if budget <= self.TRACE_INTO_MAX_BENDS)
return tuple(budget for budget in (2, 4) if budget <= self.TRACE_INTO_MAX_BENDS)
return tuple(band for band in ((1, 1), (3, 3)) if band[1] <= max_bends)
bands: list[tuple[int, int]] = []
if max_bends >= 0:
bands.append((0, 2 if max_bends >= 2 else 0))
if max_bends >= 4:
bands.append((4, 4))
return tuple(bands)
def route_request(
self,
family: PrimitiveKind,
context: RoutePortContext,
*,
length: float | None = None,
jog: float | None = None,
ccw: SupportsBool | None = None,
constrain_jog: bool = False,
max_bends: int | None = None,
**kwargs: Any,
) -> RouteRequest:
"""Build normalized solver input for one route leg."""
return RouteRequest(
family=family,
tool=context.tool,
in_ptype=context.port.ptype,
route_kwargs=kwargs,
length=length,
jog=jog,
ccw=ccw,
out_ptype=kwargs.get('out_ptype'),
constrain_jog=constrain_jog,
max_bends=max_bends,
)
def solver_for_request(self, request: RouteRequest) -> Solver:
"""Construct the solver for a route request."""
return Solver(request)
def route_leg_from_candidate(
self,
context: RoutePortContext,
candidate: Candidate,
*,
plug_into: str | None,
) -> RouteLeg:
"""Attach a solved candidate to its source Pather context."""
return RouteLeg(
portspec=context.portspec,
start_port=context.port.copy(),
tool=context.tool,
candidate=candidate,
plug_into=plug_into,
)
def plan_leg(
self,
@ -929,31 +1004,23 @@ class RoutingPlanner:
**kwargs: Any,
) -> RouteLeg:
"""Solve one route leg and attach it to its source Pather context."""
request = RouteRequest(
request = self.route_request(
family=family,
tool=context.tool,
in_ptype=context.port.ptype,
route_kwargs=kwargs,
context=context,
length=length,
jog=jog,
ccw=ccw,
out_ptype=kwargs.get('out_ptype'),
constrain_jog=constrain_jog,
max_bends=max_bends,
**kwargs,
)
try:
candidate = Solver(request).solve()
candidate = self.solver_for_request(request).solve()
except BuildError as err:
if family == 'u' and length is None and not getattr(err, 'fatal', False):
raise BuildError('No legal primitive offer for omitted-length U-turn') from err
raise
return RouteLeg(
portspec=context.portspec,
start_port=context.port.copy(),
tool=context.tool,
candidate=candidate,
plug_into=plug_into,
)
return self.route_leg_from_candidate(context, candidate, plug_into=plug_into)
def prepared_route_action_from_leg(
self,
@ -1204,30 +1271,36 @@ class RoutingPlanner:
desired.rotation = port_dst.rotation - pi
desired.ptype = out_ptype
family, length, jog, ccw = self.trace_into_spec(context_src.port, desired)
leg = None
last_error: Exception | None = None
for max_bends in self.trace_into_bend_budgets(family):
try:
leg = self.plan_leg(
request = self.route_request(
family,
context_src,
length=length,
jog=jog,
ccw=ccw,
plug_into=portspec_dst if plug_destination else None,
constrain_jog=family == 'bend',
max_bends=max_bends,
max_bends=self.TRACE_INTO_MAX_BENDS,
**(dict(kwargs) | {'out_ptype': out_ptype}),
)
solver = self.solver_for_request(request)
candidate = None
last_error: Exception | None = None
for min_bends, max_bends in self.trace_into_bend_bands(family):
try:
candidate = solver.solve(min_bends=min_bends, max_bends=max_bends)
break
except (BuildError, NotImplementedError) as err:
if route_error_is_fatal(err):
raise
last_error = err
if leg is None:
if candidate is None:
if last_error is not None:
raise last_error
raise BuildError('No legal primitive offer for trace_into route')
leg = self.route_leg_from_candidate(
context_src,
candidate,
plug_into=portspec_dst if plug_destination else None,
)
renames = ((thru, context_src.portspec),) if thru is not None else ()
return self.prepared_result_from_legs(
(leg,),

82
masque/test/test_pather_trace_into.py
View file

@ -7,7 +7,7 @@ from numpy.testing import assert_equal
from masque import Library, PathTool, Port, Pather
from masque.builder.planner import PreparedRouteResult, RoutePlanningError, RoutePortContext, RoutingPlanner
from masque.builder.planner.planner import Candidate, RouteLeg
from masque.builder.planner.planner import Candidate, RouteRequest
from masque.builder.tools import BendOffer, PrimitiveOffer, StraightOffer, Tool
from masque.error import BuildError, PortError
@ -264,38 +264,37 @@ def test_pather_trace_into_rejects_reserved_route_kwargs(
assert len(p._paths['A']) == 0
class TraceIntoBudgetPlanner(RoutingPlanner):
def __init__(self, successes: set[int], fatal_at: set[int] | None = None) -> None:
class TraceIntoBudgetSolver:
def __init__(self, successes: set[tuple[int, int]], fatal_at: set[tuple[int, int]] | None = None) -> None:
self.successes = successes
self.fatal_at = set() if fatal_at is None else fatal_at
self.attempts: list[int | None] = []
self.attempts: list[tuple[int, int]] = []
def plan_leg(
def solve(
self,
family: Any,
context: RoutePortContext,
*,
length: float | None = None,
jog: float | None = None,
ccw: Any = None,
plug_into: str | None = None,
constrain_jog: bool = False,
min_bends: int = 0,
max_bends: int | None = None,
**kwargs: Any,
) -> RouteLeg:
_ = family, length, jog, ccw, constrain_jog, kwargs
self.attempts.append(max_bends)
if max_bends in self.fatal_at:
) -> Candidate:
assert max_bends is not None
band = (min_bends, max_bends)
self.attempts.append(band)
if band in self.fatal_at:
raise RoutePlanningError('fatal', fatal=True)
if max_bends not in self.successes:
if band not in self.successes:
raise BuildError('try next budget')
return RouteLeg(
portspec=context.portspec,
start_port=context.port.copy(),
tool=context.tool,
candidate=Candidate((), context.port.copy(), 0.0, 0, 0.0),
plug_into=plug_into,
)
return Candidate((), Port((0, 0), rotation=0, ptype='wire'), 0.0, 0, 0.0)
class TraceIntoBudgetPlanner(RoutingPlanner):
def __init__(self, successes: set[tuple[int, int]], fatal_at: set[tuple[int, int]] | None = None) -> None:
self.solver = TraceIntoBudgetSolver(successes, fatal_at=fatal_at)
self.solver_requests = 0
def solver_for_request(self, request: RouteRequest) -> Any:
_ = request
self.solver_requests += 1
return self.solver
def prepared_result_from_legs(
self,
@ -310,30 +309,43 @@ class TraceIntoBudgetPlanner(RoutingPlanner):
@pytest.mark.parametrize(
('dst', 'successes', 'attempts'),
[
(Port((-10, 0), rotation=pi, ptype='wire'), {2}, [2]),
(Port((-10, 0), rotation=pi, ptype='wire'), {4}, [2, 4]),
(Port((-10, -10), rotation=3 * pi / 2, ptype='wire'), {1}, [1]),
(Port((-10, -10), rotation=3 * pi / 2, ptype='wire'), {3}, [1, 3]),
(Port((-10, 0), rotation=pi, ptype='wire'), {(0, 2)}, [(0, 2)]),
(Port((-10, 0), rotation=pi, ptype='wire'), {(4, 4)}, [(0, 2), (4, 4)]),
(Port((-10, -10), rotation=3 * pi / 2, ptype='wire'), {(1, 1)}, [(1, 1)]),
(Port((-10, -10), rotation=3 * pi / 2, ptype='wire'), {(3, 3)}, [(1, 1), (3, 3)]),
],
)
def test_trace_into_uses_staged_bend_budgets(
def test_trace_into_reuses_solver_across_staged_bend_bands(
dst: Port,
successes: set[int],
attempts: list[int],
successes: set[tuple[int, int]],
attempts: list[tuple[int, int]],
) -> None:
planner = TraceIntoBudgetPlanner(successes)
context = RoutePortContext('src', Port((0, 0), rotation=0, ptype='wire'), PathTool(layer='M1', width=1, ptype='wire'))
planner.plan_trace_into(context, 'dst', dst, out_ptype=None, plug_destination=True, thru=None)
assert planner.attempts == attempts
assert planner.solver.attempts == attempts
assert planner.solver_requests == 1
def test_trace_into_staged_bend_budget_stops_on_fatal_error() -> None:
planner = TraceIntoBudgetPlanner({4}, fatal_at={2})
planner = TraceIntoBudgetPlanner({(4, 4)}, fatal_at={(0, 2)})
context = RoutePortContext('src', Port((0, 0), rotation=0, ptype='wire'), PathTool(layer='M1', width=1, ptype='wire'))
with pytest.raises(RoutePlanningError, match='fatal'):
planner.plan_trace_into(context, 'dst', Port((-10, 0), rotation=pi, ptype='wire'), out_ptype=None, plug_destination=True, thru=None)
assert planner.attempts == [2]
assert planner.solver.attempts == [(0, 2)]
assert planner.solver_requests == 1
def test_trace_into_bend_bands_respect_max_bends() -> None:
class OneBendPlanner(RoutingPlanner):
TRACE_INTO_MAX_BENDS = 1
planner = OneBendPlanner()
assert planner.trace_into_bend_bands('straight') == ((0, 0),)
assert planner.trace_into_bend_bands('s') == ((0, 0),)
assert planner.trace_into_bend_bands('bend') == ((1, 1),)