add picamera2 mock

Author: previ

coderbot/cv/camera.py

@@ -26,7 +26,7 @@
 from threading import Condition
 import numpy as np
 from picamera2 import Picamera2
-from picamera2.encoders import Encoder, MJPEGEncoder, H264Encoder
+from picamera2.encoders import MJPEGEncoder, H264Encoder
 from picamera2.outputs import FileOutput, FfmpegOutput
 
 class Camera(object):

docker/stub/Dockerfile

@@ -64,4 +64,4 @@ ADD docker/stub/start.sh /coderbot/.
 ARG CODERBOT_VERSION
 ENV CODERBOT_VERSION=${CODERBOT_VERSION}
 
-ENTRYPOINT /coderbot/start.sh
+ENTRYPOINT ["coderbot/start.sh"]

stub/picamera2/encoders.py

@@ -1,10 +1,2 @@
-class Encoder():
-    pass
-
-class MJPEGEncoder(Encoder):
-    def __init__(self, bitrate):
-        pass
-
-class H264Encoder(Encoder):
-    def __init__(self, bitrate=100000):
-        pass
+from picamera2_mock import Picamera2MJPEGEncoderMock as MJPEGEncoder
+from picamera2_mock import Picamera2H264EncoderMock as H264Encoder

stub/picamera2/outputs.py

@@ -2,8 +2,11 @@ class Output():
     pass
 
 class FileOutput(Output):
-    def __init__(self, outout):
-        pass
+    def __init__(self, output):
+        self.output = output
+
+    def outputframe(self, frame):
+        self.output.write(frame)
 
 class FfmpegOutput(Output):
     def __init__(video_filename):

test/picamera2_mock.py

@@ -9,6 +9,33 @@
 
 logger = logging.getLogger()
 
+class Picamera2EoncoderMock(threading.Thread):
+    def __init__(self, output):
+        self.output = output
+        self.image_jpeg = open('test/test_image.jpeg', 'rb').read()
+        self.exit = False
+        super().__init__()
+
+    def close(self):
+        self.output[0].close()
+
+    def stop(self):
+        self.exit = True
+
+    def run(self):
+        while(not self.exit):
+            self.output[0].outputframe(self.image_jpeg)
+            time.sleep(0.03)
+
+class Picamera2MJPEGEncoderMock(Picamera2EoncoderMock):
+    def __init__(self, output=None, quality=20):
+        super().__init__(output)
+        self.quality = quality
+
+class Picamera2H264EncoderMock(Picamera2EoncoderMock):
+    def __init__(self, output=None):
+        super().__init__(output)
+
 class Picamera2Mock(object):
     """Implements PiCamera mock class
     PiCamera is the library used to access the integrated Camera, this mock class emulates the capture functions in order to test the streamer loop.
@@ -53,35 +80,13 @@ def __init__(self, buffer, video):
     def start(self):
         pass
 
-    def start_recording(self, buffer, format, splitter_port, quality=None, bitrate=None, resize=None):
-        """mock start_recording"""
-        print(format)
-        if format == "bgra" and resize:
-            self.images[format] = cv2.resize(self.images[format], resize)
-        if format == "h264":
-            f = open("test/test.h264", "rb")
-            video = f.read()
-            f.close()
-            self.splitter_recorders[splitter_port] = self.VideoRecorder(buffer, video)
-        else:
-            self.splitter_recorders[splitter_port] = self.ImageRecorder(buffer, self.images[format])
-            self.splitter_recorders[splitter_port].start() 
-
-    def stop_recording(self, splitter_port):
-        if splitter_port < 2:
-            self.splitter_recorders[splitter_port].go = False
-            self.splitter_recorders[splitter_port].join()
-        else:
-            recorder = self.splitter_recorders[splitter_port]
-            f = open(recorder.buffer, "wb")
-            f.write(recorder.video)
-            f.close()
-
-    def start_encoder(self, encoder):
-        pass
+    def start_encoder(self, encoder, output=None):
+        encoder.start()
+        output
 
     def stop_encoder(self, encoders):
-        pass
+        for encoder in encoders:
+            encoder.stop()
 
     def capture_buffer(self):
         return self.images["bgra"]