Returning frames as PIL images

sample_scripts/multicamera_demo.py

@@ -1,34 +1,24 @@
 #!/usr/bin/env python3
+"""Reads framegrab configuration from a yaml file, creates FrameGrabber objects, and grabs images from each camera.
+Remember to adjust sample_config.yaml according to your needs.
+"""
 
 from framegrab import FrameGrabber
 import yaml
-import cv2
 
-# load the configurations from yaml
 config_path = 'sample_config.yaml'
 with open(config_path, 'r') as f:
     configs = yaml.safe_load(f)['image_sources']
 
 print('Loaded the following configurations from yaml:')
 print(configs)
 
-# Create the grabbers
 grabbers = FrameGrabber.create_grabbers(configs)
 
-while True:
-    # Get a frame from each camera
-    for camera_name, grabber in grabbers.items():
-        frame = grabber.grab()
-
-        cv2.imshow(camera_name, frame)
-
-    key = cv2.waitKey(30)
-
-    if key == ord('q'):
-        break
-
-cv2.destroyAllWindows()
+for camera_name, grabber in grabbers.items():
+    frame = grabber.grabimg()
+    frame.show()
 
 for grabber in grabbers.values():
     grabber.release()
-
+

sample_scripts/single_camera_demo.py

@@ -1,9 +1,7 @@
 #!/usr/bin/env python3
-"""Finds a single USB camera (or built-in webcam) and displays its feed in a window. 
-Press 'q' to quit.
+"""Finds a single USB camera (or built-in webcam), grabs an image and displays the image in a window.
 """
 
-import cv2
 from framegrab import FrameGrabber
 
 config = {
@@ -13,15 +11,8 @@
 
 grabber = FrameGrabber.create_grabber(config)
 
-while True:
-    frame = grabber.grab()
+frame = grabber.grabimg()
 
-    cv2.imshow('FrameGrab Single-Camera Demo', frame)
-
-    key = cv2.waitKey(30)
-    if key == ord('q'):
-        break
-
-cv2.destroyAllWindows()
+frame.show()
 
 grabber.release()

src/framegrab/grabber.py

@@ -12,6 +12,7 @@
 import cv2
 import numpy as np
 import yaml
+from PIL import Image
 
 from .unavailable_module import UnavailableModule
 
@@ -317,12 +318,24 @@ def autodiscover(warmup_delay: float = 1.0) -> dict:
 
     @abstractmethod
     def grab(self) -> np.ndarray:
-        """Read a frame from the camera, zoom and crop if called for, and then perform any camera-specific
-        postprocessing operations.
-        Returns a frame.
+        """Grabs a single frame from the configured camera device,
+        then performs post-processing operations such as cropping and zooming based
+        on the grabber's configuration.
+
+        Returns a numpy array.
         """
         pass
 
+    def grabimg(self) -> Image:
+        """Grabs a single frame from the configured camera device,
+        then performs post-processing operations such as cropping and zooming based
+        on the grabber's configuration.
+
+        Returns a PIL image.
+        """
+        frame = self.grab()[:, :, ::-1]  # convert from BGR to RGB, which PIL expects
+        return Image.fromarray(frame)
+
     def _autogenerate_name(self) -> None:
         """For generating and assigning unique names for unnamed FrameGrabber objects.
 

src/framegrab/motion.py

@@ -1,6 +1,8 @@
 import logging
+from typing import Union
 
 import numpy as np
+from PIL import Image
 
 logger = logging.getLogger(__name__)
 
@@ -36,7 +38,10 @@ def pixel_threshold(self, img: np.ndarray, threshold_val: float = None) -> bool:
                 logger.debug(f"No motion detected: {pct_hi:.3f}% < {self.pixel_pct_threshold}%")
             return False
 
-    def motion_detected(self, new_img: np.ndarray) -> bool:
+    def motion_detected(self, new_img: Union[np.ndarray, Image.Image]) -> bool:
+        if isinstance(new_img, Image.Image):
+            new_img = np.array(new_img)
+
         if self.unused:
             self.base_img = new_img
             self.base2 = self.base_img

test/test_framegrab_with_mock_camera.py

@@ -5,6 +5,8 @@
 import os
 import unittest
 from framegrab.grabber import FrameGrabber, RTSPFrameGrabber
+import numpy as np
+from PIL import Image
 
 class TestFrameGrabWithMockCamera(unittest.TestCase):
     def test_crop_pixels(self):
@@ -86,7 +88,7 @@ def test_zoom(self):
         frame = grabber.grab()
 
         grabber.release()
-
+        
         assert frame.shape == (240, 320, 3)
 
     def test_attempt_create_grabber_with_invalid_input_type(self):
@@ -157,11 +159,10 @@ def test_attempt_create_more_grabbers_than_exist(self):
         # Try to connect to another grabber, this should raise an exception because there are only 3 mock cameras available
         try:
             FrameGrabber.create_grabber({'input_type': 'mock'})
-            self.fail()
+            self.fail() # we shouldn't get here
         except ValueError:
             pass
         finally:
-            # release all the grabbers
             for grabber in grabbers.values():
                 grabber.release()
 
@@ -213,3 +214,36 @@ def test_substitute_rtsp_url_without_placeholder(self):
         new_config = RTSPFrameGrabber._substitute_rtsp_password(config)
 
         assert  new_config == config
+
+    def test_grab_returns_np_array(self):
+        """Make sure that the grab method returns a numpy array.
+        """
+        config = {
+            'input_type': 'mock',
+        }
+
+        grabber = FrameGrabber.create_grabber(config)
+
+        frame = grabber.grab()
+
+        assert isinstance(frame, np.ndarray)
+
+        grabber.release()
+
+
+    def test_grabimg_returns_pil_image(self):
+        """Make sure that the grabimg method returns a PIL Image 
+        and that the mode is 'RGB'.
+        """
+        config = {
+            'input_type': 'mock',
+        }
+
+        grabber = FrameGrabber.create_grabber(config)
+
+        frame = grabber.grabimg()
+
+        assert isinstance(frame, Image.Image)
+        assert frame.mode == 'RGB'
+
+        grabber.release()

test/test_motdet.py

@@ -1,5 +1,6 @@
 import unittest
 import numpy as np
+from PIL import Image
 from framegrab.motion import MotionDetector
 
 class TestMotionDetector(unittest.TestCase):
@@ -67,3 +68,12 @@ def test_detect_motion_with_configured_threshold(self):
         self.motion_detector.motion_detected(img1)  # Initialize base image
         self.motion_detector.motion_detected(img1)  # again to really reset
         self.assertTrue(self.motion_detector.motion_detected(img3))
+
+    def test_that_motet_can_take_pil_image_or_numpy_image(self):
+        pil_img = Image.new('RGB', (100, 100), color='red')
+        for _ in range(10):
+            self.motion_detector.motion_detected(pil_img)
+
+        numpy_img = np.full((100, 100, 3), 255, dtype=np.uint8)
+        for _ in range(10):
+            self.motion_detector.motion_detected(numpy_img)