Thomas Ales [M S E]
530 lines
16 KiB
Python
530 lines
16 KiB
Python
"""
|
|
uC480 Camera Driver
|
|
Driver for IDS/Thorlabs uEye uC480 cameras using pyueye library.
|
|
Provides camera control, live streaming, and image capture capabilities.
|
|
"""
|
|
|
|
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 contextlib import contextmanager
|
|
from typing import Optional, Tuple
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
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
|
|
self.color_mode = ueye.IS_CM_BGR8_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
|
|
self.set_exposure(10.0) # 10ms default exposure
|
|
self.set_pixel_clock(30) # 30MHz default pixel clock
|
|
self.set_framerate(30.0) # 30fps default
|
|
|
|
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
|
|
|
|
self.is_capturing = True
|
|
logger.info("Video capture started")
|
|
return True
|
|
|
|
def stop_capture(self) -> bool:
|
|
"""
|
|
Stop continuous video capture.
|
|
|
|
Returns:
|
|
True if successful, False otherwise
|
|
"""
|
|
if not self.is_capturing:
|
|
return True
|
|
|
|
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
|
|
|
|
@contextmanager
|
|
def _capture_paused(self):
|
|
"""
|
|
Context manager that temporarily stops live video while a camera
|
|
parameter is being changed, then restarts it. Many IDS cameras
|
|
return IS_CANT_COMMUNICATE_WITH_DRIVER (17) or IS_NO_SUCCESS (-1)
|
|
when gain/exposure commands are issued during active capture.
|
|
"""
|
|
with self._settings_lock:
|
|
was_capturing = self.is_capturing
|
|
if was_capturing:
|
|
ueye.is_StopLiveVideo(self.h_cam, ueye.IS_WAIT)
|
|
self.is_capturing = False
|
|
try:
|
|
yield
|
|
finally:
|
|
if was_capturing:
|
|
ret = ueye.is_CaptureVideo(self.h_cam, ueye.IS_DONT_WAIT)
|
|
if ret == ueye.IS_SUCCESS:
|
|
self.is_capturing = True
|
|
else:
|
|
logger.error(f"Failed to restart capture after settings change: {ret}")
|
|
|
|
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
|
|
frame = np.reshape(array, (self.height, self.width, self.bytes_per_pixel))
|
|
|
|
# Convert to QImage (BGR to RGB)
|
|
height, width, channel = frame.shape
|
|
bytes_per_line = self.bytes_per_pixel * width
|
|
|
|
# Convert BGR to RGB
|
|
rgb_frame = frame[:, :, ::-1].copy()
|
|
|
|
q_image = QImage(
|
|
rgb_frame.data,
|
|
width,
|
|
height,
|
|
bytes_per_line,
|
|
QImage.Format.Format_RGB888
|
|
)
|
|
|
|
# Make a copy since the numpy array will be deleted
|
|
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 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.debug(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.
|
|
|
|
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:
|
|
logger.debug(f"Framerate set to {fps}fps")
|
|
return True
|
|
else:
|
|
logger.error(f"Failed to set framerate: {ret}")
|
|
return False
|
|
|
|
def get_framerate(self) -> Optional[float]:
|
|
"""
|
|
Get current framerate.
|
|
|
|
Returns:
|
|
Framerate in fps, or None if failed
|
|
"""
|
|
if not self.is_initialized:
|
|
return None
|
|
|
|
fps = ueye.c_double()
|
|
ret = ueye.is_GetFramesPerSecond(self.h_cam, fps)
|
|
|
|
if ret == ueye.IS_SUCCESS:
|
|
return fps.value
|
|
else:
|
|
return None
|
|
|
|
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:
|
|
frame = self.camera.get_frame()
|
|
if frame is not None:
|
|
self.frame_ready.emit(frame)
|
|
else:
|
|
# Small delay on error to prevent CPU spinning
|
|
self.msleep(10)
|
|
|
|
self.camera.stop_capture()
|
|
|
|
def stop(self):
|
|
"""Stop the streaming thread."""
|
|
self.running = False
|
|
self.wait()
|