File size: 5,397 Bytes
9d6c005 | 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 | """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()
# Duplicate deterministic rays overwrite the same evidence. They do not
# accumulate fictitious statistical precision.
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
|