hippocampaif/core_knowledge/object_system.py

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