adsist-cms/python-detector-worker
3
0
Fork 0
Code Issues Pull requests Projects Releases Packages 1 Wiki Activity Actions

refactor: enhance FFmpegRTSPReader to implement persistent file locking for PPM frame reading
All checks were successful
Build Worker Base and Application Images / check-base-changes (push) Successful in 9s
Details
Build Worker Base and Application Images / build-base (push) Has been skipped
Details
Build Worker Base and Application Images / build-docker (push) Successful in 3m8s
Details
Build Worker Base and Application Images / deploy-stack (push) Successful in 14s
Details

Browse source
This commit is contained in:
Siwat Sirichai 2025-09-26 03:04:53 +07:00
parent fe0da18d0f
commit a12e3efa12
1 changed files with 41 additions and 20 deletions
Download patch file Download diff file Expand all files Collapse all files

61
core/streaming/readers.py
View file

@ -10,6 +10,7 @@ import requests
import numpy as np import numpy as np
import os import os
import subprocess import subprocess
import fcntl
from typing import Optional, Callable from typing import Optional, Callable
from watchdog.observers import Observer from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler from watchdog.events import FileSystemEventHandler
@ -94,7 +95,7 @@ class FFmpegRTSPReader:
self.temp_file = f"/tmp/claude/camera_{self.camera_id.replace(' ', '_')}.raw" self.temp_file = f"/tmp/claude/camera_{self.camera_id.replace(' ', '_')}.raw"
os.makedirs("/tmp/claude", exist_ok=True) os.makedirs("/tmp/claude", exist_ok=True)
# Use PPM format - uncompressed with header, supports -update 1 # Use PPM format with single file (will use file locking for concurrency)
self.temp_file = f"/tmp/claude/camera_{self.camera_id.replace(' ', '_')}.ppm" self.temp_file = f"/tmp/claude/camera_{self.camera_id.replace(' ', '_')}.ppm"
cmd = [ cmd = [
@ -176,31 +177,51 @@ class FFmpegRTSPReader:
if self.frame_ready_event.wait(timeout=restart_check_interval): if self.frame_ready_event.wait(timeout=restart_check_interval):
self.frame_ready_event.clear() self.frame_ready_event.clear()
# Read PPM frame (uncompressed with header) # Read PPM frame with persistent lock attempts until new inotify
try: try:
if os.path.exists(self.temp_file): if os.path.exists(self.temp_file):
# Read PPM with OpenCV (handles RGB->BGR conversion automatically) # Keep trying to acquire lock until new inotify event or success
frame = cv2.imread(self.temp_file) max_attempts = 50 # ~500ms worth of attempts
for attempt in range(max_attempts):
# Check if new inotify event arrived (cancel current attempt)
if self.frame_ready_event.is_set():
break
if frame is not None and frame.shape == (self.height, self.width, 3): try:
# Call frame callback directly with open(self.temp_file, 'rb') as f:
if self.frame_callback: # Try to acquire shared lock (non-blocking)
self.frame_callback(self.camera_id, frame) fcntl.flock(f.fileno(), fcntl.LOCK_SH | fcntl.LOCK_NB)
frame_count += 1 # Success! File is locked, safe to read
frame = cv2.imread(self.temp_file)
# Log progress if frame is not None and frame.shape == (self.height, self.width, 3):
current_time = time.time() # Call frame callback directly
if current_time - last_log_time >= 30: if self.frame_callback:
logger.info(f"Camera {self.camera_id}: {frame_count} PPM frames processed reactively") self.frame_callback(self.camera_id, frame)
last_log_time = current_time
else:
logger.debug(f"Camera {self.camera_id}: Invalid PPM frame")
else:
logger.debug(f"Camera {self.camera_id}: PPM file not found yet")
except (IOError, OSError) as e: frame_count += 1
logger.debug(f"Camera {self.camera_id}: File read error: {e}")
# Log progress
current_time = time.time()
if current_time - last_log_time >= 30:
logger.info(f"Camera {self.camera_id}: {frame_count} PPM frames processed with persistent locking")
last_log_time = current_time
# Invalid frame - just skip, no logging needed
# Successfully processed frame
break
except (OSError, IOError):
# File is still locked, wait a bit and try again
time.sleep(0.01) # 10ms wait between attempts
continue
# If we get here, exhausted attempts or file not ready - just continue
except (IOError, OSError):
# File errors are routine, just continue
pass
except Exception as e: except Exception as e:
logger.error(f"Camera {self.camera_id}: Error in reactive frame reading: {e}") logger.error(f"Camera {self.camera_id}: Error in reactive frame reading: {e}")