""" uC480 Camera Driver Driver for IDS/Thorlabs uEye uC480 cameras using pyueye library. Provides camera control, live streaming, and image capture capabilities. """ import glob import os import re import time import numpy as np from pyueye import ueye from PyQt6.QtCore import QThread, pyqtSignal, QObject from PyQt6.QtGui import QImage import logging import threading from typing import List, Optional, Tuple logger = logging.getLogger(__name__) IDS_USB_VENDOR_ID = "1409" def find_camera_bus_conflicts() -> List[Tuple[str, str, str]]: """Return (tty_name, bus, product) for serial adapters sharing a USB host controller with the IDS camera. A uC480 USB2 camera at high pixel clock needs nearly the whole 480 Mbit/s of its host controller. When a full-speed serial adapter (ESP32 CDC, FTDI, …) on the same controller has its port *open*, the kernel's periodic split transactions starve the camera's bulk stream — measured on this rig: 22 fps with all ports closed, <1.5 fps with either the T3R (ttyACM0) or BBD202 (ttyUSB1) port open, full recovery on close. The only real fix is plugging the camera into a port on a different controller (e.g. a USB-3/xHCI port); this check exists so the UI can say that instead of silently dropping frames. """ cam_buses = set() for vid_path in glob.glob("/sys/bus/usb/devices/*/idVendor"): try: with open(vid_path) as f: if f.read().strip() != IDS_USB_VENDOR_ID: continue with open(os.path.join(os.path.dirname(vid_path), "busnum")) as f: cam_buses.add(f.read().strip()) except OSError: continue conflicts: List[Tuple[str, str, str]] = [] if not cam_buses: return conflicts tty_paths = glob.glob("/sys/class/tty/ttyUSB*") + glob.glob("/sys/class/tty/ttyACM*") for tty_path in sorted(tty_paths): real = os.path.realpath(os.path.join(tty_path, "device")) m = re.search(r"/usb(\d+)/", real) if not m or m.group(1) not in cam_buses: continue # Walk up from the interface dir to the USB device dir for its name product = "" d = real for _ in range(6): d = os.path.dirname(d) if os.path.exists(os.path.join(d, "busnum")): try: with open(os.path.join(d, "product")) as f: product = f.read().strip() except OSError: pass break conflicts.append((os.path.basename(tty_path), m.group(1), product)) if conflicts: devs = ", ".join(f"/dev/{n} ({p})" if p else f"/dev/{n}" for n, _, p in conflicts) logger.warning( f"Camera shares USB bus {sorted(cam_buses)} with serial adapters: {devs}. " f"Opening any of these ports will collapse the camera frame rate — " f"move the camera to a port on another USB controller." ) return conflicts class UC480Camera(QObject): """ Driver class for uC480 camera. Handles initialization, configuration, and image acquisition. """ # Signals frame_ready = pyqtSignal(QImage) # Emitted when a new frame is captured error_occurred = pyqtSignal(str) # Emitted when an error occurs def __init__(self, camera_id: int = 1): """ Initialize the uC480 camera driver. Args: camera_id: Camera ID (1-based; use is_GetCameraList to find IDs) """ super().__init__() self.camera_id = camera_id # IS_ALLOW_STARTER_FW_UPLOAD (0x10000): instructs the SDK to block # inside is_InitCamera until firmware upload and USB re-enumeration # finish. Without this flag, the handle becomes invalid the moment # the device reconnects and the SDK segfaults on the very next call. self.h_cam = ueye.HIDS(camera_id | 0x10000) self.is_initialized = False self.is_capturing = False # Memory and image info self.mem_ptr = ueye.c_mem_p() self.mem_id = ueye.int() self.pitch = ueye.INT() # Camera info self.sensor_info = ueye.SENSORINFO() self.cam_info = ueye.CAMINFO() self.rect_aoi = ueye.IS_RECT() # Image dimensions self.width = 0 self.height = 0 self.bits_per_pixel = 24 # Default to 24-bit color self.bytes_per_pixel = 3 # RGB (not BGR) so get_frame() can hand the buffer straight to QImage # without a per-frame channel-reversal copy. self.color_mode = ueye.IS_CM_RGB8_PACKED # Lock to serialize parameter changes that require stopping live video self._settings_lock = threading.Lock() def initialize(self) -> bool: """ Initialize the camera and allocate memory. Returns: True if successful, False otherwise """ try: # Initialize camera. After is_ExitCamera the UI124x series # resets and re-enumerates on USB (firmware reload), so retry # for up to ~10 s if IS_CANT_OPEN_DEVICE is returned. for attempt in range(20): ret = ueye.is_InitCamera(self.h_cam, None) if ret == ueye.IS_SUCCESS: break if ret == ueye.IS_CANT_OPEN_DEVICE and attempt < 19: logger.debug(f"Camera not ready (IS_CANT_OPEN_DEVICE), retrying ({attempt+1}/20)…") time.sleep(0.5) # h_cam value is consumed by a failed init on some SDK # versions; recreate it to avoid IS_INVALID_CAMERA_HANDLE self.h_cam = ueye.HIDS(self.camera_id | 0x10000) else: logger.error(f"Failed to initialize camera: {ret}") self.error_occurred.emit(f"Failed to initialize camera: {ret}") return False # Get sensor info ret = ueye.is_GetSensorInfo(self.h_cam, self.sensor_info) if ret != ueye.IS_SUCCESS: logger.error(f"Failed to get sensor info: {ret}") self.cleanup() return False # Get camera info ret = ueye.is_GetCameraInfo(self.h_cam, self.cam_info) if ret != ueye.IS_SUCCESS: logger.error(f"Failed to get camera info: {ret}") self.cleanup() return False # Set color mode ret = ueye.is_SetColorMode(self.h_cam, self.color_mode) if ret != ueye.IS_SUCCESS: logger.error(f"Failed to set color mode: {ret}") self.cleanup() return False # Get maximum image size self.width = self.sensor_info.nMaxWidth.value self.height = self.sensor_info.nMaxHeight.value # Set Area of Interest (AOI) to maximum size self.rect_aoi.s32X = ueye.int(0) self.rect_aoi.s32Y = ueye.int(0) self.rect_aoi.s32Width = ueye.int(self.width) self.rect_aoi.s32Height = ueye.int(self.height) ret = ueye.is_AOI(self.h_cam, ueye.IS_AOI_IMAGE_SET_AOI, self.rect_aoi, ueye.sizeof(self.rect_aoi)) if ret != ueye.IS_SUCCESS: logger.error(f"Failed to set AOI: {ret}") self.cleanup() return False # Allocate image memory ret = ueye.is_AllocImageMem( self.h_cam, self.width, self.height, self.bits_per_pixel, self.mem_ptr, self.mem_id ) if ret != ueye.IS_SUCCESS: logger.error(f"Failed to allocate image memory: {ret}") self.cleanup() return False # Set active memory ret = ueye.is_SetImageMem(self.h_cam, self.mem_ptr, self.mem_id) if ret != ueye.IS_SUCCESS: logger.error(f"Failed to set active memory: {ret}") self.cleanup() return False # Get pitch (bytes per line) ret = ueye.is_GetImageMemPitch(self.h_cam, self.pitch) if ret != ueye.IS_SUCCESS: logger.error(f"Failed to get pitch: {ret}") self.cleanup() return False self.is_initialized = True logger.info(f"Camera initialized: {self.width}x{self.height}, {self.bits_per_pixel}bpp") # Set default settings. Pixel clock caps the max sensor readout # rate, which in turn caps achievable fps regardless of the # requested framerate below — use the sensor's max rather than # a hardcoded guess, since a too-low clock silently forces # is_SetFrameRate to negotiate down to a much lower actual fps. self.set_exposure(10.0) # 10ms default exposure clock_range = self.get_pixel_clock_range() if clock_range is not None: min_clock, max_clock, _ = clock_range self.set_pixel_clock(max_clock) else: self.set_pixel_clock(30) # fallback if range query fails self.set_framerate(30.0) # 30fps default (actual may be lower) return True except Exception as e: logger.error(f"Exception during camera initialization: {e}") self.error_occurred.emit(f"Exception during initialization: {e}") self.cleanup() return False def cleanup(self): """Release camera resources.""" if self.is_capturing: self.stop_capture() if self.mem_ptr: ueye.is_FreeImageMem(self.h_cam, self.mem_ptr, self.mem_id) self.mem_ptr = None if self.is_initialized: ret = ueye.is_ExitCamera(self.h_cam) self.is_initialized = False if ret == ueye.IS_SUCCESS: logger.info("Camera resources released") else: logger.error(f"is_ExitCamera failed: {ret} — camera handle may still be held by daemon") def start_capture(self) -> bool: """ Start continuous video capture. Returns: True if successful, False otherwise """ if not self.is_initialized: logger.error("Camera not initialized") return False if self.is_capturing: logger.warning("Camera already capturing") return True ret = ueye.is_CaptureVideo(self.h_cam, ueye.IS_DONT_WAIT) if ret != ueye.IS_SUCCESS: logger.error(f"Failed to start capture: {ret}") self.error_occurred.emit(f"Failed to start capture: {ret}") return False ret = ueye.is_EnableEvent(self.h_cam, ueye.IS_SET_EVENT_FRAME) if ret != ueye.IS_SUCCESS: logger.error(f"Failed to enable frame event: {ret}") self.is_capturing = True logger.info("Video capture started") return True def wait_for_frame(self, timeout_ms: int = 200) -> bool: """ Block until the camera signals that a new frame has landed in image memory (or until timeout_ms elapses). Without this, a caller polling get_frame() in a tight loop just re-reads the same still-unfinished/unchanged buffer as fast as the GIL allows — burning CPU without raising the delivered frame rate, and occasionally reading a frame mid-write (tearing). Returns: True if a new frame arrived, False on timeout/error. """ if not self.is_capturing: return False ret = ueye.is_WaitEvent(self.h_cam, ueye.IS_SET_EVENT_FRAME, timeout_ms) return ret == ueye.IS_SUCCESS def stop_capture(self) -> bool: """ Stop continuous video capture. Returns: True if successful, False otherwise """ if not self.is_capturing: return True ueye.is_DisableEvent(self.h_cam, ueye.IS_SET_EVENT_FRAME) ret = ueye.is_StopLiveVideo(self.h_cam, ueye.IS_WAIT) self.is_capturing = False # Always reset, even if the call fails if ret != ueye.IS_SUCCESS: logger.error(f"Failed to stop capture: {ret}") return False logger.info("Video capture stopped") return True def get_frame(self) -> Optional[QImage]: """ Capture a single frame from the camera. Returns: QImage if successful, None otherwise """ if not self.is_initialized: logger.error("Camera not initialized") return None # Create numpy array from image memory try: array = ueye.get_data( self.mem_ptr, self.width, self.height, self.bits_per_pixel, self.pitch, copy=True ) # Reshape to image dimensions (view, no copy — array already owns # its memory since get_data() was called with copy=True above) frame = np.reshape(array, (self.height, self.width, self.bytes_per_pixel)) height, width, channel = frame.shape bytes_per_line = self.bytes_per_pixel * width q_image = QImage( frame.data, width, height, bytes_per_line, QImage.Format.Format_RGB888 ) # Must copy: this QImage crosses threads via a queued signal, # which hands the slot a new Python wrapper around the same # frame that gets delivered after `frame` may already be GC'd — # confirmed by testing that skipping this copy corrupts pixels. return q_image.copy() except Exception as e: logger.error(f"Failed to get frame: {e}") self.error_occurred.emit(f"Failed to get frame: {e}") return None def set_exposure(self, exposure_ms: float) -> bool: """ Set camera exposure time. Args: exposure_ms: Exposure time in milliseconds Returns: True if successful, False otherwise """ if not self.is_initialized: return False exposure = ueye.c_double(exposure_ms) ret = ueye.is_Exposure( self.h_cam, ueye.IS_EXPOSURE_CMD_SET_EXPOSURE, exposure, ueye.sizeof(exposure) ) if ret == ueye.IS_SUCCESS: logger.debug(f"Exposure set to {exposure_ms}ms") return True else: logger.error(f"Failed to set exposure: {ret}") return False def get_exposure(self) -> Optional[float]: """ Get current exposure time. Returns: Exposure time in milliseconds, or None if failed """ if not self.is_initialized: return None exposure = ueye.c_double() ret = ueye.is_Exposure( self.h_cam, ueye.IS_EXPOSURE_CMD_GET_EXPOSURE, exposure, ueye.sizeof(exposure) ) if ret == ueye.IS_SUCCESS: return exposure.value else: return None def get_pixel_clock_range(self) -> Optional[Tuple[int, int, int]]: """ Query the sensor's supported pixel clock range. Returns: (min_mhz, max_mhz, increment_mhz), or None if the query failed """ if not self.is_initialized: return None clock_range = (ueye.c_uint * 3)() ret = ueye.is_PixelClock( self.h_cam, ueye.IS_PIXELCLOCK_CMD_GET_RANGE, clock_range, ueye.sizeof(clock_range) ) if ret == ueye.IS_SUCCESS: return clock_range[0].value, clock_range[1].value, clock_range[2].value else: logger.error(f"Failed to get pixel clock range: {ret}") return None def set_pixel_clock(self, pixel_clock_mhz: int) -> bool: """ Set camera pixel clock. Args: pixel_clock_mhz: Pixel clock in MHz Returns: True if successful, False otherwise """ if not self.is_initialized: return False ret = ueye.is_PixelClock( self.h_cam, ueye.IS_PIXELCLOCK_CMD_SET, ueye.c_uint(pixel_clock_mhz), ueye.sizeof(ueye.c_uint) ) if ret == ueye.IS_SUCCESS: logger.info(f"Pixel clock set to {pixel_clock_mhz}MHz") return True else: logger.error(f"Failed to set pixel clock: {ret}") return False def set_framerate(self, fps: float) -> bool: """ Set camera framerate. The SDK negotiates the requested value against the current pixel clock/exposure/AOI and may return a lower actual rate — that negotiated value is what's logged and returned, not the request, since silently trusting the request hides the real cap. Args: fps: Frames per second Returns: True if successful, False otherwise """ if not self.is_initialized: return False new_fps = ueye.c_double(fps) actual_fps = ueye.c_double() ret = ueye.is_SetFrameRate(self.h_cam, new_fps, actual_fps) if ret == ueye.IS_SUCCESS: if actual_fps.value < fps * 0.9: logger.warning( f"Requested {fps}fps but camera negotiated only " f"{actual_fps.value:.1f}fps (pixel clock/exposure/AOI-limited)" ) else: logger.info(f"Framerate set to {actual_fps.value:.1f}fps") return True else: logger.error(f"Failed to set framerate: {ret}") return False def set_gain(self, master_gain: int) -> bool: """ Set camera master gain. Args: master_gain: Gain value (0-100) Returns: True if successful, False otherwise """ if not self.is_initialized: return False if master_gain < 0 or master_gain > 100: logger.error(f"Gain value {master_gain} out of range (0-100)") return False ret = ueye.is_SetHardwareGain( self.h_cam, master_gain, ueye.IS_IGNORE_PARAMETER, ueye.IS_IGNORE_PARAMETER, ueye.IS_IGNORE_PARAMETER ) if ret == ueye.IS_SUCCESS: logger.debug(f"Master gain set to {master_gain}") return True elif ret == ueye.IS_CANT_COMMUNICATE_WITH_DRIVER: logger.error( f"Hardware gain not supported by this camera model " f"(IS_CANT_COMMUNICATE_WITH_DRIVER). " f"Consider using gain boost instead." ) return False else: logger.error(f"Failed to set gain: {ret}") return False def get_sensor_info(self) -> dict: """ Get camera sensor information. Returns: Dictionary with sensor information """ if not self.is_initialized: return {} return { 'sensor_name': self.sensor_info.strSensorName.decode('utf-8'), 'max_width': self.sensor_info.nMaxWidth.value, 'max_height': self.sensor_info.nMaxHeight.value, 'color_mode': self.sensor_info.nColorMode.value, 'pixel_size': self.sensor_info.wPixelSize.value / 100.0, # in µm } def __del__(self): """Destructor - ensure cleanup.""" self.cleanup() class CameraStreamThread(QThread): """ Thread for continuous camera frame acquisition and streaming. """ frame_ready = pyqtSignal(QImage) error_occurred = pyqtSignal(str) def __init__(self, camera: UC480Camera): """ Initialize the camera stream thread. Args: camera: UC480Camera instance """ super().__init__() self.camera = camera self.running = False def run(self): """Main thread loop for frame acquisition.""" self.running = True if not self.camera.start_capture(): self.error_occurred.emit("Failed to start camera capture") return while self.running: # Block until the camera actually has a new frame ready, instead # of re-reading (and re-copying/re-emitting) the same buffer as # fast as possible. The short timeout just bounds how quickly a # stop() request is noticed. if not self.camera.wait_for_frame(200): continue frame = self.camera.get_frame() if frame is not None: self.frame_ready.emit(frame) self.camera.stop_capture() def stop(self): """Stop the streaming thread.""" self.running = False self.wait()