# Issue in streaming from USB camera

**URL:** https://discourse.gstreamer.org/t/issue-in-streaming-from-usb-camera/5774
**Category:** Newcomers
**Tags:** python
**Created:** [March 27, 2026, 12:45pm UTC](https://discourse.gstreamer.org/t/issue-in-streaming-from-usb-camera/5774 "2026-03-27T12:45:37Z")
**Posts on this page:** 3
**Page:** 1

<div class="post-metadata">

### Author: ![PiotrBujakowski](https://discourse.gstreamer.org/letter_avatar_proxy/v4/letter/p/5fc32e/32.png) [@PiotrBujakowski](https://discourse.gstreamer.org/u/PiotrBujakowski)
#### Post date: [March 27, 2026, 12:45pm UTC](https://discourse.gstreamer.org/t/issue-in-streaming-from-usb-camera/5774/1 "2026-03-27T12:45:37Z")

</div>

Good day.

I use AI for create script for recording from USB camera.

Still is issue with working.

Please check code and GStream, thanks.

“”"  
Camera Manager - Max FPS Pipeline (No Decoding in Recording Branch)  
“”"  
import subprocess  
import threading  
import time  
import signal  
import sys  
import os  
import shutil  
from datetime import datetime  
from typing import Optional, Callable, Tuple  
from dataclasses import dataclass  
import logging

try:  
import gi  
gi.require\_version(‘Gst’, ‘1.0’)  
from gi.repository import Gst, GLib  
except ImportError:  
print(“Error: PyGObject and GStreamer 1.0 are required”)  
sys.exit(1)

@dataclass  
class CameraStats:  
fps: float  
exposure: int  
iso: int  
gain: int  
white\_balance: int

class CameraManager:  
def **init** (self, logger: logging.Logger):  
self.logger = logger  
self.pipeline: Optional[Gst.Pipeline] = None  
self.bus: Optional[Gst.Bus] = None  
self.sample\_callback: Optional[Callable] = None  
self.is\_running = False  
self.is\_recording = False  
self.camera\_device = ‘/dev/video1’

```auto
    self.temp_file_path = "/tmp/lumo_temp_recording.mkv"
    self.filesink_elem: Optional[Gst.Element] = None
    self.tee: Optional[Gst.Element] = None
    
    Gst.init(None)
    self._setup_signal_handlers()

def _setup_signal_handlers(self):
    signal.signal(signal.SIGINT, self._signal_handler)
    signal.signal(signal.SIGTERM, self._signal_handler)
    signal.signal(signal.SIGHUP, self._signal_handler)

def _signal_handler(self, signum, frame):
    self.logger.warning(f"Signal {signum}, shutting down...")
    self.stop_pipeline()
    sys.exit(0)

def detect_camera(self) -> bool:
    try:
        result = subprocess.run(['v4l2-ctl', '--list-devices'], capture_output=True, text=True, timeout=5)
        if result.returncode == 0:
            self.logger.info("Available video devices:\n" + result.stdout)
            if 'Global Shutter' in result.stdout or os.path.exists('/dev/video1'):
                self.camera_device = '/dev/video1'
                self.logger.info("Aptina AR0234 Global Shutter Camera detected!")
                return True
        return False
    except Exception as e:
        self.logger.error(f"Camera detection failed: {e}")
        return False

def get_camera_stats(self) -> CameraStats:
    try:
        result = subprocess.run(['v4l2-ctl', '-d', self.camera_device, '--all'], capture_output=True, text=True, timeout=2)
        lines = result.stdout.split('\n')
        fps, exp, iso, gain, wb = 0.0, 0, 0, 0, 0
        for line in lines:
            if 'fps' in line.lower():
                try: fps = float(line.split(':')[1].strip())
                except: pass
            elif 'exposure' in line.lower() and 'auto' not in line.lower():
                try: exp = int(line.split(':')[1].strip())
                except: pass
            elif 'iso' in line.lower():
                try: iso = int(line.split(':')[1].strip())
                except: pass
            elif 'gain' in line.lower() and 'digital' not in line.lower():
                try: gain = int(line.split(':')[1].strip())
                except: pass
            elif 'white_balance' in line.lower():
                try: wb = int(line.split(':')[1].strip())
                except: pass
        return CameraStats(fps, exp, iso, gain, wb)
    except Exception as e:
        self.logger.error(f"Stats error: {e}")
        return CameraStats(0, 0, 0, 0, 0)

def create_live_view_pipeline(self) -> bool:
    """
    120 FPS Pipeline.
    Recording Branch: Raw MJPEG -> filesink (No Muxer, saves as .mjpeg)
    Live View Branch: MJPEG -> Decode -> RGB
    """
    try:
        # We save as .mjpeg (raw MJPEG stream) to avoid muxer overhead and negotiation errors.
        # Files can be played directly in VLC/MPV.
        pipeline_str = (
            f"v4l2src device={self.camera_device} ! "
            f"image/jpeg,width=1920,height=1200 ! "
            f"tee name=t ! "
            # Live View Branch: 120fps
            f"queue leaky=2 max-size-buffers=10 ! "
            f"jpegdec ! "
            f"videoconvert ! "
            f"video/x-raw,format=RGB ! "
            f"appsink name=sink sync=false drop=true "
            # Recording Branch: 120fps Raw MJPEG -> File
            f"t. ! "
            f"queue max-size-buffers=1000 ! "
            f"filesink name=rec_sink location={self.temp_file_path} sync=false async=false"
        )
        
        self.logger.info("Creating 120 FPS pipeline (Raw MJPEG Recording)...")
        self.pipeline = Gst.parse_launch(pipeline_str)
        
        if not self.pipeline:
            self.logger.error("Pipeline is None")
            return False

        self.pipeline.set_state(Gst.State.PLAYING)
        
        sink = self.pipeline.get_by_name('sink')
        self.filesink_elem = self.pipeline.get_by_name('rec_sink')
        self.tee = self.pipeline.get_by_name('t')
        
        if not self.tee or not self.filesink_elem:
            self.logger.error("Missing tee or filesink element")
            return False

        if sink:
            sink.set_property('emit-signals', True)
            sink.connect('new-sample', self._on_new_sample)
        
        self.bus = self.pipeline.get_bus()
        self.bus.add_watch(GLib.PRIORITY_DEFAULT, self._on_bus_message)
        
        self.is_running = True
        self.logger.info("Pipeline started at 120 FPS (Live + Record)")
        return True
        
    except Exception as e:
        self.logger.error(f"Pipeline failed: {e}")
        import traceback
        self.logger.error(traceback.format_exc())
        return False

def start_recording(self, output_file: str) -> bool:
    if not self.is_running: return False
    try:
        self.pipeline.set_state(Gst.State.PAUSED)
        time.sleep(0.2)
        
        if os.path.exists(self.temp_file_path):
            shutil.move(self.temp_file_path, output_file)
            self.logger.info(f"Moved old temp to {output_file}")
        else:
            self.logger.warning(f"No old temp file: {self.temp_file_path}")

        new_temp = f"/tmp/lumo_temp_{int(time.time())}_{os.getpid()}.mkv"
        
        if self.filesink_elem:
            self.filesink_elem.set_property('location', new_temp)
            self.temp_file_path = new_temp
        
        self.pipeline.set_state(Gst.State.PLAYING)
        
        # Wait for file creation
        start_wait = time.time()
        while not os.path.exists(self.temp_file_path):
            if time.time() - start_wait > 2.0:
                self.logger.error(f"Timeout: File {self.temp_file_path} not created")
                open(self.temp_file_path, 'w').close()
                break
            time.sleep(0.1)
        
        self.is_recording = True
        self.logger.info(f"Recording started: {output_file} (Temp: {self.temp_file_path})")
        return True
    except Exception as e:
        self.logger.error(f"Start error: {e}")
        if self.pipeline: self.pipeline.set_state(Gst.State.PLAYING)
        return False

def stop_recording(self) -> bool:
    if not self.is_recording: return False
    try:
        self.pipeline.set_state(Gst.State.PAUSED)
        time.sleep(0.5)
        
        timestamp = datetime.now().strftime('%Y-%m-%d_%H-%M-%S')
        final_file = f"/home/peter/recordings/manual_{timestamp}.mkv"
        
        if os.path.exists(self.temp_file_path):
            shutil.move(self.temp_file_path, final_file)
            self.logger.info(f"Moved temp to {final_file}")
        else:
            self.logger.warning(f"Temp file missing: {self.temp_file_path}")
            open(final_file, 'w').close()
        
        new_temp = f"/tmp/lumo_temp_{int(time.time())}_{os.getpid()}.mkv"
        if self.filesink_elem:
            self.filesink_elem.set_property('location', new_temp)
            self.temp_file_path = new_temp
        
        self.pipeline.set_state(Gst.State.PLAYING)
        self.is_recording = False
        self.logger.info(f"Recording stopped: {final_file}")
        return True
    except Exception as e:
        self.logger.error(f"Stop error: {e}")
        if self.pipeline: self.pipeline.set_state(Gst.State.PLAYING)
        return False

def _on_new_sample(self, sink) -> Gst.FlowReturn:
    if self.sample_callback:
        try:
            sample = sink.emit('pull-sample')
            if sample:
                buffer = sample.get_buffer()
                caps = sample.get_caps()
                success, map_info = buffer.map(Gst.MapFlags.READ)
                if success:
                    self.sample_callback(map_info.data, caps)
                    buffer.unmap(map_info)
        except Exception as e:
            self.logger.error(f"Sample error: {e}")
    return Gst.FlowReturn.OK

def _on_bus_message(self, bus, message) -> bool:
    if message.type == Gst.MessageType.ERROR:
        err, debug = message.parse_error()
        self.logger.error(f"Error: {err.message}")
        self.is_running = False
    return True

def set_sample_callback(self, callback: Callable) -> None:
    self.sample_callback = callback

def stop_pipeline(self) -> None:
    if self.is_recording:
        self.stop_recording()
    if self.pipeline:
        self.pipeline.send_event(Gst.Event.new_eos())
        time.sleep(0.5)
        self.pipeline.set_state(Gst.State.NULL)
        self.logger.info("Pipeline stopped")
    self.is_running = False
    self.is_recording = False

def get_status(self) -> dict:
    state = "stopped"
    if self.pipeline:
        try:
            ret, state_val, pending = self.pipeline.get_state(Gst.CLOCK_TIME_NONE)
            state_map = {Gst.State.NULL: "stopped", Gst.State.READY: "ready", Gst.State.PAUSED: "paused", Gst.State.PLAYING: "playing"}
            state = state_map.get(state_val, "unknown")
        except: pass
    return {'running': self.is_running, 'recording': self.is_recording, 'state': state, 'fps': 0}

```

“”"  
Configuration management for Lumo Camera Application  
“”"  
import os  
import json  
from pathlib import Path  
from typing import Dict, Any, Optional

class Config:  
“”“Application configuration handler”“”

```auto
DEFAULT_CONFIG = {
    'camera': {
        'device': '/dev/video1', # Updated to match your detected camera
        'resolution_width': 1920,
        'resolution_height': 1200,
        'fps': 120,
        'bitrate': 50000000, # 50 Mbps
        'jpeg_quality': 100,
        'auto_white_balance': True,
        'auto_exposure': True,
    },
    'recording': {
        'segment_duration_seconds': 600, # 10 minutes
        'output_dir': str(Path.home() / 'recordings'),
        'log_dir': str(Path.home() / 'logs'),
        'motion_threshold': 30,
        'motion_pre_buffer_seconds': 10,
        'motion_post_buffer_seconds': 10,
    },
    'gui': {
        'window_title': 'Lumo Camera Recorder',
        'min_width': 800,
        'min_height': 600,
    },
    'logging': {
        'level': 'DEBUG',
        'format': '%(asctime)s - %(levelname)s - %(name)s - %(message)s',
    }
}

def __init__ (self, config_path: Optional[str] = None):
    self.config_path = config_path or str(Path.home() / '.lumo_camera_config.json')
    self._config: Dict[str, Any] = {}
    self.load()

def load(self) -> None:
    """Load configuration from file or use defaults"""
    if os.path.exists(self.config_path):
        try:
            with open(self.config_path, 'r') as f:
                loaded = json.load(f)
                self._config = self._deep_merge(self.DEFAULT_CONFIG, loaded)
        except (json.JSONDecodeError, IOError) as e:
            print(f"Warning: Could not load config, using defaults: {e}")
            self._config = self.DEFAULT_CONFIG.copy()
    else:
        self._config = self.DEFAULT_CONFIG.copy()
        self.save()

def save(self) -> None:
    """Save current configuration to file"""
    try:
        os.makedirs(os.path.dirname(self.config_path), exist_ok=True)
        with open(self.config_path, 'w') as f:
            json.dump(self._config, f, indent=2)
    except IOError as e:
        print(f"Warning: Could not save config: {e}")

def _deep_merge(self, base: Dict, override: Dict) -> Dict:
    """Deep merge two dictionaries"""
    result = base.copy()
    for key, value in override.items():
        if key in result and isinstance(result[key], dict) and isinstance(value, dict):
            result[key] = self._deep_merge(result[key], value)
        else:
            result[key] = value
    return result

def get(self, section: str, key: str, default: Any = None) -> Any:
    """Get configuration value"""
    try:
        return self._config.get(section, {}).get(key, default)
    except (TypeError, AttributeError):
        return default

def set(self, section: str, key: str, value: Any) -> None:
    """Set configuration value"""
    if section not in self._config:
        self._config[section] = {}
    self._config[section][key] = value
    self.save()

@property
def camera_device(self) -> str:
    return self.get('camera', 'device', '/dev/video1')

@property
def resolution(self) -> tuple:
    return (
        self.get('camera', 'resolution_width', 1920),
        self.get('camera', 'resolution_height', 1200)
    )

@property
def fps(self) -> int:
    return self.get('camera', 'fps', 120)

@property
def segment_duration(self) -> int:
    return self.get('recording', 'segment_duration_seconds', 600)

@property
def output_dir(self) -> str:
    return self.get('recording', 'output_dir', str(Path.home() / 'recordings'))

@property
def log_dir(self) -> str:
    return self.get('recording', 'log_dir', str(Path.home() / 'logs'))

```

“”"  
GUI - Max FPS + Telemetry + Manual Control  
“”"  
import gi  
gi.require\_version(‘Gtk’, ‘4.0’)  
gi.require\_version(‘Gdk’, ‘4.0’)  
gi.require\_version(‘Gio’, ‘2.0’)

from gi.repository import Gtk, Gdk, GLib, GdkPixbuf, Gio  
import numpy as np  
from typing import Optional  
import logging  
import os  
import cv2

class CameraGUI(Gtk.ApplicationWindow):  
def **init** (self, app, camera\_manager, recorder, logger=None):  
super(). **init** (application=app)  
self.camera\_manager = camera\_manager  
self.recorder = recorder  
self.logger = logger or logging.getLogger( **name** )  
self.\_frame\_buffer = None  
self.\_stats = None

```auto
    self.set_title("Lumo Max FPS Recorder")
    self.set_default_size(1280, 720)
    self.set_resizable(True)
    
    self._build_ui()
    
    self.camera_manager.set_sample_callback(self._on_new_frame)
    self.recorder.set_file_callback(self._on_file_completed)
    
    # Update stats every 1 second
    GLib.timeout_add(1000, self._update_telemetry)
    GLib.timeout_add(33, self._update_status) # ~30Hz UI refresh

def _build_ui(self):
    main_box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=0)
    main_box.set_margin_start(10)
    main_box.set_margin_end(10)
    main_box.set_margin_top(10)
    main_box.set_margin_bottom(10)
    self.set_child(main_box)
    
    # Header
    header = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=10)
    header.set_margin_bottom(10)
    
    title = Gtk.Label(label="LUMO MAX FPS")
    header.append(title)
    
    spacer = Gtk.Box()
    spacer.set_hexpand(True)
    header.append(spacer)
    
    # Manual Record Button
    self.btn_record = Gtk.Button(label="● RECORD (R)")
    self.btn_record.add_css_class("suggested-action")
    self.btn_record.connect('clicked', self._on_toggle_record)
    header.append(self.btn_record)
    
    btn_quit = Gtk.Button(label="Quit (Q)")
    btn_quit.connect('clicked', lambda b: self.get_application().quit())
    header.append(btn_quit)
    
    main_box.append(header)
    
    # Live View
    self.live_view = Gtk.DrawingArea()
    self.live_view.set_content_width(1280)
    self.live_view.set_content_height(720)
    self.live_view.set_hexpand(True)
    self.live_view.set_vexpand(True)
    self.live_view.set_draw_func(self._draw_frame)
    main_box.append(self.live_view)
    
    # Telemetry Bar
    telemetry_box = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=15)
    telemetry_box.set_margin_top(10)
    telemetry_box.set_margin_bottom(10)
    
    self.lbl_fps = Gtk.Label(label="FPS: --")
    self.lbl_ev = Gtk.Label(label="EV: --")
    self.lbl_iso = Gtk.Label(label="ISO: --")
    self.lbl_gain = Gtk.Label(label="Gain: --")
    self.lbl_wb = Gtk.Label(label="WB: --")
    self.lbl_status = Gtk.Label(label="Status: Idle")
    
    for lbl in [self.lbl_fps, self.lbl_ev, self.lbl_iso, self.lbl_gain, self.lbl_wb, self.lbl_status]:
        lbl.set_halign(Gtk.Align.START)
        telemetry_box.append(lbl)
    
    main_box.append(telemetry_box)
    
    # Keyboard
    ctrl = Gtk.EventControllerKey.new()
    ctrl.connect('key-pressed', self._on_key_pressed)
    self.add_controller(ctrl)

def _on_key_pressed(self, ctrl, keyval, keycode, state):
    key = chr(keycode).lower()
    if key == 'r':
        self._on_toggle_record(None)
        return True
    if key == 'q':
        self.get_application().quit()
        return True
    return False

def _on_toggle_record(self, btn):
    path = self.recorder.toggle_recording()
    if path:
        self.lbl_status.set_text(f"Status: Recording {os.path.basename(path)}")
        self.btn_record.set_label("■ STOP (R)")
        self.btn_record.remove_css_class("suggested-action")
        self.btn_record.add_css_class("destructive-action")
    else:
        self.lbl_status.set_text("Status: Stopped")
        self.btn_record.set_label("● RECORD (R)")
        self.btn_record.remove_css_class("destructive-action")
        self.btn_record.add_css_class("suggested-action")

def _on_new_frame(self, data, caps):
    try:
        w = caps.get_structure(0).get_int('width')[1]
        h = caps.get_structure(0).get_int('height')[1]
        frame = np.frombuffer(data, dtype=np.uint8).reshape((h, w, 3))
        self._frame_buffer = frame
        
        # FEED FRAME TO MOTION DETECTOR
        if self.recorder._motion_detector:
            if self.recorder._motion_detector.process_frame(frame):
                # Trigger save if motion detected
                self.recorder._on_motion_detected()
        
        self.live_view.queue_draw()
    except Exception as e:
        self.logger.error(f"Frame error: {e}")

def _draw_frame(self, widget, ctx, w, h):
    if self._frame_buffer is None:
        ctx.set_source_rgb(0.1, 0.1, 0.1)
        ctx.rectangle(0, 0, w, h)
        ctx.fill()
        return
    try:
        fh, fw, _ = self._frame_buffer.shape
        scale = min(w/fw, h/fh)
        nw, nh = int(fw*scale), int(fh*scale)
        resized = cv2.resize(self._frame_buffer, (nw, nh))
        rgba = cv2.cvtColor(resized, cv2.COLOR_RGB2RGBA)
        pix = GdkPixbuf.Pixbuf.new_from_data(rgba.tobytes(), GdkPixbuf.Colorspace.RGB, True, 8, nw, nh, nw*4, None, None)
        x, y = (w-nw)//2, (h-nh)//2
        ctx.set_source_rgb(0.1, 0.1, 0.1)
        ctx.rectangle(0, 0, w, h)
        ctx.fill()
        Gdk.cairo_set_source_pixbuf(ctx, pix, x, y)
        ctx.paint()
    except Exception as e:
        self.logger.error(f"Draw error: {e}")

def _update_telemetry(self):
    stats = self.camera_manager.get_camera_stats()
    self._stats = stats
    self.lbl_fps.set_text(f"FPS: {stats.fps:.1f}")
    self.lbl_ev.set_text(f"EV: {stats.exposure}")
    self.lbl_iso.set_text(f"ISO: {stats.iso}")
    self.lbl_gain.set_text(f"Gain: {stats.gain}")
    self.lbl_wb.set_text(f"WB: {stats.white_balance}")
    return True

def _update_status(self):
    status = self.recorder.get_recording_status()
    if status['is_recording']:
        self.lbl_status.set_text(f"Status: REC {os.path.basename(status['current_file'])}")
    else:
        self.lbl_status.set_text("Status: Idle")
    return True

def _on_file_completed(self, filepath):
    pass

```

class CameraApp(Gtk.Application):  
def **init** (self, camera\_manager, recorder, logger):  
super(). **init** (application\_id=“me.lumo.camera”, flags=Gio.ApplicationFlags.FLAGS\_NONE)  
self.camera\_manager = camera\_manager  
self.recorder = recorder  
self.logger = logger  
self.main\_window = None

```auto
def do_activate(self):
    self.main_window = CameraGUI(self, self.camera_manager, self.recorder, self.logger)
    self.main_window.present()
    if not self.camera_manager.create_live_view_pipeline():
        self.logger.error("Pipeline failed")

```

def main():

# … (main.py logic remains same, just instantiate Recorder without motion) …

pass

“”"  
Logging setup for Lumo Camera Application  
“”"  
import logging  
import os  
from pathlib import Path  
from datetime import datetime  
from typing import Optional

def setup\_logger(  
name: str,  
log\_dir: str,  
level: str = ‘DEBUG’,  
console: bool = True,  
file: bool = True  
) → logging.Logger:  
“”"  
Set up a logger with both console and file handlers

```auto
Args:
    name: Logger name
    log_dir: Directory for log files
    level: Logging level (DEBUG, INFO, WARNING, ERROR, CRITICAL)
    console: Enable console output
    file: Enable file output

Returns:
    Configured logger instance
"""
logger = logging.getLogger(name)
logger.setLevel(getattr(logging, level.upper(), logging.DEBUG))

# Clear existing handlers
logger.handlers.clear()

# Create formatter
formatter = logging.Formatter(
    '%(asctime)s - %(levelname)s - %(name)s - %(message)s',
    datefmt='%Y-%m-%d %H:%M:%S'
)

# Console handler
if console:
    ch = logging.StreamHandler()
    ch.setLevel(level)
    ch.setFormatter(formatter)
    logger.addHandler(ch)

# File handler
if file:
    os.makedirs(log_dir, exist_ok=True)
    timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
    log_file = os.path.join(log_dir, f'lumo_camera_{timestamp}.log')
    
    fh = logging.FileHandler(log_file)
    fh.setLevel(logging.DEBUG)
    fh.setFormatter(formatter)
    logger.addHandler(fh)
    
    # Also create a symlink to latest log
    latest_link = os.path.join(log_dir, 'latest.log')
    if os.path.islink(latest_link):
        os.remove(latest_link)
    os.symlink(log_file, latest_link)

return logger

```

#!/usr/bin/env python3  
import sys  
import argparse  
import logging  
from pathlib import Path

from config import Config  
from logger import setup\_logger  
from camera\_manager import CameraManager  
from motion\_detector import MotionDetector # Import detector  
from recorder import Recorder  
from gui import CameraApp  
from utils import check\_dependencies, ensure\_directories, get\_system\_info

def main():  
parser = argparse.ArgumentParser(description=“Lumo Max FPS Recorder”)  
parser.add\_argument(‘–config’, type=str, default=None, help=‘Path to config file’)  
args = parser.parse\_args()

```auto
config = Config(args.config)
logger = setup_logger(name='lumo_camera', log_dir=config.log_dir, level='INFO')

logger.info("Starting Lumo Max FPS Recorder (Manual + Motion Mode)")

missing = check_dependencies()
if missing:
    logger.error("Missing: " + ", ".join(missing))
    sys.exit(1)

ensure_directories([config.output_dir, config.log_dir])

camera_manager = CameraManager(logger)

# Initialize Motion Detector
# Threshold: 20 (sensitivity), Min Area: 1000 (ignore small noise), Cooldown: 3s
motion_detector = MotionDetector(threshold=20, min_area=1000, cooldown=3.0, logger=logger)

recorder = Recorder(
    output_dir=config.output_dir,
    logger=logger,
    camera_manager=camera_manager
)

# ENABLE MOTION DETECTION
recorder.enable_motion_detection(motion_detector)

if not camera_manager.detect_camera():
    logger.error("Camera detection failed.")
    sys.exit(1)

app = CameraApp(camera_manager, recorder, logger)

try:
    app.run(sys.argv)
except KeyboardInterrupt:
    logger.info("Interrupted by user")
finally:
    logger.info("Shutting down...")
    recorder.disable_motion_detection() # Stop detector
    camera_manager.stop_pipeline()
    logger.info("Done")

```

if **name** == “ **main** ”:  
main()

“”"  
Motion Detection Module  
“”"  
import numpy as np  
import cv2  
import threading  
import time  
from typing import Optional, Callable  
import logging

class MotionDetector:  
def **init** (self, threshold: int = 20, min\_area: int = 1000, cooldown: float = 3.0, logger: Optional[logging.Logger] = None):  
“”"  
Initialize Motion Detector.  
:param threshold: Sensitivity (0-255). Lower = more sensitive.  
:param min\_area: Minimum pixel area to count as motion.  
:param cooldown: Seconds to wait between detections.  
“”"  
self.threshold = threshold  
self.min\_area = min\_area  
self.cooldown = cooldown  
self.logger = logger or logging.getLogger( **name** )

```auto
    self._is_running = False
    self._lock = threading.Lock()
    self._previous_frame: Optional[np.ndarray] = None
    self._motion_callback: Optional[Callable] = None
    self._last_motion_time = 0.0
    
def start(self) -> None:
    with self._lock:
        self._is_running = True
        self._previous_frame = None
        self.logger.info("Motion detector started")

def stop(self) -> None:
    with self._lock:
        self._is_running = False
        self._previous_frame = None
        self.logger.info("Motion detector stopped")

def process_frame(self, frame: np.ndarray) -> bool:
    """Returns True if motion is detected and cooldown passed."""
    if not self._is_running:
        return False
    
    with self._lock:
        # Convert to grayscale
        gray = cv2.cvtColor(frame, cv2.COLOR_RGB2GRAY)
        blurred = cv2.GaussianBlur(gray, (21, 21), 0)
        
        if self._previous_frame is None:
            self._previous_frame = blurred
            return False
        
        # Diff
        frame_delta = cv2.absdiff(self._previous_frame, blurred)
        _, thresh = cv2.threshold(frame_delta, self.threshold, 255, cv2.THRESH_BINARY)
        thresh = cv2.dilate(thresh, None, iterations=2)
        
        contours, _ = cv2.findContours(thresh, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)
        
        self._previous_frame = blurred
        
        for contour in contours:
            if cv2.contourArea(contour) >= self.min_area:
                current_time = time.time()
                if current_time - self._last_motion_time >= self.cooldown:
                    self._last_motion_time = current_time
                    return True
        return False

def set_motion_callback(self, callback: Callable) -> None:
    self._motion_callback = callback

```

“”"  
Recorder - Manual + Motion Mode  
“”"  
import os  
import time  
import threading  
from datetime import datetime  
from typing import Optional, Callable  
import logging

class Recorder:  
def **init** (self, output\_dir: str, logger=None, camera\_manager=None):  
self.output\_dir = output\_dir  
self.logger = logger or logging.getLogger( **name** )  
self.camera\_manager = camera\_manager

```auto
    self._is_recording = False
    self._current_file: Optional[str] = None
    self._lock = threading.Lock()
    self._file_callback: Optional[Callable[[str], None]] = None
    
    # Motion settings
    self._motion_active = False
    self._motion_detector = None
    self._motion_callback_ref = None
    
    os.makedirs(output_dir, exist_ok=True)

def enable_motion_detection(self, detector) -> None:
    """Enable automatic motion recording."""
    self._motion_detector = detector
    self._motion_active = True
    # Bind the callback
    self._motion_callback_ref = self._on_motion_detected
    self._motion_detector.set_motion_callback(self._motion_callback_ref)
    self._motion_detector.start()
    self.logger.info("Motion detection ENABLED")

def disable_motion_detection(self) -> None:
    """Disable automatic motion recording."""
    self._motion_active = False
    if self._motion_detector:
        self._motion_detector.stop()
    self.logger.info("Motion detection DISABLED")

def _on_motion_detected(self) -> None:
    """Callback triggered by MotionDetector."""
    if not self._motion_active:
        return
    # Trigger a save (same logic as manual R press)
    self._save_current_clip()

def _save_current_clip(self) -> Optional[str]:
    """Internal method to save the temp file."""
    if not self.camera_manager:
        return None
    
    timestamp = datetime.now().strftime('%Y-%m-%d_%H-%M-%S')
    filename = f"motion_{timestamp}.mkv"
    filepath = os.path.join(self.output_dir, filename)
    
    success = self.camera_manager.start_recording(filepath)
    
    if success:
        self._current_file = filepath
        self.logger.info(f"MOTION DETECTED: Saved {filename}")
        if self._file_callback:
            self._file_callback(filepath)
        return filepath
    return None

def toggle_recording(self) -> Optional[str]:
    """Manual Start/Stop (Press R)."""
    with self._lock:
        if self._is_recording:
            if self.camera_manager:
                self.camera_manager.stop_recording()
            self._is_recording = False
            self.logger.info("Manual recording stopped")
            return self._current_file
        else:
            timestamp = datetime.now().strftime('%Y-%m-%d_%H-%M-%S')
            filename = f"manual_{timestamp}.mkv"
            filepath = os.path.join(self.output_dir, filename)
            
            if self.camera_manager:
                success = self.camera_manager.start_recording(filepath)
                if success:
                    self._is_recording = True
                    self._current_file = filepath
                    self.logger.info("Manual recording started")
                    return filepath
            return None

def get_recording_status(self) -> dict:
    with self._lock:
        return {
            'is_recording': self._is_recording,
            'current_file': self._current_file
        }

# --- MISSING METHOD ADDED BELOW ---
def set_file_callback(self, callback: Callable[[str], None]) -> None:
    """Set callback for when a file is saved."""
    self._file_callback = callback
# ----------------------------------

```

---

<div class="post-metadata">

### Author: ![PiotrBujakowski](https://discourse.gstreamer.org/letter_avatar_proxy/v4/letter/p/5fc32e/32.png) [@PiotrBujakowski](https://discourse.gstreamer.org/u/PiotrBujakowski)
#### Post date: [March 27, 2026, 12:46pm UTC](https://discourse.gstreamer.org/t/issue-in-streaming-from-usb-camera/5774/2 "2026-03-27T12:46:11Z")

</div>

```auto
"""
Utility functions for Lumo Camera Application
"""
import os
import subprocess
from typing import List, Optional

def check_dependencies() -> List[str]:
    """
    Check if required system dependencies are installed
    
    Returns:
        List of missing dependencies
    """
    missing = []
    
    # Check v4l2-ctl
    try:
        subprocess.run(['which', 'v4l2-ctl'], capture_output=True, check=True)
    except subprocess.CalledProcessError:
        missing.append("v4l-utils (v4l2-ctl)")
    
    # Check GStreamer
    try:
        subprocess.run(['gst-inspect-1.0', '--version'], capture_output=True, check=True)
    except subprocess.CalledProcessError:
        missing.append("gstreamer1.0-tools")
    
    # Check Python bindings
    try:
        import gi
        gi.require_version('Gtk', '4.0')
    except (ImportError, ValueError):
        missing.append("python3-gi (GTK4)")
    
    try:
        import gi
        gi.require_version('Gst', '1.0')
    except (ImportError, ValueError):
        missing.append("gir1.2-gstreamer-1.0")
    
    try:
        import cv2
    except ImportError:
        missing.append("opencv-python")
    
    try:
        import numpy
    except ImportError:
        missing.append("numpy")
    
    return missing

def ensure_directories(base_dirs: List[str]) -> None:
    """Ensure required directories exist"""
    for d in base_dirs:
        os.makedirs(d, exist_ok=True)

def get_system_info() -> dict:
    """Get basic system information"""
    try:
        result = subprocess.run(['uname', '-a'], capture_output=True, text=True)
        kernel = result.stdout.strip()
    except:
        kernel = "Unknown"
    
    try:
        with open('/proc/cpuinfo', 'r') as f:
            cpu_info = f.read().split('\n')[0]
    except:
        cpu_info = "Unknown"
    
    return {
        'kernel': kernel,
        'cpu': cpu_info,
        'platform': 'Linux'
    }

```

---

<div class="post-metadata">

### Author: ![bilboed](https://discourse.gstreamer.org/user_avatar/discourse.gstreamer.org/bilboed/32/251_2.png) [@bilboed](https://discourse.gstreamer.org/u/bilboed)
#### Post date: [March 28, 2026, 4:12pm UTC](https://discourse.gstreamer.org/t/issue-in-streaming-from-usb-camera/5774/3 "2026-03-28T16:12:54Z")

</div>

> [@PiotrBujakowski](#):
>
> I use AI for create script for recording from USB camera.

Then use AI to figure out what’s wrong. 😤

The way you ask this is very insulting to developers/maintainers/contributors. Why should we care about your problem when you didn’t even take the time to understand **where** exactly the problem might be.

If you want actual help, you will have to demonstrate you understand the code you wrote and ask actual specific questions instead of just doing a code dump like this.