started different cameras

camera/base_camera.py

@@ -0,0 +1,12 @@
+import abc
+import numpy as np
+
+class Camera(abc.ABC):
+    @abc.abstractmethod
+    def get_frame(self):
+        """BGR (numpy array) или None"""
+        pass
+
+    @abc.abstractmethod
+    def release(self):
+        pass

camera/factory.py

@@ -0,0 +1,10 @@
+from .web_camera import WebCamera
+from .oak_camera import OakCamera
+
+def create_camera(camera_type='web', **kwargs):
+    if camera_type == 'web':
+        return WebCamera(**kwargs)
+    elif camera_type == 'oak':
+        return OakCamera(**kwargs)
+    else:
+        raise ValueError(f"Unknown camera type: {camera_type}")

camera/oak_camera.py

@@ -0,0 +1,32 @@
+import depthai as dai
+import numpy as np
+from camera.base_camera import Camera
+
+class OakCamera(Camera):
+    def __init__(self, resolution=(640, 480), fps=30, mirror=False):
+        self.pipeline = dai.Pipeline()
+        cam = self.pipeline.create(dai.node.ColorCamera)
+        cam.setPreviewSize(resolution[0], resolution[1])
+        cam.setInterleaved(False)
+        cam.setFps(fps)
+        cam.setColorOrder(dai.ColorCameraProperties.ColorOrder.BGR)
+
+        xout = self.pipeline.create(dai.node.XLinkOut)
+        xout.setStreamName("preview")
+        cam.preview.link(xout.input)
+
+        self.device = dai.Device(self.pipeline)
+        self.q = self.device.getOutputQueue(name="preview", maxSize=4, blocking=False)
+        self.mirror = mirror
+
+    def get_frame(self):
+        in_frame = self.q.tryGet()
+        if in_frame is None:
+            return None
+        frame = in_frame.getCvFrame()
+        if self.mirror:
+            frame = cv2.flip(frame, 1)
+        return frame
+
+    def release(self):
+        self.device.close()

camera/web_camera.py

@@ -0,0 +1,20 @@
+import cv2
+from camera.base_camera import Camera
+
+class WebCamera(Camera):
+    def __init__(self, camera_id=0, mirror=False):
+        self.cap = cv2.VideoCapture(camera_id)
+        self.mirror = mirror
+        if not self.cap.isOpened():
+            raise RuntimeError("Cannot open web camera")
+
+    def get_frame(self):
+        ret, frame = self.cap.read()
+        if not ret:
+            return None
+        if self.mirror:
+            frame = cv2.flip(frame, 1)
+        return frame
+
+    def release(self):
+        self.cap.release()