File size: 9,943 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 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 | """Typed 0.3 world records: affine trajectories, supports, compatibility, relations."""
from __future__ import annotations
from dataclasses import dataclass
from typing import Any, Mapping, Sequence
from .canonical import require_hash, verify_hash
from .expression import validate_expression
from .interval import ClosedInterval
from .rational import Q
@dataclass(frozen=True, slots=True)
class AffineField:
initial: Q
rate: Q
@classmethod
def from_record(cls, record: Mapping[str, Any]) -> "AffineField":
return cls(Q.from_value(record.get("initial", 0)), Q.from_value(record.get("rate", 0)))
@dataclass(frozen=True, slots=True)
class AffineTrajectory:
id: str
domain: ClosedInterval
fields: Mapping[str, AffineField]
source_program: str | None = None
content_hash: str | None = None
@classmethod
def from_record(cls, record: Mapping[str, Any]) -> "AffineTrajectory":
require_hash(record, label="trajectory")
if record.get("kind") != "affine-rational-trajectory":
raise ValueError("trajectory kind must be affine-rational-trajectory")
fields_record = record.get("fields")
if not isinstance(fields_record, Mapping) or not fields_record:
raise ValueError("trajectory requires a nonempty fields object")
fields = {str(name): AffineField.from_record(value) for name, value in fields_record.items()}
return cls(
id=str(record["id"]),
domain=ClosedInterval.from_value(record["domain"]),
fields=fields,
source_program=None if record.get("source_program") is None else str(record["source_program"]),
content_hash=str(record["content_hash"]),
)
def require_time(self, time: Q) -> None:
if not self.domain.contains(time):
raise ValueError(f"time {time} is outside trajectory domain {self.domain}")
def state_at(self, time: Q) -> dict[str, Q]:
time = Q.from_value(time)
self.require_time(time)
return {name: field.initial + field.rate * time for name, field in self.fields.items()}
def field_rate(self, name: str) -> Q:
try:
return self.fields[name].rate
except KeyError as exc:
raise ValueError(f"trajectory has no field {name}") from exc
def field_affine(self, name: str) -> tuple[Q, Q]:
try:
field = self.fields[name]
except KeyError as exc:
raise ValueError(f"trajectory has no field {name}") from exc
return field.rate, field.initial
def field_interval(self, name: str, time: ClosedInterval) -> ClosedInterval:
intersection = self.domain.intersection(time)
if intersection is None or intersection != time:
raise ValueError("requested interval is outside trajectory domain")
field = self.fields.get(name)
if field is None:
raise ValueError(f"trajectory has no field {name}")
return ClosedInterval.hull([
field.initial + field.rate * time.lower,
field.initial + field.rate * time.upper,
])
def state_interval(self, time: ClosedInterval) -> dict[str, ClosedInterval]:
return {name: self.field_interval(name, time) for name in self.fields}
@dataclass(frozen=True, slots=True)
class Support:
id: str
bounds: Mapping[str, ClosedInterval]
content_hash: str
@classmethod
def from_record(cls, record: Mapping[str, Any]) -> "Support":
require_hash(record, label="support")
if record.get("kind") != "interval-support":
raise ValueError("support kind must be interval-support")
bounds_record = record.get("bounds", {})
if not isinstance(bounds_record, Mapping):
raise ValueError("support bounds must be an object")
return cls(str(record["id"]), {
str(name): ClosedInterval.from_value(value)
for name, value in bounds_record.items()
}, str(record["content_hash"]))
def accepts_point(self, state: Mapping[str, Q]) -> bool:
return all(name in state and bound.contains(state[name]) for name, bound in self.bounds.items())
def contains_interval(self, state: Mapping[str, ClosedInterval]) -> bool:
return all(name in state and state[name].subset_of(bound) for name, bound in self.bounds.items())
@dataclass(frozen=True, slots=True)
class Compatibility:
id: str
equals: Mapping[str, Q]
content_hash: str
@classmethod
def from_record(cls, record: Mapping[str, Any]) -> "Compatibility":
require_hash(record, label="compatibility")
if record.get("kind") != "field-equality-compatibility":
raise ValueError("compatibility kind must be field-equality-compatibility")
equals = record.get("equals", {})
if not isinstance(equals, Mapping):
raise ValueError("compatibility equals must be an object")
return cls(str(record["id"]), {str(k): Q.from_value(v) for k, v in equals.items()},
str(record["content_hash"]))
def accepts(self, state: Mapping[str, Q]) -> bool:
return all(state.get(name) == expected for name, expected in self.equals.items())
@dataclass(frozen=True, slots=True)
class TransitionOp:
field: str
mode: str
value: Q
@classmethod
def from_record(cls, record: Mapping[str, Any]) -> "TransitionOp":
field = str(record.get("field", ""))
mode = str(record.get("mode", ""))
if not field:
raise ValueError("transition operation requires field")
if mode not in {"set", "add", "xor"}:
raise ValueError(f"unsupported transition mode {mode}")
value = Q.from_value(record.get("value", 0))
if mode == "xor" and not value.is_integer():
raise ValueError("xor transition value must be an integer")
return cls(field, mode, value)
def to_record(self) -> dict[str, Any]:
return {"field": self.field, "mode": self.mode, "value": self.value.to_record()}
@dataclass(frozen=True, slots=True)
class Relation:
id: str
priority: int
expression: Mapping[str, Any]
support_id: str
compatibility_id: str
active_time: ClosedInterval
event_id: str
transition: tuple[TransitionOp, ...]
content_hash: str
@classmethod
def from_record(cls, record: Mapping[str, Any]) -> "Relation":
require_hash(record, label="relation")
if record.get("kind") != "continuous-zero-relation":
raise ValueError("relation kind must be continuous-zero-relation")
expression = record.get("expression")
if not isinstance(expression, Mapping):
raise ValueError("relation expression must be an object")
validate_expression(expression)
transition_record = record.get("transition", [])
if not isinstance(transition_record, list):
raise ValueError("relation transition must be an array")
return cls(
id=str(record["id"]),
priority=int(record.get("priority", 0)),
expression=expression,
support_id=str(record["support_id"]),
compatibility_id=str(record["compatibility_id"]),
active_time=ClosedInterval.from_value(record["active_time"]),
event_id=str(record["event_id"]),
transition=tuple(TransitionOp.from_record(item) for item in transition_record),
content_hash=str(record["content_hash"]),
)
@dataclass(frozen=True, slots=True)
class IntervalWorld:
schema: str
seed_sha256: str
trajectory: AffineTrajectory
supports: Mapping[str, Support]
compatibilities: Mapping[str, Compatibility]
relations: tuple[Relation, ...]
solver: Mapping[str, Any]
content_hash: str
@classmethod
def from_record(cls, record: Mapping[str, Any]) -> "IntervalWorld":
require_hash(record, label="interval world")
if record.get("schema") != "TOM-WORLD-INTERVAL-EVENTS-0.3":
raise ValueError("unsupported interval world schema")
trajectory = AffineTrajectory.from_record(record["trajectory"])
supports_list = record.get("supports", [])
compat_list = record.get("compatibilities", [])
relations_list = record.get("relations", [])
if not all(isinstance(x, list) for x in (supports_list, compat_list, relations_list)):
raise ValueError("supports, compatibilities, and relations must be arrays")
supports = {obj.id: obj for obj in map(Support.from_record, supports_list)}
compatibilities = {obj.id: obj for obj in map(Compatibility.from_record, compat_list)}
relations = tuple(map(Relation.from_record, relations_list))
if len(supports) != len(supports_list):
raise ValueError("duplicate support id")
if len(compatibilities) != len(compat_list):
raise ValueError("duplicate compatibility id")
if len({r.id for r in relations}) != len(relations):
raise ValueError("duplicate relation id")
for relation in relations:
if relation.support_id not in supports:
raise ValueError(f"relation {relation.id} references unknown support")
if relation.compatibility_id not in compatibilities:
raise ValueError(f"relation {relation.id} references unknown compatibility")
solver = record.get("solver", {})
if not isinstance(solver, Mapping):
raise ValueError("solver profile must be an object")
return cls(
schema=str(record["schema"]),
seed_sha256=str(record["seed_sha256"]),
trajectory=trajectory,
supports=supports,
compatibilities=compatibilities,
relations=relations,
solver=dict(solver),
content_hash=str(record["content_hash"]),
)
|