""" 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)