security_vision/trap
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

optional camera

Browse source
This commit is contained in:
Ruben van de Ven 2025-07-11 14:28:25 +02:00
parent 12852c3ae6
commit 0e29371e94
1 changed files with 9 additions and 2 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

11
trap/video_sources.py
View file

@ -124,7 +124,10 @@ class SingleCvVideoSource(VideoSource):
class RtspSource(SingleCvVideoSource): class RtspSource(SingleCvVideoSource):
def __init__(self, video_url: str | Path, camera: Camera = None): def __init__(self, video_url: str | Path, camera: Camera = None):
gst = f"rtspsrc location={video_url} latency=0 buffer-mode=auto ! decodebin ! videoconvert ! appsink max-buffers=0 drop=true" # keep max 1 frame in app-buffer (0 = unlimited)
# When using gstreamer 1.28 drop=true is deprecated, use: leaky-type=2 which frame to drop: https://gstreamer.freedesktop.org/documentation/applib/gstappsrc.html?gi-language=c
gst = f"rtspsrc location={video_url} latency=0 buffer-mode=auto ! decodebin ! videoconvert ! appsink max-buffers=1 drop=true"
logger.info(f"Capture gstreamer (gst-launch-1.0): {gst}") logger.info(f"Capture gstreamer (gst-launch-1.0): {gst}")
self.video = cv2.VideoCapture(gst, cv2.CAP_GSTREAMER) self.video = cv2.VideoCapture(gst, cv2.CAP_GSTREAMER)
self.frame_idx = 0 self.frame_idx = 0
@ -209,7 +212,7 @@ class CameraSource(SingleCvVideoSource):
self.video.set(cv2.CAP_PROP_FPS, self.camera.fps) self.video.set(cv2.CAP_PROP_FPS, self.camera.fps)
self.frame_idx = 0 self.frame_idx = 0
def get_video_source(video_sources: List[UrlOrPath], camera: Camera, frame_offset=0, frame_end:Optional[int]=None, loop=False): def get_video_source(video_sources: List[UrlOrPath], camera: Optional[Camera] = None, frame_offset=0, frame_end:Optional[int]=None, loop=False):
if str(video_sources[0]).isdigit(): if str(video_sources[0]).isdigit():
# numeric input is a CV camera # numeric input is a CV camera
@ -230,3 +233,7 @@ def get_video_source(video_sources: List[UrlOrPath], camera: Camera, frame_offse
return FilelistSource(video_sources, offset = frame_offset, end=frame_end, loop=loop) return FilelistSource(video_sources, offset = frame_offset, end=frame_end, loop=loop)
# os.environ["OPENCV_FFMPEG_CAPTURE_OPTIONS"] = "fflags;nobuffer|flags;low_delay|avioflags;direct|rtsp_transport;udp" # os.environ["OPENCV_FFMPEG_CAPTURE_OPTIONS"] = "fflags;nobuffer|flags;low_delay|avioflags;direct|rtsp_transport;udp"
def get_video_source_from_str(video_sources: List[str]):
paths = [UrlOrPath(s) for s in video_sources]
return get_video_source(paths)