import os import cv2 import numpy as np import time import pprint # ====================== ВЫБОР ТИПА КАМЕР ====================== USE_RPI_CAMERAS = False # True = Raspberry Pi камеры, False = USB (Daheng/Gige) Mono8 # ============================================================== w, h = 640, 480 if USE_RPI_CAMERAS: from picamera2 import Picamera2 from libcamera import Transform, controls os.environ["LIBCAMERA_LOG_LEVELS"] = "3" _TRANSFORMS = { 0: Transform(vflip=1), # подбери под свою установку 1: Transform(), } else: import gxipy as gx # Глобальное хранение камер для USB (чтобы избежать GC на RPi) _global_usb_cams = None # ====================== ВСПОМОГАТЕЛЬНЫЕ ФУНКЦИИ ====================== def linear_contrast_stretch(image): if image.ndim != 2: return image min_val, max_val = image.min(), image.max() if max_val > min_val: stretched = (image.astype(np.float32) - min_val) * (255.0 / (max_val - min_val)) return np.clip(stretched, 0, 255).astype(np.uint8) return image # ====================== RPI КАМЕРЫ ====================== def make_rpi_cameras(): picams = [Picamera2(i) for i in range(2)] base_config = { "FrameDurationLimits": (16667, 16667), # ~60 fps "Saturation": 0, } for i, cam in enumerate(picams): config = cam.create_preview_configuration( main={"size": (w, h), "format": "YUV420"}, transform=_TRANSFORMS[i], controls=base_config ) cam.configure(config) time.sleep(0.5) for cam in picams: cam.start() time.sleep(1.0) return picams def capture_rpi(picams): reqs = [cam.capture_request() for cam in picams] frames = [] for req in reqs: arr = req.make_array("main") # YUV420 → только Y-канал (первые 2/3 высоты) y_h = arr.shape[0] * 2 // 3 y_channel = arr[:y_h, :] y_channel = linear_contrast_stretch(y_channel) frames.append(y_channel) req.release() return frames # ====================== USB (GXIPY) КАМЕРЫ ====================== def make_usb_cameras(): global _global_usb_cams device_manager = gx.DeviceManager() dev_num, dev_info_list = device_manager.update_all_device_list() if dev_num < 2: raise RuntimeError(f"Найдено только {dev_num} USB/Gige камер, нужно минимум 2!") cam1 = device_manager.open_device_by_index(1) cam2 = device_manager.open_device_by_index(2) # Настраиваем обе камеры for cam in [cam1, cam2]: cam.TriggerMode.set(0) # Непрерывный режим # Настройки для Mono8 for cam in (cam1, cam2): remote_device_feature1 = cam1.get_remote_device_feature_control() remote_device_feature2 = cam2.get_remote_device_feature_control() # Только читаем текущие значения для каждой камеры exposure_time_feature1 = remote_device_feature1.get_float_feature("ExposureTime") exposure_time_feature2 = remote_device_feature2.get_float_feature("ExposureTime") print("Камера 1 экспозиция: %s" % exposure_time_feature1.get()) print("Камера 2 экспозиция: %s" % exposure_time_feature2.get()) gain_value1 = remote_device_feature1.get_float_feature("Gain") gain_value2 = remote_device_feature2.get_float_feature("Gain") print("Камера 1 gain: %s" % gain_value1.get()) print("Камера 2 gain: %s" % gain_value2.get()) # Попробуем принудительно Mono8 (если камера поддерживает) try: cam.PixelFormat.set(gx.GxPixelFormatEntry.MONO8) except: pass pixel_format = cam1.PixelFormat.get() print("Pixel format:", pixel_format) print(cam) # Установка разрешения for cam in (cam1, cam2): cam.Width.set(w) cam.Height.set(h) cam1.stream_on() cam2.stream_on() time.sleep(1.0) # Сохраняем глобально, чтобы GC не убил на RPi _global_usb_cams = (cam1, cam2) return cam1, cam2 def capture_usb(): global _global_usb_cams if _global_usb_cams is None: raise RuntimeError("USB камеры не инициализированы!") cam1, cam2 = _global_usb_cams frames = [None, None] raw1 = cam1.data_stream[0].get_image() raw2 = cam2.data_stream[0].get_image() if raw1: img1 = raw1.get_numpy_array() if img1 is not None: img1 = linear_contrast_stretch(img1) frames[0] = cv2.cvtColor(img1, cv2.COLOR_GRAY2BGR) # для одинакового отображения if raw2: img2 = raw2.get_numpy_array() if img2 is not None: img2 = linear_contrast_stretch(img2) frames[1] = cv2.cvtColor(img2, cv2.COLOR_GRAY2BGR) return frames def cleanup_usb(): global _global_usb_cams if _global_usb_cams is None: return cam1, cam2 = _global_usb_cams cam1.stream_off() cam2.stream_off() cam1.close_device() cam2.close_device() _global_usb_cams = None # ====================== ОСНОВНОЙ ЦИКЛ ====================== def main(): cv2.startWindowThread() print(f"Запуск в режиме: {'Raspberry Pi камеры' if USE_RPI_CAMERAS else 'USB/Gige камеры (Mono8)'}") if USE_RPI_CAMERAS: cameras = make_rpi_cameras() cap_func = lambda: capture_rpi(cameras) cleanup = lambda: [cam.stop() for cam in cameras] + [cam.close() for cam in cameras] else: make_usb_cameras() # Инициализируем глобально cap_func = capture_usb # Прямая ссылка, без параметров cleanup = cleanup_usb # Прямая ссылка на cleanup try: fps_counter = 0 fps_start = time.time() while True: frames = cap_func() # Отображаем только валидные кадры display_frames = [] for i, f in enumerate(frames): if f is None: h, w = 480, 640 f = np.zeros((h, w, 3), dtype=np.uint8) cv2.putText(f, f"Cam {i+1}: NO SIGNAL", (50, 240), cv2.FONT_HERSHEY_SIMPLEX, 1, (0, 0, 255), 2) else: # Добавляем номер камеры cv2.putText(f, f"Cam {i+1}", (10, 30), cv2.FONT_HERSHEY_SIMPLEX, 0.8, (0, 255, 0), 2) display_frames.append(f) combined = np.hstack(display_frames) # FPS overlay fps_counter += 1 if time.time() - fps_start >= 1.0: fps = fps_counter / (time.time() - fps_start) fps_counter = 0 fps_start = time.time() print(f"FPS: {fps:.1f}") cv2.imshow('Cam 0 | Cam 1', combined) if cv2.waitKey(1) == ord('q'): break except KeyboardInterrupt: print("\nПрервано пользователем") finally: cleanup() cv2.destroyAllWindows() print("Камеры отключены") if __name__ == "__main__": main()