Protect dicts used in different threads to avoid race conditions

Author: saratpoluri

controller/src/controller/detections_builder.py

@@ -76,8 +76,13 @@ def prepareObjDict(scene, obj, update_visibility, include_sensors=False):
   if include_sensors:
     sensors_output = {}
 
+    # Copy sensor data while holding lock, then release
+    with chain_data._lock:
+      env_state_copy = dict(chain_data.env_sensor_state)
+      attr_events_copy = dict(chain_data.attr_sensor_events)
+
     # Environmental sensors: readings + exposure as structured object
-    for sensor_id, state in chain_data.env_sensor_state.items():
+    for sensor_id, state in env_state_copy.items():
       values = state['readings'] if 'readings' in state and state['readings'] else []
 
       # Calculate total exposure (including current value if present)
@@ -94,7 +99,7 @@ def prepareObjDict(scene, obj, update_visibility, include_sensors=False):
       }
 
     # Attribute sensors: events as structured object
-    for sensor_id, events in chain_data.attr_sensor_events.items():
+    for sensor_id, events in attr_events_copy.items():
       if events:
         sensors_output[sensor_id] = {
           'values': events

controller/src/controller/moving_object.py

@@ -34,6 +34,7 @@ class ChainData:
   active_sensors: set = field(default_factory=set)
   env_sensor_state: Dict = field(default_factory=dict)  # {'sensor_id': {'last_reading': (ts, val), 'exposure': {...}}}
   attr_sensor_events: Dict = field(default_factory=dict)  # {'sensor_id': [(ts, val), ...]}
+  _lock: Lock = field(default_factory=Lock)
 
 class Chronoloc:
   def __init__(self, point: Point, when: datetime, bounds: Rectangle):

controller/src/controller/scene.py

@@ -339,9 +339,10 @@ def processSensorData(self, jdata, when):
         # instead of appending, but frequent value changes can still cause unbounded growth.
         cur_value_float = float(cur_value)
         for obj in objects_in_sensor:
-          if sensor_id in obj.chain_data.env_sensor_state:
-            # Update exposure incrementally
-            state = obj.chain_data.env_sensor_state[sensor_id]
+          with obj.chain_data._lock:
+            if sensor_id in obj.chain_data.env_sensor_state:
+              # Update exposure incrementally
+              state = obj.chain_data.env_sensor_state[sensor_id]
             last_time = state['exposure']['last_time']
             last_value = state['exposure']['last_value']
 
@@ -352,38 +353,38 @@ def processSensorData(self, jdata, when):
                 avg_value = (last_value + cur_value_float) / 2.0
                 state['exposure']['total'] += avg_value * dt
 
-            state['exposure']['last_time'] = timestamp_epoch
-            state['exposure']['last_value'] = cur_value_float
-
-            # Update readings array: append if value changed, update timestamp if same
-            if 'readings' not in state:
-              state['readings'] = []
-            if state['readings'] and state['readings'][-1][1] == cur_value_float:
-              # Value unchanged - update timestamp
-              state['readings'][-1] = (timestamp_str, cur_value_float)
+              state['exposure']['last_time'] = timestamp_epoch
+              state['exposure']['last_value'] = cur_value_float
+
+              # Update readings array: append if value changed, update timestamp if same
+              if 'readings' not in state:
+                state['readings'] = []
+              if state['readings'] and state['readings'][-1][1] == cur_value_float:
+                # Value unchanged - update timestamp
+                state['readings'][-1] = (timestamp_str, cur_value_float)
+              else:
+                # Value changed - append new reading
+                state['readings'].append((timestamp_str, cur_value_float))
             else:
-              # Value changed - append new reading
-              state['readings'].append((timestamp_str, cur_value_float))
-          else:
-            # First reading - initialize from entry time (or first_seen for scene-wide sensors)
-            entry_str = obj.chain_data.regions.get(sensor_id, {}).get('entered')
-            if not entry_str:
-              # No entry time (scene-wide sensor or pre-existing object) - use first_seen
-              entry_str = get_iso_time(obj.first_seen)
-
-            entry_epoch = get_epoch_time(entry_str)
-            dt = timestamp_epoch - entry_epoch
-            # Only calculate initial exposure if time moved forward
-            initial_exposure = cur_value_float * dt if dt > 0 else 0.0
-
-            obj.chain_data.env_sensor_state[sensor_id] = {
-              'readings': [(timestamp_str, cur_value_float)],
-              'exposure': {
-                'total': initial_exposure,
-                'last_time': timestamp_epoch,
-                'last_value': cur_value_float
+              # First reading - initialize from entry time (or first_seen for scene-wide sensors)
+              entry_str = obj.chain_data.regions.get(sensor_id, {}).get('entered')
+              if not entry_str:
+                # No entry time (scene-wide sensor or pre-existing object) - use first_seen
+                entry_str = get_iso_time(obj.first_seen)
+
+              entry_epoch = get_epoch_time(entry_str)
+              dt = timestamp_epoch - entry_epoch
+              # Only calculate initial exposure if time moved forward
+              initial_exposure = cur_value_float * dt if dt > 0 else 0.0
+
+              obj.chain_data.env_sensor_state[sensor_id] = {
+                'readings': [(timestamp_str, cur_value_float)],
+                'exposure': {
+                  'total': initial_exposure,
+                  'last_time': timestamp_epoch,
+                  'last_value': cur_value_float
+                }
               }
-            }
 
       elif sensor.singleton_type == "attribute":
         # Event history tracking - append discrete events (or update timestamp if value unchanged)
@@ -392,14 +393,15 @@ def processSensorData(self, jdata, when):
         # Convert to string for consistent type comparison (attributes can be non-numeric)
         cur_value_str = str(cur_value)
         for obj in objects_in_sensor:
-          if sensor_id not in obj.chain_data.attr_sensor_events:
-            obj.chain_data.attr_sensor_events[sensor_id] = []
-
-          events = obj.chain_data.attr_sensor_events[sensor_id]
-          if events and events[-1][1] == cur_value_str:
-            # Value unchanged - update timestamp of last event instead of appending
-            events[-1] = (timestamp_str, cur_value_str)
-          else:
+          with obj.chain_data._lock:
+            if sensor_id not in obj.chain_data.attr_sensor_events:
+              obj.chain_data.attr_sensor_events[sensor_id] = []
+
+            events = obj.chain_data.attr_sensor_events[sensor_id]
+            if events and events[-1][1] == cur_value_str:
+              # Value unchanged - update timestamp of last event instead of appending
+              events[-1] = (timestamp_str, cur_value_str)
+            else:
             # Value changed - append new event
             events.append((timestamp_str, cur_value_str))
 
@@ -617,37 +619,39 @@ def _updateRegionEvents(self, detectionType, regions, now, now_str, curObjects):
           if region.singleton_type == "environmental":
             # Use 'now' directly (already epoch time) to avoid precision loss from string conversion
 
-            if hasattr(region, 'value') and hasattr(region, 'lastWhen'):
-              # Sensor has cached value - initialize with it
-              ts_str = get_iso_time(region.lastWhen)
-              last_value = float(region.value) if region.value is not None else None
-
-              # Start exposure accumulation from entry time, not sensor reading time
-              # Exposure = value * (now - entry_time)
-              obj.chain_data.env_sensor_state[key] = {
-                'readings': [(ts_str, region.value)],
-                'exposure': {
-                  'total': 0.0,
-                  'last_time': now,
-                  'last_value': last_value
+            with obj.chain_data._lock:
+              if hasattr(region, 'value') and hasattr(region, 'lastWhen'):
+                # Sensor has cached value - initialize with it
+                ts_str = get_iso_time(region.lastWhen)
+                last_value = float(region.value) if region.value is not None else None
+
+                # Start exposure accumulation from entry time, not sensor reading time
+                # Exposure = value * (now - entry_time)
+                obj.chain_data.env_sensor_state[key] = {
+                  'readings': [(ts_str, region.value)],
+                  'exposure': {
+                    'total': 0.0,
+                    'last_time': now,
+                    'last_value': last_value
+                  }
                 }
-              }
-            else:
-              # No cached value yet
-              obj.chain_data.env_sensor_state[key] = {
-                'readings': [],
-                'exposure': {
-                  'total': 0.0,
-                  'last_time': now,
-                  'last_value': None
+              else:
+                # No cached value yet
+                obj.chain_data.env_sensor_state[key] = {
+                  'readings': [],
+                  'exposure': {
+                    'total': 0.0,
+                    'last_time': now,
+                    'last_value': None
+                  }
                 }
-              }
 
           elif region.singleton_type == "attribute":
             # Attribute sensors only tag objects present when MQTT arrives
             # Do NOT initialize with cached values (those belong to other objects)
-            if key not in obj.chain_data.attr_sensor_events:
-              obj.chain_data.attr_sensor_events[key] = []
+            with obj.chain_data._lock:
+              if key not in obj.chain_data.attr_sensor_events:
+                obj.chain_data.attr_sensor_events[key] = []
 
       if (len(new) or len(old)) and now - region.when > DEBOUNCE_DELAY:
         log.debug("REGION EVENT", key, now_str, regionObjects, len(objects))
@@ -685,18 +689,19 @@ def _updateRegionEvents(self, detectionType, regions, now, now_str, curObjects):
       # Always clean up exited objects, even if debounce prevented event emission
       for obj in regionObjects:
         if obj.gid in old:
-          obj.chain_data.regions.pop(key, None)
+          with obj.chain_data._lock:
+            obj.chain_data.regions.pop(key, None)
 
-          # Clean up sensor tracking on exit
-          if region.singleton_type is not None:
-            obj.chain_data.active_sensors.discard(key)
+            # Clean up sensor tracking on exit
+            if region.singleton_type is not None:
+              obj.chain_data.active_sensors.discard(key)
 
-            # Environmental sensors: clear state on exit (data doesn't persist)
-            if region.singleton_type == "environmental":
-              obj.chain_data.env_sensor_state.pop(key, None)
+              # Environmental sensors: clear state on exit (data doesn't persist)
+              if region.singleton_type == "environmental":
+                obj.chain_data.env_sensor_state.pop(key, None)
 
-            # Attribute sensors: keep event history (data persists after exit)
-            # attr_sensor_events[key] intentionally not removed
+              # Attribute sensors: keep event history (data persists after exit)
+              # attr_sensor_events[key] intentionally not removed
 
     return updated