| """Canonical finite transport memory: evidence is a value, not a frame.""" |
| import numpy as np |
|
|
|
|
| class WorldMemory: |
| """Fixed-capacity receiver x emitter visibility table for the reference. |
| |
| A scene namespace owns stable receiver and emitter IDs. Reusing an integer |
| ID for a different surface requires a new namespace/reset. Each entry is |
| deterministic visibility evidence at an exact point/emitter, not an |
| independent noisy training sample. Geometry epochs revoke trust while |
| retaining the old value as a fallible control variate. |
| """ |
| def __init__(self, receivers, terms, namespace="scene"): |
| if receivers < 1 or terms < 1 or not namespace: |
| raise ValueError("Need a nonempty namespace and positive capacity") |
| self.namespace = str(namespace) |
| self.values = np.full((receivers, terms), np.nan, np.float32) |
| self.epochs = np.full((receivers, terms), -1, np.int32) |
| self.epoch = 0 |
| self.tick = 0 |
|
|
| @property |
| def nbytes(self): |
| return self.values.nbytes + self.epochs.nbytes |
|
|
| def _receiver_ids(self, receivers): |
| rows = np.asarray(receivers) |
| if rows.size == 0 and rows.ndim == 1: |
| return rows.astype(np.int64) |
| if (rows.ndim != 1 or not np.issubdtype(rows.dtype, np.integer) |
| or (rows < 0).any() or (rows >= len(self.values)).any()): |
| raise ValueError("Canonical receiver IDs must be valid nonnegative integers") |
| return rows |
|
|
| def predict(self, receivers, prior): |
| v = self.values[self._receiver_ids(receivers)] |
| prior = np.asarray(prior, float) |
| if prior.shape != v.shape or not np.isfinite(prior).all() or ((prior < 0)|(prior > 1)).any(): |
| raise ValueError("Visibility prior must have matching shape in [0,1]") |
| return np.where(np.isnan(v), prior, v).astype(np.float64) |
|
|
| def trusted(self, receivers): |
| return self.epochs[self._receiver_ids(receivers)] == self.epoch |
|
|
| def commit(self, receivers, indices, visibility, revise_on_conflict=False): |
| rows = np.asarray(receivers) |
| j = np.asarray(indices) |
| v = np.asarray(visibility) |
| if (rows.ndim != 1 or j.ndim != 2 or len(rows) != len(j) or j.shape != v.shape |
| or not np.issubdtype(rows.dtype, np.integer) or not np.issubdtype(j.dtype, np.integer) |
| or (rows < 0).any() or (rows >= len(self.values)).any() |
| or (j < 0).any() or (j >= self.values.shape[1]).any() |
| or not np.isfinite(v).all() or ((v != 0)&(v != 1)).any()): |
| raise ValueError("Expected valid exact binary visibility observations") |
| old = self.values[rows[:, None], j] |
| was_trusted = self.epochs[rows[:, None], j] == self.epoch |
| conflicts = int(np.count_nonzero(was_trusted & np.isfinite(old) & (old != v))) |
| if revise_on_conflict and conflicts: |
| self.notify_geometry_change() |
| |
| |
| self.values[rows[:, None], j] = v |
| self.epochs[rows[:, None], j] = self.epoch |
| return conflicts |
|
|
| def notify_geometry_change(self): |
| if self.epoch == np.iinfo(np.int32).max: |
| self.epochs.fill(-1) |
| self.values.fill(np.nan) |
| self.epoch = 0 |
| else: |
| self.epoch += 1 |
|
|
| def retain_only(self, receivers): |
| keep = np.zeros(len(self.values), bool) |
| keep[self._receiver_ids(receivers)] = True |
| self.values[~keep] = np.nan |
| self.epochs[~keep] = -1 |
|
|
| def advance(self, ticks=1): |
| if not isinstance(ticks, (int, np.integer)) or ticks < 0: |
| raise ValueError("ticks must be a nonnegative integer") |
| self.tick += int(ticks) |
|
|
| def save(self, path): |
| np.savez_compressed(path, values=self.values, epochs=self.epochs, |
| epoch=np.int64(self.epoch), tick=np.int64(self.tick), |
| namespace=np.array(self.namespace)) |
|
|
| @classmethod |
| def load(cls, path, namespace): |
| with np.load(path, allow_pickle=False) as data: |
| if str(data["namespace"]) != str(namespace): |
| raise ValueError("Scene namespace mismatch: refusing stale identity alias") |
| values = np.array(data["values"], copy=True) |
| epochs = np.array(data["epochs"], copy=True) |
| if values.ndim != 2 or epochs.shape != values.shape or values.dtype != np.float32 or epochs.dtype != np.int32: |
| raise ValueError("Malformed memory arrays") |
| finite = values[np.isfinite(values)] |
| if np.isinf(values).any() or ((finite != 0)&(finite != 1)).any(): |
| raise ValueError("Memory must contain binary observations or NaN") |
| epoch, tick = int(data["epoch"]), int(data["tick"]) |
| if epoch < 0 or epoch > np.iinfo(np.int32).max or tick < 0 or (epochs < -1).any() or (epochs > epoch).any(): |
| raise ValueError("Malformed memory time/epoch") |
| if (np.isnan(values) != (epochs == -1)).any(): |
| raise ValueError("Missing evidence and provenance disagree") |
| obj = cls(*values.shape, namespace=namespace) |
| obj.values, obj.epochs, obj.epoch, obj.tick = values, epochs, epoch, tick |
| return obj |
|
|
