File size: 9,359 Bytes
32d978d | 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 | """
Object System — Core Knowledge of Objects
Implements Spelke's 4 principles of object perception:
1. Cohesion: Objects are bounded, connected wholes
2. Continuity: Objects trace continuous spatiotemporal paths
3. Contact: Objects don't pass through each other
4. Permanence: Objects persist when occluded
These are innate priors on state transitions, NOT learned from data.
They constrain belief updates during free-energy minimization.
Author: Algorembrant, Rembrant Oyangoren Albeos (2026)
"""
import numpy as np
from typing import Optional
class TrackedObject:
"""A single object tracked by the core object system."""
__slots__ = ['obj_id', 'position', 'velocity', 'size', 'visible',
'occluded_frames', 'confidence', 'last_seen_position']
def __init__(self, obj_id: int, position: np.ndarray, size: float = 1.0):
self.obj_id = obj_id
self.position = np.asarray(position, dtype=np.float64)
self.velocity = np.zeros_like(self.position)
self.size = size
self.visible = True
self.occluded_frames = 0
self.confidence = 1.0
self.last_seen_position = self.position.copy()
class ObjectSystem:
"""
Innate object reasoning system.
Maintains a set of tracked objects and enforces core knowledge
constraints on their state transitions. These are hard priors—
not soft preferences—that cannot be overridden by sensory evidence
alone (just like infants who look longer at "impossible" events).
"""
def __init__(self, max_objects: int = 20, max_occlusion_frames: int = 60):
"""
Args:
max_objects: Maximum number of simultaneously tracked objects.
max_occlusion_frames: How long an occluded object persists in memory
before being garbage-collected.
"""
self.max_objects = max_objects
self.max_occlusion_frames = max_occlusion_frames
self.objects: dict[int, TrackedObject] = {}
self._next_id = 0
def register_object(self, position: np.ndarray, size: float = 1.0) -> int:
"""
Register a newly detected object.
Returns:
Object ID for future reference.
"""
if len(self.objects) >= self.max_objects:
# Evict least confident object
worst_id = min(self.objects, key=lambda k: self.objects[k].confidence)
del self.objects[worst_id]
obj = TrackedObject(self._next_id, position, size)
self.objects[self._next_id] = obj
self._next_id += 1
return obj.obj_id
def update(self, detections: list[dict]) -> list[dict]:
"""
Update object states given new sensory detections.
Enforces all 4 Spelke principles as hard constraints.
Args:
detections: List of dicts with 'position' (ndarray) and 'size' (float).
Returns:
List of violation dicts if any principle is violated (surprise signals).
"""
violations = []
matched_ids = set()
# --- Associate detections with existing objects (continuity) ---
for det in detections:
det_pos = np.asarray(det['position'], dtype=np.float64)
det_size = det.get('size', 1.0)
best_id = None
best_dist = float('inf')
for obj_id, obj in self.objects.items():
if obj_id in matched_ids:
continue
# Predict where object should be (continuity prior)
predicted_pos = obj.position + obj.velocity
dist = np.linalg.norm(det_pos - predicted_pos)
if dist < best_dist:
best_dist = dist
best_id = obj_id
if best_id is not None and best_dist < det_size * 5.0:
obj = self.objects[best_id]
# --- CONTINUITY CHECK ---
displacement = np.linalg.norm(det_pos - obj.position)
if displacement > obj.size * 10.0 and obj.visible:
violations.append({
'type': 'continuity_violation',
'object_id': best_id,
'expected': obj.position + obj.velocity,
'observed': det_pos,
'surprise': displacement / (obj.size * 10.0)
})
# --- COHESION CHECK ---
if abs(det_size - obj.size) / max(obj.size, 0.01) > 0.5:
violations.append({
'type': 'cohesion_violation',
'object_id': best_id,
'expected_size': obj.size,
'observed_size': det_size,
'surprise': abs(det_size - obj.size) / obj.size
})
# Update object state
obj.velocity = det_pos - obj.position
obj.position = det_pos.copy()
obj.size = det_size
obj.visible = True
obj.occluded_frames = 0
obj.confidence = min(1.0, obj.confidence + 0.1)
obj.last_seen_position = det_pos.copy()
matched_ids.add(best_id)
else:
# Detection too far from any prediction — possible teleportation
# Check if there's a visible, unmatched object that might be this one
for obj_id, obj in self.objects.items():
if obj_id not in matched_ids and obj.visible:
displacement = np.linalg.norm(det_pos - obj.position)
if displacement > obj.size * 10.0:
violations.append({
'type': 'continuity_violation',
'object_id': obj_id,
'expected': obj.position + obj.velocity,
'observed': det_pos,
'surprise': displacement / (obj.size * 10.0)
})
# Re-associate the detection with this object
obj.velocity = det_pos - obj.position
obj.position = det_pos.copy()
obj.visible = True
obj.occluded_frames = 0
obj.last_seen_position = det_pos.copy()
matched_ids.add(obj_id)
break
else:
# Genuinely new object
new_id = self.register_object(det_pos, det_size)
matched_ids.add(new_id)
# --- PERMANENCE: Unmatched objects become occluded, NOT deleted ---
for obj_id, obj in list(self.objects.items()):
if obj_id not in matched_ids:
obj.visible = False
obj.occluded_frames += 1
# Continue predicting position (continuity during occlusion)
obj.position = obj.position + obj.velocity
obj.confidence *= 0.95 # Slow decay
# Only garbage-collect after extended occlusion
if obj.occluded_frames > self.max_occlusion_frames:
del self.objects[obj_id]
# --- CONTACT: Check for interpenetration ---
obj_list = list(self.objects.values())
for i in range(len(obj_list)):
for j in range(i + 1, len(obj_list)):
a, b = obj_list[i], obj_list[j]
dist = np.linalg.norm(a.position - b.position)
min_dist = (a.size + b.size) / 2.0
if dist < min_dist:
violations.append({
'type': 'contact_violation',
'object_ids': (a.obj_id, b.obj_id),
'overlap': min_dist - dist,
'surprise': (min_dist - dist) / min_dist
})
return violations
def predict_occluded(self, obj_id: int) -> Optional[np.ndarray]:
"""
Predict where an occluded object should be right now.
This is object permanence: the object still EXISTS even though
it's not visible. Infants (and this system) maintain a belief
about its continued trajectory.
Returns:
Predicted position, or None if object is not tracked.
"""
obj = self.objects.get(obj_id)
if obj is None:
return None
return obj.position.copy()
def get_visible_objects(self) -> list[TrackedObject]:
"""Get all currently visible objects."""
return [o for o in self.objects.values() if o.visible]
def get_all_objects(self) -> list[TrackedObject]:
"""Get all objects including occluded (permanence)."""
return list(self.objects.values())
@property
def num_objects(self) -> int:
"""Total tracked objects (visible + occluded)."""
return len(self.objects)
|