scene_controller.py

# SPDX-FileCopyrightText: (C) 2021 - 2026 Intel Corporation
# SPDX-License-Identifier: Apache-2.0

import orjson
import os
from collections import defaultdict

import ntplib

from controller.cache_manager import CacheManager
from controller.child_scene_controller import ChildSceneController
from controller.controller_mode import ControllerMode
from controller.detections_builder import (buildDetectionsDict,
                                           buildDetectionsList,
                                           computeCameraBounds)
from controller.scene import Scene
from scene_common import log
from scene_common.geometry import Point, Region, Tripwire
from scene_common.mqtt import PubSub
from scene_common.schema import SchemaValidation
from scene_common.timestamp import adjust_time, get_epoch_time, get_iso_time
from scene_common.transform import applyChildTransform
from controller.observability import metrics
from controller.time_chunking import (DEFAULT_CHUNKING_RATE_FPS,
                                      MINIMAL_CHUNKING_RATE_FPS,
                                      MAXIMAL_CHUNKING_RATE_FPS)
from controller.tracking import EFFECTIVE_OBJECT_UPDATE_RATE, DEFAULT_SUSPENDED_TRACK_TIMEOUT_SECS
AVG_FRAMES = 100

class SceneController:

  def __init__(self, rewrite_bad_time, rewrite_all_time, max_lag, mqtt_broker,
               mqtt_auth, rest_url, rest_auth, client_cert, root_cert, ntp_server,
               tracker_config_file, schema_file, visibility_topic, data_source, reid_config_file=None):
    self.cert = client_cert
    self.root_cert = root_cert
    self.rewrite_bad_time = rewrite_bad_time
    self.rewrite_all_time = rewrite_all_time
    self.max_lag = max_lag
    self.regulate_cache = {}
    self.broker = mqtt_broker
    self.mqtt_auth = mqtt_auth
    self.tracker_config_data = {}
    self.tracker_config_file = tracker_config_file
    self.reid_config_data = {}
    self.reid_config_file = reid_config_file

    if tracker_config_file is not None and not ControllerMode.isAnalyticsOnly():
      self.extractTrackerConfigData(tracker_config_file)
    elif ControllerMode.isAnalyticsOnly():
      log.info("Analytics-only mode: Skipping tracker configuration file loading")

    if reid_config_file is not None and not ControllerMode.isAnalyticsOnly():
      self.extractReidConfigData(reid_config_file)
    elif ControllerMode.isAnalyticsOnly():
      log.info("Analytics-only mode: Skipping reid configuration file loading")

    self.last_time_sync = None
    self.ntp_server = ntp_server
    self.ntp_client = ntplib.NTPClient()
    self.time_offset = 0

    self.schema_val = SchemaValidation(schema_file, is_multi_message=True)

    # Initialize scene-data schema validator for analytics-only mode
    self.scene_data_schema_validator = None
    if ControllerMode.isAnalyticsOnly():
      from pathlib import Path
      schema_filename = 'scene-data.schema.json'
      schema_path = Path(os.environ.get('SCENESCAPE_HOME')) / 'tracker' / 'schema' / schema_filename
      if schema_path.exists():
        try:
          log.info(f"Loading scene-data schema from: {schema_path}")
          self.scene_data_schema_validator = SchemaValidation(str(schema_path), is_multi_message=False)
          log.info("Scene-data schema validator initialized")
        except Exception as e:
          log.error(f"Failed to initialize scene-data schema validator from {schema_path}: {e}")
      else:
        log.error(f"Scene-data schema file not found at: {schema_path}")

    self.pubsub = PubSub(mqtt_auth, client_cert, root_cert, mqtt_broker, keepalive=60)
    self.pubsub.onConnect = self.onConnect
    self.pubsub.connect()

    self.cache_manager = CacheManager(data_source, rest_url, rest_auth, root_cert, self.tracker_config_data, self.reid_config_data)

    self.visibility_topic = visibility_topic
    log.info(f"Publishing camera visibility info on {self.visibility_topic} topic.")
    return

  def extractTrackerConfigData(self, tracker_config_file):
    if not os.path.exists(tracker_config_file) and not os.path.isabs(tracker_config_file):
      script = os.path.realpath(__file__)
      tracker_config_file = os.path.join(os.path.dirname(script), tracker_config_file)
    with open(tracker_config_file) as json_file:
      tracker_config = orjson.loads(json_file.read())
      self.tracker_config_data["max_unreliable_time"] = tracker_config["max_unreliable_time_s"]
      self.tracker_config_data["non_measurement_time_dynamic"] = tracker_config["non_measurement_time_dynamic_s"]
      self.tracker_config_data["non_measurement_time_static"] = tracker_config["non_measurement_time_static_s"]
      self.tracker_config_data["effective_object_update_rate"] = self._extractTrackerRate(tracker_config, "effective_object_update_rate", EFFECTIVE_OBJECT_UPDATE_RATE)
      self._extractTimeChunkingEnabled(tracker_config)
      self.tracker_config_data["time_chunking_rate_fps"] = self._extractTrackerRate(tracker_config, "time_chunking_rate_fps", DEFAULT_CHUNKING_RATE_FPS, MINIMAL_CHUNKING_RATE_FPS, MAXIMAL_CHUNKING_RATE_FPS)
      self.tracker_config_data["suspended_track_timeout_secs"] = tracker_config.get("suspended_track_timeout_secs", DEFAULT_SUSPENDED_TRACK_TIMEOUT_SECS)

      if "persist_attributes" in tracker_config:
        if isinstance(tracker_config["persist_attributes"], dict):
          self.tracker_config_data["persist_attributes"] = tracker_config["persist_attributes"]
        else:
          log.error("Invalid persist_attributes format in tracker config file")
          self.tracker_config_data["persist_attributes"] = {}
    return

  def extractReidConfigData(self, reid_config_file):
    """Extract REID configuration from reid-config.json file"""
    if not os.path.exists(reid_config_file) and not os.path.isabs(reid_config_file):
      script = os.path.realpath(__file__)
      reid_config_file = os.path.join(os.path.dirname(script), reid_config_file)
    with open(reid_config_file) as json_file:
      reid_config = orjson.loads(json_file.read())
      self.reid_config_data = reid_config
      log.info(f"Loaded REID configuration from {reid_config_file}: {self.reid_config_data}")
    return

  def _extractTrackerRate(self, tracker_config, parameter_name, default_rate, min_rate=None, max_rate=None):
    """Extract and validate rate parameter from tracker config."""

    if parameter_name not in tracker_config:
      log.warning(f"{parameter_name} not specified in tracker configuration, will use default rate of {default_rate} fps.")
      return default_rate

    try:
      rate_fps = int(tracker_config[parameter_name])
      if rate_fps <= 0:
        raise ValueError(f"{parameter_name} must be a positive integer.")
      if min_rate is not None and rate_fps < min_rate:
        raise ValueError(f"{parameter_name} must be at least {min_rate}.")
      if max_rate is not None and rate_fps > max_rate:
        raise ValueError(f"{parameter_name} must be at most {max_rate}.")
      log.info(f"{parameter_name}: {rate_fps}")
      return rate_fps
    except (ValueError, TypeError) as e:
      raise ValueError(f"Invalid value for {parameter_name} in tracker configuration") from e

  def _extractTimeChunkingEnabled(self, tracker_config):
    """Extract and validate time_chunking_enabled flag"""
    if "time_chunking_enabled" not in tracker_config:
      log.warning("Time chunking enabled flag missing in tracker config file, disabling time chunking.")
      self.tracker_config_data["time_chunking_enabled"] = False
      return

    try:
      self.tracker_config_data["time_chunking_enabled"] = bool(tracker_config["time_chunking_enabled"])
      log.info(f"Time chunking enabled: {self.tracker_config_data['time_chunking_enabled']}")
    except (ValueError, TypeError):
      raise ValueError("Invalid value for time_chunking_enabled in tracker config file.")
    return

  def loopForever(self):
    return self.pubsub.loopForever()

  def publishDetections(self, scene, objects, ts, otype, jdata, camera_id):
    if not hasattr(scene, 'lastPubCount'):
      scene.lastPubCount = {}

    if not hasattr(scene, 'last_published_detection'):
      scene.last_published_detection = defaultdict(lambda: None)
    metric_attributes = {
      "camera": camera_id if camera_id is not None else "unknown",
      "category": otype,
      "scene": scene.name
    }
    metrics.record_object_count(len(objects), metric_attributes)

    if not ControllerMode.isAnalyticsOnly():
      self.publishSceneDetections(scene, objects, otype, jdata)
    self.publishRegulatedDetections(scene, objects, otype, jdata, camera_id)
    self.publishRegionDetections(scene, objects, otype, jdata)
    return

  def shouldPublish(self, last, now, max_delay):
    return last is None or now - last >= max_delay

  def publishSceneDetections(self, scene, objects, otype, jdata):
    # Full rate output (30fps): exclude sensor data for performance
    jdata['objects'] = buildDetectionsList(objects, scene, self.visibility_topic == 'unregulated', include_sensors=False)
    olen = len(jdata['objects'])
    cid = scene.name + "/" + otype
    if olen > 0 or cid not in scene.lastPubCount or scene.lastPubCount[cid] > 0:
      if 'debug_hmo_start_time' in jdata:
        jdata['debug_hmo_processing_time'] = get_epoch_time() - jdata['debug_hmo_start_time']
      # Convert numpy types to native Python types for JSON serialization
      jstr = orjson.dumps(jdata, option=orjson.OPT_SERIALIZE_NUMPY)
      new_topic = PubSub.formatTopic(PubSub.DATA_SCENE, scene_id=scene.uid,
                                     thing_type=otype)
      self.pubsub.publish(new_topic, jstr)
      # External detections need sensor data, so pass objects to rebuild
      self.publishExternalDetections(scene, otype, objects, jdata)
      scene.lastPubCount[cid] = olen
    return

  def publishExternalDetections(self, scene, otype, objects, jdata_base):
    # External rate output (0.5fps): include sensor data
    now = get_epoch_time()
    if self.shouldPublish(scene.last_published_detection[otype], now, 1/scene.external_update_rate):
      scene.last_published_detection[otype] = get_epoch_time()

      # Rebuild detections list with sensor data included
      jdata = jdata_base.copy()
      jdata['objects'] = buildDetectionsList(objects, scene, self.visibility_topic == 'unregulated', include_sensors=True)
      jstr = orjson.dumps(jdata, option=orjson.OPT_SERIALIZE_NUMPY)

      scene_hierarchy_topic = PubSub.formatTopic(PubSub.DATA_EXTERNAL, scene_id=scene.uid,
                                                 thing_type=otype)
      self.pubsub.publish(scene_hierarchy_topic, jstr)
    return

  def publishRegulatedDetections(self, scene_obj, msg_objects, otype, jdata, camera_id):
    update_rate = self.calculateRate()
    scene_uid = scene_obj.uid

    if scene_uid not in self.regulate_cache:
      self.regulate_cache[scene_uid] = {
        'objects': {},
        'rate': {},
        'last': None
      }
    scene = self.regulate_cache[scene_uid]
    # Regulated rate output (5fps): include sensor data
    scene['objects'][otype] = buildDetectionsList(msg_objects, scene_obj, self.visibility_topic == 'unregulated', include_sensors=True)
    if camera_id is not None:
      scene['rate'][camera_id] = jdata.get('rate', None)
    elif ControllerMode.isAnalyticsOnly() and 'rate' in jdata:
      camera_ids = set()
      for obj in jdata.get('objects', []):
        camera_ids.update(obj.get('visibility', []))

      scene_rate = jdata['rate']
      configured_cameras = set(scene_obj.cameras.keys())
      for cam_id in camera_ids:
        if cam_id in configured_cameras:
          scene['rate'][cam_id] = scene_rate

    now = get_epoch_time()
    if self.shouldPublish(scene['last'], now, 1/scene_obj.regulated_rate):
      # If we're doing Regulated visibility, then we need to compute for all
      # the objects in the cache
      objects = []
      is_regulated = self.visibility_topic == 'regulated'

      msg_objects_lookup = {}
      if is_regulated and not ControllerMode.isAnalyticsOnly():
        for obj in msg_objects:
          msg_objects_lookup[obj.gid] = obj

      for key in scene['objects']:
        for obj in scene['objects'][key]:
          if is_regulated and not ControllerMode.isAnalyticsOnly():
            aobj = msg_objects_lookup.get(obj['id'], None)
            if aobj is not None:
              computeCameraBounds(scene_obj, aobj, obj)
          objects.append(obj)
      log.debug(f"Publishing regulated: scene={scene_uid}, objects_count={len(objects)}, types={list(scene['objects'].keys())}")
      new_jdata = {
        'timestamp': jdata['timestamp'],
        'objects': objects,
        'id': jdata['id'],
        'name': jdata['name'],
        'scene_rate': round(1 / update_rate, 1),
        'rate': scene['rate'],
      }
      jstr = orjson.dumps(new_jdata, option=orjson.OPT_SERIALIZE_NUMPY)
      topic = PubSub.formatTopic(PubSub.DATA_REGULATED, scene_id=scene_uid)
      self.pubsub.publish(topic, jstr)
      scene['last'] = now

    return

  def publishRegionDetections(self, scene, objects, otype, jdata):
    for rname in scene.regions:
      robjects = []
      for obj in objects:
        if rname in obj.chain_data.regions:
          robjects.append(obj)
      # Region-specific detections: include sensor data
      jdata['objects'] = buildDetectionsList(
        robjects, scene, False, include_sensors=True,
        include_region_dwell=True, current_time=get_epoch_time(jdata['timestamp']))
      olen = len(jdata['objects'])
      rid = scene.name + "/" + rname + "/" + otype
      if olen > 0 or rid not in scene.lastPubCount or scene.lastPubCount[rid] > 0:
        jstr = orjson.dumps(jdata, option=orjson.OPT_SERIALIZE_NUMPY)
        new_topic = PubSub.formatTopic(PubSub.DATA_REGION, scene_id=scene.uid,
                                       region_id=rname, thing_type=otype)
        self.pubsub.publish(new_topic, jstr)
        scene.lastPubCount[rid] = olen
    return

  def publishEvents(self, scene, ts_str):
    for event_type in scene.events:
      for _, region in scene.events[event_type]:
        etype = None
        metadata = None

        if isinstance(region, Tripwire):
          etype = 'tripwire'
          metadata = region.serialize()

        elif isinstance(region, Region):
          etype = 'region'
          metadata = region.serialize()
          metadata['fromSensor'] = (region.singleton_type != None)

        event_data = {
          'timestamp': ts_str,
          'scene_id': scene.uid,
          'scene_name': scene.name,
          etype + '_id': region.uuid,
          etype + '_name': region.name,
        }
        detections_dict, num_objects = self._buildAllRegionObjsList(scene, region, event_data)
        self._buildEnteredObjsList(scene, region, event_data, detections_dict)
        self._buildExitedObjsList(scene, region, event_data)

        log.debug("EVENT DATA", event_data)
        if hasattr(region, 'value'):
          event_data['value'] = region.value
        event_data['metadata'] = metadata
        if not isinstance(region, Tripwire) or num_objects > 0:
          event_topic = PubSub.formatTopic(PubSub.EVENT,
                                           region_type=etype, event_type=event_type,
                                           scene_id=scene.uid, region_id=region.uuid)
          self.pubsub.publish(event_topic, orjson.dumps(event_data, option=orjson.OPT_SERIALIZE_NUMPY))

    # Clear objects and count events after publishing (but preserve 'value' events for sensors)
    scene.events.pop('objects', None)
    scene.events.pop('count', None)

    self._clearSensorValuesOnExit(scene)

    return

  def _buildAllRegionObjsList(self, scene, region, event_data):
    counts = {}
    num_objects = 0
    all_objects = []
    for otype, objects in region.objects.items():
      counts[otype] = len(objects)
      num_objects += counts[otype]
      all_objects += objects
    event_data['counts'] = counts
    detections_dict = buildDetectionsDict(
      all_objects, scene, include_sensors=True,
      include_region_dwell=True, current_time=get_epoch_time(event_data['timestamp']))
    event_data['objects'] = list(detections_dict.values())
    return detections_dict, num_objects

  def _buildEnteredObjsList(self, scene, region, event_data, detections_dict):
    entered = getattr(region, 'entered', {})
    event_data['entered'] = []
    missing_objs = []
    for entered_list in entered.values():
      for item in entered_list:
        # For sensor value events, objects may not be in detections_dict
        if item.gid in detections_dict:
          event_data['entered'].append(detections_dict[item.gid])
        else:
          missing_objs.append(item)

    # Build any objects not in detections_dict (e.g., from sensor events)
    if missing_objs:
      entered_objs = buildDetectionsList(
        missing_objs, scene, False, include_sensors=True,
        include_region_dwell=True, current_time=get_epoch_time(event_data['timestamp']))
      event_data['entered'].extend(entered_objs)

  def _buildExitedObjsList(self, scene, region, event_data):
    exited = getattr(region, 'exited', {})
    event_data['exited'] = []
    exited_dict = {}
    for exited_list in exited.values():
      exited_objs = []
      for exited_obj, dwell in exited_list:
        exited_dict[exited_obj.gid] = dwell
        exited_objs.extend([exited_obj])
      # Exit events: include sensor data (timestamped readings and attribute events)
      exited_objs = buildDetectionsList(
        exited_objs, scene, False, include_sensors=True,
        include_region_dwell=True, current_time=get_epoch_time(event_data['timestamp']))
      exited_data = [{'object': exited_obj, 'dwell': exited_dict[exited_obj['id']]} for exited_obj in exited_objs]
      event_data['exited'].extend(exited_data)
    return

  def _clearSensorValuesOnExit(self, scene):
    """
    Clears region entered/exited arrays after events have been published.
    Note: Sensor state cleanup (readings arrays, etc.) is handled
    in _updateRegionEvents before this method is called. This method only clears
    the event arrays to prevent stale data from being published in subsequent frames.
    """
    for event_type in scene.events:
      for region_name, region in scene.events[event_type]:
        region.exited = {}
        region.entered = {}
    return

  # Message handling
  def handleSensorMessage(self, client, userdata, message):
    """
    Handle a sensor message such as this
    MQTT Topic: scenescape/data/sensor/02:42:ac:11:00:05.1
        {"timestamp": "2018-09-12T19:03:49.600z",
         "subtype": "humidity",
         "value": "21.7",
         "id": "02:42:ac:11:00:05.1",
         "status": "green" }
    """

    message = message.payload.decode('utf-8')
    jdata = orjson.loads(message)

    if not self.schema_val.validateMessage("singleton", jdata, check_format=True):
      return

    sensor_id = jdata['id']
    scene = self.cache_manager.sceneWithSensorID(sensor_id)
    if scene is None:
      return

    if self.rewrite_all_time:
      ts = get_epoch_time()
      jdata['timestamp'] = get_iso_time(ts)
    else:
      ts = get_epoch_time(jdata['timestamp'])

    if not scene.processSensorData(jdata, when=ts):
      log.error("Sensor fail", sensor_id)
      self.cache_manager.invalidate()
      return

    jdata['scene_id'] = scene.uid
    jdata['scene_name'] = scene.name

    self.publishEvents(scene, jdata['timestamp'])
    return

  def handleMovingObjectMessage(self, client, userdata, message):

    topic = PubSub.parseTopic(message.topic)
    jdata = orjson.loads(message.payload.decode('utf-8'))

    metric_attributes = {
        "topic": message.topic,
        "camera": jdata.get("id", "unknown"),
    }
    metrics.inc_messages(metric_attributes)
    with metrics.time_mqtt_handler(metric_attributes):
      if 'camera_id' in topic and not self.schema_val.validateMessage("detector", jdata):
        return

      now = get_epoch_time()
      self.time_offset, self.last_time_sync = adjust_time(now, self.ntp_server, self.ntp_client,
                                                      self.last_time_sync, self.time_offset,
                                                      ntplib.NTPException)
      now += self.time_offset
      if 'updatecamera' in jdata:
        return

      jdata['debug_hmo_start_time'] = now
      self.cache_manager.refreshScenesForCamParams(jdata)

      if self.rewrite_all_time:
        msg_when = now
        jdata['timestamp'] = get_iso_time(now)
      else:
        msg_when = get_epoch_time(jdata['timestamp'])

      lag = abs(now - msg_when)
      if lag > self.max_lag:
        if not self.rewrite_bad_time:
          metric_attributes["reason"] = "fell_behind"
          metrics.inc_dropped(metric_attributes)
          log.warning("{} FELL BEHIND by {}. SKIPPING {}".format(message.topic, lag, jdata['id']))
          return
        msg_when = now

      camera_id = None
      if topic['_topic_id'] == PubSub.DATA_EXTERNAL:
        detection_types = [topic['thing_type']]
        sender_id = topic['scene_id']
        success, scene = self._handleChildSceneObject(sender_id, jdata, detection_types[0], msg_when)
      else:
        detection_types = jdata['objects'].keys()
        camera_id = sender_id = topic['camera_id']
        sender = self.cache_manager.sceneWithCameraID(sender_id)
        if sender is None:
          log.error("UNKNOWN SENDER", sender_id)
          return
        scene = sender

        # If no detection types in the message, add empty arrays for all tracked types
        # This must be done BEFORE processCameraData so the tracker processes them
        if not detection_types:
          detection_types = list(scene.tracker.trackers.keys())
          for dtype in detection_types:
            jdata['objects'][dtype] = []

        success = scene.processCameraData(jdata, when=msg_when)

      if not success:
        log.error("Camera fail", sender_id, scene.name)
        self.cache_manager.invalidate()
        return

      jdata['id'] = scene.uid
      jdata['name'] = scene.name
      for detection_type in detection_types:
        jdata['unique_detection_count'] = scene.tracker.getUniqueIDCount(detection_type)
        self.publishDetections(scene, scene.tracker.currentObjects(detection_type),
                              msg_when, detection_type, jdata, camera_id)
        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.
    When analytics-only mode is enabled, this also publishes analytics results.
    """
    topic = PubSub.parseTopic(message.topic)
    jdata = orjson.loads(message.payload.decode('utf-8'))

    scene_id = topic['scene_id']
    detection_type = topic['thing_type']
    log.debug(f"Received scene data message: scene={scene_id}, type={detection_type}, objects={len(jdata.get('objects', []))}")

    scene = self.cache_manager.sceneWithID(scene_id)
    if scene is None:
      log.warning(f"Scene not found for tracked objects, ignoring scene_id={scene_id}")
      return

    if ControllerMode.isAnalyticsOnly() and self.scene_data_schema_validator is not None:
      if not self.scene_data_schema_validator.validate(jdata, check_format=True):
        log.error(f"Scene data validation failed for scene={scene_id}, type={detection_type}")
        return

    tracked_objects = jdata.get('objects', [])

    scene.updateTrackedObjects(detection_type, tracked_objects)

    if ControllerMode.isAnalyticsOnly():
      analytics_objects = scene.getTrackedObjects(detection_type)
      log.debug(f"Analytics-only mode - received objects: scene={scene_id}, type={detection_type}, count={len(analytics_objects)}")

      msg_when = get_epoch_time(jdata.get('timestamp'))

      scene._updateEvents(detection_type, msg_when, analytics_objects)

      self.publishDetections(scene, analytics_objects, msg_when, detection_type, jdata, None)
      self.publishEvents(scene, jdata.get('timestamp'))

    return

  def _handleChildSceneObject(self, sender_id, jdata, detection_type, msg_when):
    sender = self.cache_manager.sceneWithID(sender_id)
    if sender is None:
      remote_sender = self.cache_manager.sceneWithRemoteChildID(sender_id)
      if remote_sender is None:
        log.error("UNKNOWN SENDER")
        return
      else:
        sender = remote_sender

    if not hasattr(sender, 'parent') or sender.parent is None:
      log.error("UNKNOWN PARENT", sender_id)
      return False, sender

    scene = self.cache_manager.sceneWithID(sender.parent)
    success = scene.processSceneData(jdata, sender, sender.cameraPose,
                                     detection_type, when=msg_when)
    return success, scene

  def updateCameras(self):
    for scene in self.scenes:
      for camera in scene.cameras:
        cam = scene.cameras[camera]
        if not hasattr(cam, "pose"):
          self.cache_manager.updateCamera(cam)
    return

  def updateRegulateCache(self):
    for scene in list(self.regulate_cache.keys()):
      if scene not in self.scenes:
        self.regulate_cache.pop(scene)
      else:
        for cam in scene['rate']:
          if cam not in scene.cameras:
            scene['rate'].pop(cam)
    return

  def handleDatabaseMessage(self, client, userdata, message):
    command = str(message.payload.decode("utf-8"))
    if command == "update":
      try:
        self.updateSubscriptions()
        self.updateObjectClasses()
        self.updateCameras()
        self.updateRegulateCache()
        self.updateTRSMatrix()
      except Exception as e:
        log.warning("Failed to update database: %s", e)
    return

  def calculateRate(self):
    now = get_epoch_time()
    if not hasattr(self, "regulate_rate"):
      self.regulate_last = now
      self.regulate_rate = 1
    delta = now - self.regulate_last
    self.regulate_rate *= AVG_FRAMES
    self.regulate_rate += delta
    self.regulate_rate /= AVG_FRAMES + 1
    self.regulate_last = now
    return self.regulate_rate

  # MQTT callbacks
  def onConnect(self, client, userdata, flags, rc):
    log.info("Connected with result code", rc)
    if rc != 0:
      exit(1)
    self.subscribed = set()
    self.updateSubscriptions()
    self.updateObjectClasses()
    self.updateTRSMatrix()
    topic = PubSub.formatTopic(PubSub.CMD_DATABASE)
    self.pubsub.addCallback(topic, self.handleDatabaseMessage)
    log.info("Subscribed to", topic)
    # FIXME - update subscriptions when scenes/sensors/children added/deleted/renamed
    return

  def updateObjectClasses(self):
    results = self.cache_manager.data_source.getAssets()
    if results and 'results' in results:
      for scene in self.scenes:
        if scene.tracker is not None:
          scene.tracker.updateObjectClasses(results['results'])
    return

  def updateTRSMatrix(self):
    for scene in self.cache_manager.allScenes():
      if scene.trs_xyz_to_lla is not None:
        res = self.cache_manager.data_source.setTRSMatrix(scene.uid, scene.trs_xyz_to_lla)
        if res.errors:
          log.info(
                  "Failed to update trs matrix for scene %s. Errors: %s",
                  scene.name,
                  res.errors,
                )
    return

  def republishEvents(self, client, userdata, message):
    """
    Republishes the child analytics under parent topic that
    enables parent to visualize them.
    """
    topic = PubSub.parseTopic(message.topic)
    msg = orjson.loads(message.payload.decode('utf-8'))

    sender_id = topic['scene_id']
    sender = self.cache_manager.sceneWithID(sender_id)
    if sender is None:
      remote_sender = self.cache_manager.sceneWithRemoteChildID(sender_id)
      if remote_sender is None:
        log.error("UNKNOWN SENDER")
        return
      else:
        sender = remote_sender

    if not hasattr(sender, 'parent') or sender.parent is None:
      log.error("UNKNOWN PARENT", sender_id)
      return

    scene = self.cache_manager.sceneWithID(sender.parent)
    event_topic = PubSub.formatTopic(PubSub.EVENT,
                                      region_type=topic['region_type'], event_type=topic['event_type'],
                                      scene_id=scene.uid, region_id=topic['region_id'])

    self.transformObjectsinEvent(msg, sender)

    msg['metadata'] = applyChildTransform(msg['metadata'], sender.cameraPose)
    if 'from_child_scene' not in msg['metadata']:
      msg['metadata']['from_child_scene'] = sender.name
    else:
      msg['metadata']['from_child_scene'] = sender.name + " > " + msg['metadata']['from_child_scene']
    self.pubsub.publish(event_topic, orjson.dumps(msg, option=orjson.OPT_SERIALIZE_NUMPY))
    return

  def transformObjectsinEvent(self, event, sender):
    keys = ['objects', 'entered', 'exited']
    for k in keys:
      if k == 'exited':
        for i, obj in enumerate(event[k]):
          event[k][i]['object']['translation'] = sender.cameraPose.cameraPointToWorldPoint(
                                                            Point(obj['object']['translation'])).asNumpyCartesian.tolist()
      else:
        for i, obj in enumerate(event[k]):
          event[k][i]['translation'] = sender.cameraPose.cameraPointToWorldPoint(
                                                            Point(obj['translation'])).asNumpyCartesian.tolist()
    return

  def updateSubscriptions(self):
    log.debug("UPDATE SUBSCRIPTIONS")
    self.cache_manager.invalidate()
    if not hasattr(self, 'subscribed'):
      self.subscribed = set()
    need_subscribe = set()

    if not hasattr(self, 'subscribed_children'):
      self.subscribed_children = dict()
    need_subscribe_child = dict()

    self.scenes = self.cache_manager.allScenes()
    for scene in self.scenes:
      if not ControllerMode.isAnalyticsOnly():
        for camera in scene.cameras:
          need_subscribe.add((PubSub.formatTopic(PubSub.DATA_CAMERA, camera_id=camera),
                              self.handleMovingObjectMessage))
      else:
        need_subscribe.add((PubSub.formatTopic(PubSub.DATA_SCENE, scene_id=scene.uid, thing_type="+"),
                            self.handleSceneDataMessage))

      for sensor in scene.sensors:
        need_subscribe.add((PubSub.formatTopic(PubSub.DATA_SENSOR, sensor_id=sensor),
                            self.handleSensorMessage))

      if hasattr(scene, 'children'):
        child_scenes = self.cache_manager.data_source.getChildScenes(scene.uid)

        if not ControllerMode.isAnalyticsOnly():
          for info in child_scenes.get('results', []):
            if info['child_type'] == 'local':
              self.cache_manager.sceneWithID(info['child']).retrack = info['retrack']

              need_subscribe.add((PubSub.formatTopic(PubSub.DATA_EXTERNAL,
                                                     scene_id=info['child'], thing_type="+"),
                                  self.handleMovingObjectMessage))

              need_subscribe.add((PubSub.formatTopic(PubSub.EVENT, region_type="+",
                                                    event_type="+",
                                                    scene_id=info['child'],
                                                    region_id="+"),
                                  self.republishEvents))
            else:
              child_obj = ChildSceneController(self.root_cert, info, self)
              self.cache_manager.cached_child_transforms_by_uid[info['remote_child_id']] = Scene.deserialize(info)
              need_subscribe_child[info['remote_child_id']] = child_obj
              need_subscribe.add((PubSub.formatTopic(PubSub.SYS_CHILDSCENE_STATUS, scene_id=info['remote_child_id']), child_obj.publishStatus))

    # disconnect old children clients
    for old_child, cobj in self.subscribed_children.items():
      if old_child not in need_subscribe_child:
        self.cache_manager.cached_child_transforms_by_uid.pop(old_child, 'None')
      cobj.loopStop()

    # connect to all children
    for new_child, cobj in need_subscribe_child.items():
      log.info(f"Connecting to remote child {new_child}")
      cobj.loopStart()

    self.subscribed_children = need_subscribe_child

    new = need_subscribe - self.subscribed
    old = self.subscribed - need_subscribe
    for topic, callback in old:
      self.pubsub.removeCallback(topic)
      log.info("Unsubscribed from", topic)
    for topic, callback in new:
      self.pubsub.addCallback(topic, callback)
      log.info("Subscribed to", topic)
    self.subscribed = need_subscribe
    return