Camera Factory Pattern Adoption

modules/camera/base_camera.py

@@ -0,0 +1,51 @@
+"""
+Base class for camera device.
+"""
+
+import abc
+
+import numpy as np
+
+
+class BaseCameraDevice(abc.ABC):
+    """
+    Abstract class for camera device implementations.
+    """
+
+    @classmethod
+    @abc.abstractmethod
+    def create(
+        cls, width: int, height: int
+    ) -> "tuple[True, BaseCameraDevice] | tuple[False, None]":
+        """
+        Abstract create method.
+
+        width: Width of the camera.
+        height: Height of the camera.
+
+        Return: Success, camera object.
+        """
+        raise NotImplementedError
+
+    @abc.abstractmethod
+    def __init__(self, class_private_create_key: object, camera: object) -> None:
+        """
+        Abstract private constructor.
+        """
+        raise NotImplementedError
+
+    @abc.abstractmethod
+    def __del__(self) -> None:
+        """
+        Destructor. Release hardware resources.
+        """
+        raise NotImplementedError
+
+    @abc.abstractmethod
+    def run(self) -> tuple[True, np.ndarray] | tuple[False, None]:
+        """
+        Takes a picture with camera device.
+
+        Return: Success, image with shape (height, width, channels in BGR).
+        """
+        raise NotImplementedError

modules/camera/camera_factory.py

@@ -0,0 +1,35 @@
+"""
+Factory pattern for constructing camera device class at runtime.
+"""
+
+import enum
+
+from . import base_camera
+from . import camera_opencv
+from . import camera_picamera2
+
+
+class CameraOption(enum.Enum):
+    """
+    enum for type of camera object to create.
+    """
+
+    OPENCV = 0
+    PICAM2 = 1
+
+
+def create_camera(
+    camera_option: CameraOption, width: int, height: int
+) -> tuple[True, base_camera.BaseCameraDevice] | tuple[False, None]:
+    """
+    Create a camera object based off of given parameters.
+
+    Return: Success, camera device object.
+    """
+    match camera_option:
+        case CameraOption.OPENCV:
+            return camera_opencv.CameraOpenCV.create(width, height)
+        case CameraOption.PICAM2:
+            return camera_picamera2.CameraPiCamera2.create(width, height)
+
+    return False, None

modules/camera/camera_opencv.py

@@ -0,0 +1,66 @@
+"""
+OpenCV implementation of the camera wrapper.
+"""
+
+import cv2
+import numpy as np
+
+from . import base_camera
+
+
+class CameraOpenCV(base_camera.BaseCameraDevice):
+    """
+    Class for the OpenCV implementation of the camera.
+    """
+
+    __create_key = object()
+
+    @classmethod
+    def create(cls, width: int, height: int) -> "tuple[True, CameraOpenCV] | tuple[False, None]":
+        """
+        OpenCV Camera.
+
+        width: Width of the camera.
+        height: Height of the camera.
+
+        Return: Success, camera object.
+        """
+        camera = cv2.VideoCapture(0)
+        if not camera.isOpened():
+            return False, None
+
+        camera.set(cv2.CAP_PROP_FRAME_WIDTH, width)
+        camera.set(cv2.CAP_PROP_FRAME_HEIGHT, height)
+
+        set_width = camera.get(cv2.CAP_PROP_FRAME_WIDTH)
+        set_height = camera.get(cv2.CAP_PROP_FRAME_HEIGHT)
+        if set_width != width or set_height != height:
+            return False, None
+
+        return True, CameraOpenCV(cls.__create_key, camera)
+
+    def __init__(self, class_private_create_key: object, camera: cv2.VideoCapture) -> None:
+        """
+        Private constructor, use create() method.
+        """
+        assert class_private_create_key is CameraOpenCV.__create_key, "Use create() method."
+
+        self.__camera = camera
+
+    def __del__(self) -> None:
+        """
+        Destructor. Release hardware resources.
+        """
+        self.__camera.release()
+
+    def run(self) -> tuple[True, np.ndarray] | tuple[False, None]:
+        """
+        Takes a picture with OpenCV camera.
+
+        Return: Success, image with shape (height, width, channels in BGR).
+        """
+        result, image_data = self.__camera.read()
+        if not result:
+            return False, None
+
+        return True, image_data

modules/camera/camera_picamera2.py

