feat: working version with ov cameras

Author: electricalgorithm

core/strategies/eye/base_eye_strategy.py

@@ -21,10 +21,21 @@ def get_detector(self) -> BaseDetectorStrategy:
         """This method returns the detector strategy."""
 
     @abstractmethod
-    def check_if_detected(self) -> EyeStrategyResult:
-        """This method checks if there are any protectors around."""
+    def get_frame(self) -> ndarray:
+        """This method returns the frame from the camera."""
 
     def _detect_humans(self, frame: ndarray) -> DetectorResult:
         """This method checks if there is a person in front of the camera."""
         return self.get_detector().detect_humans(frame)
+    
+    def check_if_detected(self) -> EyeStrategyResult:
+        """This method checks if there are any protectors around."""
+        # Get the frame from the camera.
+        frame = self.get_frame()
+        # Detect humans in the frame.
+        result = self._detect_humans(frame)
+
+        if result.human_found:
+            return EyeStrategyResult(image=frame, result=True)
+        return EyeStrategyResult(image=frame, result=False)
 

core/strategies/eye/camera_strategy.py

@@ -0,0 +1,44 @@
+"""
+The Camera strategy for eye strategies.
+"""
+import logging
+
+import cv2
+import numpy
+from picamera2 import Picamera2
+
+from core.strategies.detectors.base_detector_strategy import BaseDetectorStrategy
+from core.strategies.eye.base_eye_strategy import BaseEyeStrategy
+
+# Add logging support.
+logger = logging.getLogger(__name__)
+
+
+class PiCameraStrategy(BaseEyeStrategy):
+    """
+    The camera strategy for eye strategies.
+    """
+    def __init__(self):
+        self._detector = None
+
+    # Interface methods.
+    def set_detector(self, detector: BaseDetectorStrategy) -> None:
+        """This method sets the detector strategy."""
+        self._detector = detector
+
+    def get_detector(self) -> BaseDetectorStrategy:
+        """This method returns the detector strategy."""
+        return self._detector
+
+    def get_frame(self) -> numpy.ndarray:
+        """This method returns the frame from the camera."""
+        # Internal attributes
+        picam2 = Picamera2()
+        picam2.configure(picam2.create_preview_configuration(
+            main={"format": 'XRGB8888', "size": (640, 480)}
+        ))
+        picam2.start()
+        frame = picam2.capture_array()
+        picam2.close()
+        del picam2
+        return frame

core/strategies/eye/picamera_strategy.py

@@ -0,0 +1,44 @@
+"""
+The Camera strategy for eye strategies.
+"""
+import logging
+
+import cv2
+import numpy
+
+from core.strategies.detectors.base_detector_strategy import BaseDetectorStrategy
+from core.strategies.eye.base_eye_strategy import BaseEyeStrategy
+
+# Add logging support.
+logger = logging.getLogger(__name__)
+
+
+class UsbCameraStrategy(BaseEyeStrategy):
+    """
+    The camera strategy for eye strategies.
+    """
+    def __init__(self, camera_id: int = 0):
+        self._camera_id = camera_id
+        self._detector = None
+
+    # Interface methods.
+    def set_detector(self, detector: BaseDetectorStrategy) -> None:
+        """This method sets the detector strategy."""
+        self._detector = detector
+
+    def get_detector(self) -> BaseDetectorStrategy:
+        """This method returns the detector strategy."""
+        return self._detector
+
+    def get_frame(self) -> numpy.ndarray:
+        """This method returns the frame from the camera."""
+        # Create a camera object.
+        camera = cv2.VideoCapture(self._camera_id)
+        # Set the camera resolution.
+        camera.set(3, 640)
+        camera.set(4, 480)
+        # Read the frame from the camera.
+        _, frame = camera.read()
+        # Release the camera.
+        camera.release()
+        return frame

core/strategies/eye/usbcamera_strategy.py

@@ -6,7 +6,8 @@
 from core.observers.subject.wifi_subject import WiFiSubject
 from core.observers.observer.hss_observer import HomeSecuritySystemObserver
 from core.strategies.wifi.ipaddress_strategy import IpAddressStrategy
-from core.strategies.eye.camera_strategy import CameraStrategy
+from core.strategies.eye.usbcamera_strategy import UsbCameraStrategy
+from core.strategies.eye.picamera_strategy import PiCameraStrategy
 from core.strategies.notifier.whatsapp_strategy import WhatsappStrategy
 from core.strategies.detectors.hog_descriptor_strategy import HogDescriptorStrategy
 from core.utils.datatypes import WhatsappReciever, Protector
@@ -28,11 +29,11 @@ def main():
     """
     # Create a WhatsApp notifier.
     whatsapp_notifier = WhatsappStrategy()
-    whatsapp_notifier.add_reciever(WhatsappReciever("Gokhan", "+90555555555", "xxxxxx"))
+    whatsapp_notifier.add_reciever(WhatsappReciever("Gokhan", "tel_no", "api_key"))
 
     # Create a Protector within IpAddressStrategy.
     ip_address_strategy = IpAddressStrategy()
-    ip_address_strategy.add_protector(Protector("Gokhan_iPhone", "192.168.1.65"))
+    ip_address_strategy.add_protector(Protector("Gokhan_iPhone", "tel_ip"))
 
     # Create observer.
     hss_observer = HomeSecuritySystemObserver()
@@ -46,7 +47,7 @@ def main():
 
     # Run subjects.
     wifi_subject.run(ip_address_strategy)
-    camera = CameraStrategy(0)
+    camera = PiCameraStrategy()
     camera.set_detector(HogDescriptorStrategy())
     eye_subject.run(camera)