[DO NOT MERGE] [ITEP-82754] POC: Consume Tracker data from MQTT topic by Analytics

controller/src/controller/detections_builder.py

@@ -46,6 +46,10 @@ def prepareObjDict(scene, obj, update_visibility):
     'velocity': velocity.asCartesianVector
   })
 
+  # Add frame count for analytics
+  if hasattr(aobj, 'frameCount'):
+    obj_dict['frame_count'] = aobj.frameCount
+
   rotation = aobj.rotation
   if rotation is not None:
     obj_dict['rotation'] = rotation

controller/src/controller/scene.py

@@ -2,6 +2,7 @@
 # SPDX-License-Identifier: Apache-2.0
 
 import itertools
+from types import SimpleNamespace
 from typing import Optional
 
 import numpy as np
@@ -58,6 +59,9 @@ def __init__(self, name, map_file, scale=None,
     self._trs_xyz_to_lla = None
     self.use_tracker = True
 
+    # Cache for tracked objects from MQTT (for analytics)
+    self.tracked_objects_cache = {}
+
     # FIXME - only for backwards compatibility
     self.scale = scale
 
@@ -294,10 +298,86 @@ def processSensorData(self, jdata, when):
 
     return True
 
+  def updateTrackedObjects(self, detection_type, objects):
+    """
+    Update the cache of tracked objects from MQTT.
+    This is used by Analytics to consume tracked objects published by the Tracker service.
+
+    Args:
+        detection_type: The type of detection (e.g., 'person', 'vehicle')
+        objects: List of tracked objects for this detection type
+    """
+    self.tracked_objects_cache[detection_type] = objects
+    return
+
+  def getTrackedObjects(self, detection_type):
+    """
+    Get tracked objects from cache (MQTT) or fallback to direct tracker call.
+
+    Args:
+        detection_type: The type of detection
+
+    Returns:
+        List of tracked objects (MovingObject instances or serialized dicts)
+    """
+    # First try to get from cache (MQTT-based)
+    if detection_type in self.tracked_objects_cache:
+      cached_objects = self.tracked_objects_cache[detection_type]
+      if cached_objects and isinstance(cached_objects[0], dict):
+        return self._deserializeTrackedObjects(cached_objects)
+      log.debug("Using cached tracked objects for detection type:", detection_type)
+      return cached_objects
+
+    # Fallback to direct tracker call (for backward compatibility)
+    if self.tracker is not None:
+      log.debug("Using direct tracker call for detection type:", detection_type)
+      return self.tracker.currentObjects(detection_type)
+
+    return []
+
+  def _deserializeTrackedObjects(self, serialized_objects):
+    """
+    Convert serialized tracked objects to a format usable by Analytics.
+    This creates lightweight wrappers that mimic MovingObject interface.
+
+    Args:
+        serialized_objects: List of serialized object dictionaries
+
+    Returns:
+        List of object-like structures with necessary attributes
+    """
+    from types import SimpleNamespace
+
+    objects = []
+    for obj_data in serialized_objects:
+      # Create a simple object that has the necessary attributes
+      obj = SimpleNamespace()
+      obj.gid = obj_data.get('id')
+      obj.category = obj_data.get('type')
+      obj.sceneLoc = Point(obj_data.get('translation', [0, 0, 0]))
+      obj.velocity = Point(obj_data.get('velocity', [0, 0, 0])) if obj_data.get('velocity') else None
+      obj.size = obj_data.get('size')
+      obj.confidence = obj_data.get('confidence')
+      obj.frameCount = obj_data.get('frame_count', 0)
+      obj.when = get_epoch_time(obj_data.get('first_seen')) if 'first_seen' in obj_data else get_epoch_time()
+      obj.visibility = obj_data.get('visibility', [])
+
+      # Chain data for regions, sensors, and published locations
+      obj.chain_data = SimpleNamespace()
+      obj.chain_data.regions = obj_data.get('regions', {})
+      obj.chain_data.sensors = obj_data.get('sensors', {})
+      obj.chain_data.persist = obj_data.get('persistent_data', {})
+      # Initialize publishedLocations - will be populated by _updateEvents
+      obj.chain_data.publishedLocations = [obj.sceneLoc]
+
+      objects.append(obj)
+
+    return objects
+
   def _updateEvents(self, detectionType, now):
     self.events = {}
     now_str = get_iso_time(now)
-    curObjects = self.tracker.currentObjects(detectionType)
+    curObjects = self.getTrackedObjects(detectionType)
     for obj in curObjects:
       obj.chain_data.publishedLocations.insert(0, obj.sceneLoc)
 

controller/src/controller/scene_controller.py

@@ -428,6 +428,31 @@ def handleMovingObjectMessage(self, client, userdata, message):
         self.publishEvents(scene, jdata['timestamp'])
       return
 
+  def handleSceneDataMessage(self, client, userdata, message):
+    """
+    Handle scene data messages (tracked objects) published to DATA_SCENE topic.
+    This updates the Analytics cache with tracked objects from the existing topic.
+    """
+    topic = PubSub.parseTopic(message.topic)
+    jdata = orjson.loads(message.payload.decode('utf-8'))
+
+    scene_id = topic['scene_id']
+    detection_type = topic['thing_type']
+
+    scene = self.cache_manager.sceneWithID(scene_id)
+    if scene is None:
+      log.debug("Scene not found for tracked objects, ignoring", scene_id)
+      return
+
+    # Extract tracked objects from the existing DATA_SCENE message
+    tracked_objects = jdata.get('objects', [])
+
+    # Update the analytics cache with tracked objects
+    scene.updateTrackedObjects(detection_type, tracked_objects)
+
+    log.debug(f"Updated tracked objects cache for scene {scene.name}, type {detection_type}, count {len(tracked_objects)}")
+    return
+
   def _handleChildSceneObject(self, sender_id, jdata, detection_type, msg_when):
     sender = self.cache_manager.sceneWithID(sender_id)
     if sender is None:
@@ -593,6 +618,13 @@ def updateSubscriptions(self):
       for sensor in scene.sensors:
         need_subscribe.add((PubSub.formatTopic(PubSub.DATA_SENSOR, sensor_id=sensor),
                             self.handleSensorMessage))
+
+      # Subscribe to scene data (tracked objects) for Analytics to consume
+      # This reuses the existing DATA_SCENE topic that tracker already publishes to
+      need_subscribe.add((PubSub.formatTopic(PubSub.DATA_SCENE,
+                                             scene_id=scene.uid, thing_type="+"),
+                          self.handleSceneDataMessage))
+
       if hasattr(scene, 'children'):
         child_scenes = self.cache_manager.data_source.getChildScenes(scene.uid)