@@ -0,0 +1,74 @@
+"""
+Picamera2 implementation of the camera wrapper.
+"""
+
+import numpy as np
+
+# Picamera2 library only exists on Raspberry Pi
+try:
+    import picamera2
+except ImportError:
+    pass
+
+from . import base_camera
+
+# TODO: pass in as constructor parameter
+CAMERA_TIMEOUT = 1
+
+
+class CameraPiCamera2(base_camera.BaseCameraDevice):
+    """
+    Class for the Picamera2 implementation of the camera.
+    """
+
+    __create_key = object()
+
+    @classmethod
+    def create(cls, width: int, height: int) -> "tuple[True, CameraPiCamera2] | tuple[False, None]":
+        """
+        Picamera2 Camera.
+
+        width: Width of the camera.
+        height: Height of the camera.
+
+        Return: Success, camera object.
+        """
+        try:
+            camera = picamera2.Picamera2()
+
+            config = camera.create_still_configuration(
+                {"size": (width, height), "format": "RGB888"}
+            )
+            camera.configure(config)
+            camera.start()
+            return True, CameraPiCamera2(cls.__create_key, camera)
+        except RuntimeError:
+            return False, None
+
+    def __init__(self, class_private_create_key: object, camera: "picamera2.Picamera2") -> None:
+        """
+        Private constructor, use create() method.
+        """
+        assert class_private_create_key is CameraPiCamera2.__create_key, "Use create() method."
+
+        self.__camera = camera
+
+    def __del__(self) -> None:
+        """
+        Destructor. Release hardware resources.
+        """
+        self.__camera.close()
+
+    def run(self) -> tuple[True, np.ndarray] | tuple[False, None]:
+        """
+        Takes a picture with Picamera2 camera.
+
+        Return: Success, image with shape (height, width, channels in BGR).
+        """
+        try:
+            # CAMERA_TIMEOUT seconds before raising TimeoutError
+            image_data = self.__camera.capture_array(wait=CAMERA_TIMEOUT)
+        except TimeoutError:
+            return False, None
+
+        return True, image_data

tests/integration/test_camera.py → tests/integration/test_camera_opencv.py

@@ -1,34 +1,38 @@
 """
-Test camera physically.
+Test OpenCV camera physically.
 """
 
 import pathlib
 
 import cv2
 
-from modules.camera.camera_device import CameraDevice
+from modules.camera import camera_factory
 
 
+# TODO: Add camera logging
 IMAGE_LOG_PREFIX = pathlib.Path("logs", "test_log_image")
 
 
 def main() -> int:
     """
     Main function.
     """
-    device = CameraDevice(0, 100, str(IMAGE_LOG_PREFIX))
+    result, device = camera_factory.create_camera(camera_factory.CameraOption.OPENCV, 640, 480)
+    if not result:
+        print("OpenCV camera creation error.")
+        return -1
 
     IMAGE_LOG_PREFIX.parent.mkdir(parents=True, exist_ok=True)
 
     while True:
-        result, image = device.get_image()
+        result, image = device.run()
         if not result:
             print("ERROR")
             continue
 
         print(image.shape)
 
-        cv2.imshow("Camera", image)
+        cv2.imshow("OpenCV camera", image)
 
         # Delay for 1 ms
         if cv2.waitKey(1) & 0xFF == ord("q"):

tests/integration/test_camera_picamera2.py

@@ -0,0 +1,49 @@
+"""
+Test Picamera2 camera physically.
+"""
+
+import pathlib
+
+import cv2
+
+from modules.camera import camera_factory
+
+
+# TODO: Add camera logging
+IMAGE_LOG_PREFIX = pathlib.Path("logs", "test_log_image")
+
+
+def main() -> int:
+    """
+    Main function.
+    """
+    result, device = camera_factory.create_camera(camera_factory.CameraOption.PICAM2, 640, 480)
+    if not result:
+        print("Picamera2 camera creation error.")
+        return -1
+
+    IMAGE_LOG_PREFIX.parent.mkdir(parents=True, exist_ok=True)
+
+    while True:
+        result, image = device.run()
+        if not result:
+            print("ERROR")
+            continue
+
+        print(image.shape)
+
+        cv2.imshow("Picamera2 camera", image)
+
+        # Delay for 1 ms
+        if cv2.waitKey(1) & 0xFF == ord("q"):
+            break
+
+    return 0
+
+
+if __name__ == "__main__":
+    result_main = main()
+    if result_main < 0:
+        print(f"ERROR: Status code: {result_main}")
+
+    print("Done!")