File size: 5,132 Bytes
767a051 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 | """Typed continuous expression evaluation over affine TOM trajectories."""
from __future__ import annotations
from dataclasses import dataclass
from typing import Any, Mapping, Protocol
from .interval import ClosedInterval
from .rational import Q
class TrajectoryLike(Protocol):
def state_at(self, time: Q) -> Mapping[str, Q]: ...
def field_interval(self, name: str, time: ClosedInterval) -> ClosedInterval: ...
def field_rate(self, name: str) -> Q: ...
def field_affine(self, name: str) -> tuple[Q, Q]: ...
@dataclass(frozen=True, slots=True)
class DualInterval:
value: ClosedInterval
derivative: ClosedInterval
def _op(expr: Mapping[str, Any]) -> str:
value = expr.get("op")
if not isinstance(value, str):
raise ValueError("expression requires string op")
return value
def validate_expression(expr: Any, *, depth: int = 0, max_depth: int = 64) -> None:
if depth > max_depth:
raise ValueError("expression depth exceeds profile limit")
if not isinstance(expr, Mapping):
raise TypeError("relation expression nodes must be objects")
op = _op(expr)
if op == "const":
Q.from_value(expr.get("value"))
return
if op == "field":
if not isinstance(expr.get("name"), str) or not expr["name"]:
raise ValueError("field expression requires nonempty name")
return
if op == "time":
return
if op == "neg":
validate_expression(expr.get("value"), depth=depth + 1, max_depth=max_depth)
return
if op in {"add", "sub", "mul"}:
args = expr.get("args")
if not isinstance(args, list) or len(args) != 2:
raise ValueError(f"{op} expression requires exactly two args")
validate_expression(args[0], depth=depth + 1, max_depth=max_depth)
validate_expression(args[1], depth=depth + 1, max_depth=max_depth)
return
raise ValueError(f"unsupported continuous expression op {op}")
def evaluate_point(expr: Mapping[str, Any], trajectory: TrajectoryLike, time: Q) -> Q:
op = _op(expr)
if op == "const":
return Q.from_value(expr["value"])
if op == "time":
return time
if op == "field":
state = trajectory.state_at(time)
name = str(expr["name"])
if name not in state:
raise ValueError(f"trajectory has no field {name}")
return Q.from_value(state[name])
if op == "neg":
return -evaluate_point(expr["value"], trajectory, time)
a = evaluate_point(expr["args"][0], trajectory, time)
b = evaluate_point(expr["args"][1], trajectory, time)
if op == "add":
return a + b
if op == "sub":
return a - b
if op == "mul":
return a * b
raise ValueError(f"unsupported op {op}")
def evaluate_dual_interval(
expr: Mapping[str, Any],
trajectory: TrajectoryLike,
time: ClosedInterval,
) -> DualInterval:
op = _op(expr)
if op == "const":
return DualInterval(ClosedInterval.point(expr["value"]), ClosedInterval.point(0))
if op == "time":
return DualInterval(time, ClosedInterval.point(1))
if op == "field":
name = str(expr["name"])
return DualInterval(
trajectory.field_interval(name, time),
ClosedInterval.point(trajectory.field_rate(name)),
)
if op == "neg":
a = evaluate_dual_interval(expr["value"], trajectory, time)
return DualInterval(-a.value, -a.derivative)
a = evaluate_dual_interval(expr["args"][0], trajectory, time)
b = evaluate_dual_interval(expr["args"][1], trajectory, time)
if op == "add":
return DualInterval(a.value + b.value, a.derivative + b.derivative)
if op == "sub":
return DualInterval(a.value - b.value, a.derivative - b.derivative)
if op == "mul":
return DualInterval(
a.value * b.value,
a.derivative * b.value + a.value * b.derivative,
)
raise ValueError(f"unsupported op {op}")
def affine_coefficients(
expr: Mapping[str, Any], trajectory: TrajectoryLike
) -> tuple[Q, Q] | None:
"""Return (slope, intercept) when expr is affine in global time."""
op = _op(expr)
if op == "const":
return Q(0), Q.from_value(expr["value"])
if op == "time":
return Q(1), Q(0)
if op == "field":
return trajectory.field_affine(str(expr["name"]))
if op == "neg":
inner = affine_coefficients(expr["value"], trajectory)
return None if inner is None else (-inner[0], -inner[1])
left = affine_coefficients(expr["args"][0], trajectory)
right = affine_coefficients(expr["args"][1], trajectory)
if left is None or right is None:
return None
if op == "add":
return left[0] + right[0], left[1] + right[1]
if op == "sub":
return left[0] - right[0], left[1] - right[1]
if op == "mul":
if left[0] == Q(0):
return right[0] * left[1], right[1] * left[1]
if right[0] == Q(0):
return left[0] * right[1], left[1] * right[1]
return None
raise ValueError(f"unsupported op {op}")
|