diff --git a/app.py b/app.py index eb1440f..7b82d23 100644 --- a/app.py +++ b/app.py @@ -201,10 +201,11 @@ else: os.makedirs("models", exist_ok=True) logger.info("Ensured models directory exists") -# Initialize stream manager with config value -from core.streaming import initialize_stream_manager -initialize_stream_manager(max_streams=config.get('max_streams', 10)) -logger.info(f"Initialized stream manager with max_streams={config.get('max_streams', 10)}") +# Stream manager already initialized at module level with max_streams=20 +# Calling initialize_stream_manager() creates a NEW instance, breaking references +# from core.streaming import initialize_stream_manager +# initialize_stream_manager(max_streams=config.get('max_streams', 10)) +logger.info(f"Using stream manager with max_streams=20 (module-level initialization)") # Frames are now stored in the shared cache buffer from core.streaming.buffers # latest_frames = {} # Deprecated - using shared_cache_buffer instead diff --git a/core/communication/websocket.py b/core/communication/websocket.py index e53096a..d20ee32 100644 --- a/core/communication/websocket.py +++ b/core/communication/websocket.py @@ -197,18 +197,24 @@ class WebSocketHandler: async def _handle_set_subscription_list(self, message: SetSubscriptionListMessage) -> None: """Handle setSubscriptionList message for declarative subscription management.""" - logger.info(f"[RX Processing] setSubscriptionList with {len(message.subscriptions)} subscriptions") + logger.info(f"🎯 [RX Processing] setSubscriptionList with {len(message.subscriptions)} subscriptions") + for i, sub in enumerate(message.subscriptions): + logger.info(f" 📋 Sub {i+1}: {sub.subscriptionIdentifier} (model: {sub.modelId})") # Update worker state with new subscriptions worker_state.set_subscriptions(message.subscriptions) # Phase 2: Download and manage models + logger.info("📦 Starting model download phase...") await self._ensure_models(message.subscriptions) + logger.info("✅ Model download phase complete") # Phase 3 & 4: Integrate with streaming management and tracking + logger.info("🎬 Starting stream subscription update...") await self._update_stream_subscriptions(message.subscriptions) + logger.info("✅ Stream subscription update complete") - logger.info("Subscription list updated successfully") + logger.info("🏁 Subscription list updated successfully") async def _ensure_models(self, subscriptions) -> None: """Ensure all required models are downloaded and available.""" diff --git a/core/streaming/manager.py b/core/streaming/manager.py index 497f1b8..2de86e4 100644 --- a/core/streaming/manager.py +++ b/core/streaming/manager.py @@ -85,8 +85,11 @@ class StreamManager: with self._round_robin_lock: if camera_id not in self._camera_list: self._camera_list.append(camera_id) - - logger.info(f"Created tracking queue for camera {camera_id}") + logger.info(f"✅ Created tracking queue for camera {camera_id}, camera_list now has {len(self._camera_list)} cameras: {self._camera_list}") + else: + logger.warning(f"Camera {camera_id} already in camera_list") + else: + logger.debug(f"Camera {camera_id} already has tracking queue") def _remove_camera_queue(self, camera_id: str): """Remove tracking queue for a camera that's no longer active.""" @@ -153,6 +156,10 @@ class StreamManager: if not success: self._remove_subscription_internal(subscription_id) return False + else: + # Stream already exists, but ensure queue exists too + logger.info(f"Stream already exists for {camera_id}, ensuring queue exists") + self._ensure_camera_queue(camera_id) logger.info(f"Added subscription {subscription_id} for camera {camera_id} " f"({len(self._camera_subscribers[camera_id])} total subscribers)") @@ -188,6 +195,7 @@ class StreamManager: def _start_stream(self, camera_id: str, stream_config: StreamConfig) -> bool: """Start a stream for the given camera.""" try: + logger.info(f"🚀 _start_stream called for {camera_id}") if stream_config.rtsp_url: # RTSP stream using FFmpeg subprocess with CUDA acceleration logger.info(f"\033[94m[RTSP] Starting {camera_id}\033[0m") @@ -199,7 +207,9 @@ class StreamManager: reader.set_frame_callback(self._frame_callback) reader.start() self._streams[camera_id] = reader + logger.info(f"🎬 About to call _ensure_camera_queue for {camera_id}") self._ensure_camera_queue(camera_id) # Create tracking queue + logger.info(f"✅ _ensure_camera_queue completed for {camera_id}") logger.info(f"\033[92m[RTSP] {camera_id} connected\033[0m") elif stream_config.snapshot_url: @@ -214,7 +224,9 @@ class StreamManager: reader.set_frame_callback(self._frame_callback) reader.start() self._streams[camera_id] = reader + logger.info(f"🎬 About to call _ensure_camera_queue for {camera_id}") self._ensure_camera_queue(camera_id) # Create tracking queue + logger.info(f"✅ _ensure_camera_queue completed for {camera_id}") logger.info(f"\033[92m[HTTP] {camera_id} connected\033[0m") else: @@ -334,18 +346,22 @@ class StreamManager: while not self._stop_workers.is_set(): try: + logger.debug(f"Worker {threading.current_thread().name} loop iteration, stop_event={self._stop_workers.is_set()}") + # Get next camera in round-robin fashion camera_id, item = self._get_next_camera_item() if camera_id is None: # No cameras have items, sleep briefly consecutive_empty += 1 + logger.debug(f"Worker {threading.current_thread().name}: All queues empty ({consecutive_empty}/{max_consecutive_empty})") if consecutive_empty >= max_consecutive_empty: time.sleep(0.1) # Sleep 100ms if nothing to process consecutive_empty = 0 continue consecutive_empty = 0 # Reset counter when we find work + logger.info(f"Worker {threading.current_thread().name}: Processing frame from {camera_id}") frame = item['frame'] timestamp = item['timestamp'] @@ -353,11 +369,13 @@ class StreamManager: # Check if frame is too old (drop if > 1 second old) age = time.time() - timestamp if age > 1.0: - logger.debug(f"Dropping old frame for {camera_id} (age: {age:.2f}s)") + logger.warning(f"Dropping old frame for {camera_id} (age: {age:.2f}s)") continue + logger.info(f"Worker {threading.current_thread().name}: Calling tracking sync for {camera_id}") # Process tracking for this camera's frame self._process_tracking_for_camera_sync(camera_id, frame) + logger.info(f"Worker {threading.current_thread().name}: Finished tracking sync for {camera_id}") except Exception as e: logger.error(f"Error in tracking worker: {e}", exc_info=True) @@ -367,32 +385,48 @@ class StreamManager: def _get_next_camera_item(self): """Get next item from camera queues using round-robin scheduling.""" with self._round_robin_lock: - if not self._camera_list: + # Get current list of cameras from actual tracking queues (central state) + camera_list = list(self._tracking_queues.keys()) + + # Debug: show ALL state + logger.info(f"🔍 _tracking_queues keys: {list(self._tracking_queues.keys())}") + logger.info(f"🔍 _streams keys: {list(self._streams.keys())}") + logger.info(f"🔍 _subscriptions keys: {list(self._subscriptions.keys())}") + + if not camera_list: + logger.warning("⚠️ _get_next_camera_item: No cameras have tracking queues yet, but streams/subscriptions exist!") return None, None + logger.debug(f"_get_next_camera_item: {len(camera_list)} cameras with queues: {camera_list}") + attempts = 0 - max_attempts = len(self._camera_list) + max_attempts = len(camera_list) while attempts < max_attempts: - # Get current camera - if self._camera_round_robin_index >= len(self._camera_list): + # Get current camera using round-robin index + if self._camera_round_robin_index >= len(camera_list): self._camera_round_robin_index = 0 - camera_id = self._camera_list[self._camera_round_robin_index] + camera_id = camera_list[self._camera_round_robin_index] + logger.debug(f"_get_next_camera_item: Trying camera {camera_id} (attempt {attempts + 1}/{max_attempts})") # Move to next camera for next call - self._camera_round_robin_index = (self._camera_round_robin_index + 1) % len(self._camera_list) + self._camera_round_robin_index = (self._camera_round_robin_index + 1) % len(camera_list) # Try to get item from this camera's queue - if camera_id in self._tracking_queues: - try: - item = self._tracking_queues[camera_id].get_nowait() - return camera_id, item - except queue.Empty: - pass # Try next camera + queue_size = self._tracking_queues[camera_id].qsize() + logger.debug(f"_get_next_camera_item: Camera {camera_id} queue has {queue_size} items") + try: + item = self._tracking_queues[camera_id].get_nowait() + logger.info(f"_get_next_camera_item: Got item from {camera_id}") + return camera_id, item + except queue.Empty: + logger.debug(f"_get_next_camera_item: Camera {camera_id} queue empty") + pass # Try next camera attempts += 1 + logger.debug("_get_next_camera_item: All cameras empty") return None, None # All cameras empty def _process_tracking_for_camera_sync(self, camera_id: str, frame): @@ -404,7 +438,12 @@ class StreamManager: for subscription_id in subscription_ids: subscription_info = self._subscriptions.get(subscription_id) - if not subscription_info or not subscription_info.tracking_integration: + if not subscription_info: + logger.warning(f"No subscription info found for {subscription_id}") + continue + + if not subscription_info.tracking_integration: + logger.debug(f"No tracking integration for {subscription_id} (camera {camera_id}), skipping inference") continue display_id = subscription_id.split(';')[0] if ';' in subscription_id else subscription_id