146 lines
4.7 KiB
Python
146 lines
4.7 KiB
Python
from typing import Any
|
|
from collections.abc import Callable
|
|
from copy import deepcopy
|
|
|
|
import numpy
|
|
from numpy.typing import ArrayLike, NDArray
|
|
from numpy.testing import assert_allclose
|
|
|
|
from masque import Pather, Port
|
|
from masque.builder.tools import RenderStep
|
|
|
|
|
|
def closed_edge_lengths(vertices: ArrayLike) -> NDArray[numpy.float64]:
|
|
"""
|
|
Return lengths for each edge of an implicitly closed vertex loop.
|
|
"""
|
|
vv = numpy.asarray(vertices, dtype=float)
|
|
return numpy.sqrt(numpy.sum(numpy.diff(vv, axis=0, append=vv[:1]) ** 2, axis=1))
|
|
|
|
|
|
def assert_closed_edges_within(vertices: ArrayLike, max_len: float, *, atol: float = 1e-6) -> None:
|
|
"""
|
|
Assert that every edge in an implicitly closed vertex loop is no longer than `max_len`.
|
|
"""
|
|
assert numpy.all(closed_edge_lengths(vertices) <= max_len + atol)
|
|
|
|
|
|
def assert_bounds_close(shape_or_polygon: Any, expected: ArrayLike, *, atol: float = 1e-10) -> None:
|
|
"""
|
|
Assert that an object's single-shape bounds match `expected`.
|
|
"""
|
|
assert_allclose(shape_or_polygon.get_bounds_single(), expected, atol=atol)
|
|
|
|
|
|
def normalized_route_data(data: Any) -> Any:
|
|
"""
|
|
Return a deterministic, comparison-friendly representation of route data.
|
|
"""
|
|
if isinstance(data, dict):
|
|
return tuple((key, normalized_route_data(value)) for key, value in sorted(data.items(), key=lambda item: repr(item[0])))
|
|
if isinstance(data, list | tuple):
|
|
return tuple(normalized_route_data(value) for value in data)
|
|
if isinstance(data, numpy.ndarray):
|
|
return tuple(normalized_route_data(value) for value in data.tolist())
|
|
if isinstance(data, numpy.generic):
|
|
return data.item()
|
|
try:
|
|
hash(data)
|
|
except TypeError:
|
|
return repr(data)
|
|
return data
|
|
|
|
|
|
def route_step_signature(step: RenderStep) -> tuple[Any, ...]:
|
|
"""
|
|
Return the stable planning-relevant portion of one rendered route step.
|
|
"""
|
|
return (
|
|
step.opcode,
|
|
tuple(round(float(value), 9) for value in step.start_port.offset),
|
|
None if step.start_port.rotation is None else round(float(step.start_port.rotation), 9),
|
|
step.start_port.ptype,
|
|
tuple(round(float(value), 9) for value in step.end_port.offset),
|
|
None if step.end_port.rotation is None else round(float(step.end_port.rotation), 9),
|
|
step.end_port.ptype,
|
|
normalized_route_data(step.data),
|
|
)
|
|
|
|
|
|
def route_signature(pather: Pather, portspec: str) -> tuple[tuple[Any, ...], ...]:
|
|
"""
|
|
Return a deterministic signature for a pather route.
|
|
"""
|
|
return tuple(route_step_signature(step) for step in pather._paths[portspec])
|
|
|
|
|
|
def route_endpoint(pather: Pather, portspec: str) -> Port:
|
|
"""
|
|
Return the endpoint of a routed port, falling back to the live port for empty routes.
|
|
"""
|
|
steps = pather._paths[portspec]
|
|
if not steps:
|
|
return pather.pattern[portspec]
|
|
return steps[-1].end_port
|
|
|
|
|
|
def assert_route_endpoint(
|
|
pather: Pather,
|
|
portspec: str,
|
|
expected: Port,
|
|
*,
|
|
atol: float = 1e-8,
|
|
) -> None:
|
|
"""
|
|
Assert that a route endpoint matches an expected port pose and ptype.
|
|
"""
|
|
actual = route_endpoint(pather, portspec)
|
|
assert_allclose(actual.offset, expected.offset, atol=atol)
|
|
if expected.rotation is None:
|
|
assert actual.rotation is None
|
|
else:
|
|
assert actual.rotation is not None
|
|
assert numpy.isclose(actual.rotation, expected.rotation, atol=atol)
|
|
assert actual.ptype == expected.ptype
|
|
|
|
|
|
def assert_route_bend_budget(pather: Pather, portspec: str, max_bends: int) -> None:
|
|
"""
|
|
Assert a simple render-step bend budget for route signatures.
|
|
"""
|
|
bend_count = sum(1 for step in pather._paths[portspec] if step.kind == 'bend')
|
|
assert bend_count <= max_bends
|
|
|
|
|
|
def assert_route_deterministic(
|
|
make_pather: Callable[[], Pather],
|
|
route: Callable[[Pather], None],
|
|
portspec: str,
|
|
) -> None:
|
|
"""
|
|
Assert that the same route operation produces the same route signature twice.
|
|
"""
|
|
first = make_pather()
|
|
route(first)
|
|
first_signature = route_signature(first, portspec)
|
|
|
|
second = make_pather()
|
|
route(second)
|
|
assert route_signature(second, portspec) == first_signature
|
|
|
|
|
|
def assert_route_failure_does_not_mutate(
|
|
pather: Pather,
|
|
route: Callable[[], None],
|
|
expected_exception: type[BaseException],
|
|
) -> BaseException:
|
|
"""
|
|
Assert that a failing route operation leaves pending route steps untouched.
|
|
"""
|
|
before = deepcopy(dict(pather._paths))
|
|
try:
|
|
route()
|
|
except expected_exception as err:
|
|
assert dict(pather._paths) == before
|
|
return err
|
|
raise AssertionError(f'Expected {expected_exception.__name__}')
|