Clean up camera drivers a bit

Now switching between HM01B0 and OV5640 is as simple as changing only boot.py!
RP2350 has issues capturing images correctly from OV5640 (see #22). Example 3 is funky even with 5MHz XCLK, since other things use the DMA at the same time

Author: sfe-SparkFro

cv2_drivers/cameras/cv2_camera.py

@@ -1,5 +1,5 @@
-from ulab import numpy as np
-
 class CV2_Camera():
-    def __init__(self, buffer_size):
-        self.buffer = np.zeros(buffer_size, dtype=np.uint8)
+    def __init__(self):
+        pass
+
+    # TODO: Implement common methods for all cameras

cv2_drivers/cameras/dvp_camera.py

@@ -6,10 +6,9 @@ class DVP_Camera(CV2_Camera):
     def __init__(
         self,
         i2c,
-        i2c_address,
-        buffer_size
+        i2c_address
     ):
-        super().__init__(buffer_size)
+        super().__init__()
 
         self.i2c = i2c
         self.i2c_address = i2c_address

cv2_drivers/cameras/dvp_rp2_pio.py

@@ -9,8 +9,11 @@ def __init__(
         pin_hsync,
         pin_pclk,
         pin_xclk,
+        xclk_freq,
         sm_id,
-        num_data_pins
+        num_data_pins,
+        bytes_per_frame,
+        byte_swap
     ):
         self.pin_d0 = pin_d0
         self.pin_vsync = pin_vsync
@@ -19,22 +22,22 @@ def __init__(
         self.pin_xclk = pin_xclk
         self.sm_id = sm_id
 
+        # Initialize DVP pins as inputs
         for i in range(num_data_pins):
             Pin(pin_d0+i, Pin.IN)
         Pin(pin_vsync, Pin.IN)
         Pin(pin_hsync, Pin.IN)
         Pin(pin_pclk, Pin.IN)
 
+        # Set up XCLK pin if provided
         if self.pin_xclk is not None:
             self.xclk = PWM(Pin(pin_xclk))
-            self.xclk.freq(25_000_000)
-            # self.xclk.freq(15_000_000) # Test for OV5640
-            self.xclk.duty_u16(32768)
+            self.xclk.freq(xclk_freq)
+            self.xclk.duty_u16(32768) # 50% duty cycle
 
-        self.start_pio_dma(num_data_pins)
-
-    def start_pio_dma(self, num_data_pins):
+        # Copy the PIO program
         program = self._pio_read_dvp
+
         # Mask in the GPIO pins
         program[0][0] |= self.pin_hsync & 0x1F
         program[0][1] |= self.pin_pclk & 0x1F
@@ -44,40 +47,45 @@ def start_pio_dma(self, num_data_pins):
         program[0][2] &= 0xFFFFFFE0
         program[0][2] |= num_data_pins
 
+        # Create PIO state machine to capture DVP data
         self.sm = rp2.StateMachine(
             self.sm_id,
             program,
-            in_base = self.pin_d0
+            in_base = pin_d0
         )
-        self.sm.active(1)
 
+        # Create DMA controller to transfer data from PIO to buffer
         self.dma = rp2.DMA()
         req_num = ((self.sm_id // 4) << 3) + (self.sm_id % 4) + 4
+        bytes_per_transfer = 4
         dma_ctrl = self.dma.pack_ctrl(
-            size = 2, # 0 = 8-bit, 1 = 16-bit, 2 = 32-bit
+            # 0 = 1 byte, 1 = 2 bytes, 2 = 4 bytes
+            size = {1:0, 2:1, 4:2}[bytes_per_transfer],
             inc_read = False,
             treq_sel = req_num,
-            bswap = True
-            # bswap = False # Test for OV5640
+            bswap = byte_swap
         )
         self.dma.config(
             read = self.sm,
-            count = 244 * 324 // 4,
-            # count = 240 * 320 * 2 // 4, # Test for OV5640
+            count = bytes_per_frame // bytes_per_transfer,
             ctrl = dma_ctrl
         )
 
     def active(self, active = None):
+        # If no argument is provided, return the current active state
         if active == None:
             return self.sm.active()
 
+        # Set the active state of the state machine
         self.sm.active(active)
 
+        # If active, set up the VSYNC interrupt handler
         if active:
             Pin(self.pin_vsync).irq(
                 trigger = Pin.IRQ_FALLING,
                 handler = lambda pin: self._vsync_handler()
             )
+        # If not active, disable the VSYNC interrupt handler
         else:
             Pin(self.pin_vsync).irq(
                 handler = None
@@ -100,6 +108,9 @@ def _vsync_handler(self):
         # Start the DMA
         self.dma.active(True)
 
+    # Here is the PIO program, which is configurable to mask in the GPIO pins
+    # and the number of data pins. It must be configured before the state
+    # machine is created
     @rp2.asm_pio(
             in_shiftdir = rp2.PIO.SHIFT_LEFT,
             push_thresh = 32,

cv2_drivers/cameras/hm01b0.py

@@ -184,16 +184,7 @@ def __init__(
         i2c_address = 0x24,
         num_data_pins = 1
     ):
-        super().__init__(i2c, i2c_address, (244, 324))
-
-        # for i in range(len(self.INIT_COMMANDS)):
-        #     if self.INIT_COMMANDS[i][0] == 0x3059:
-        #         if num_data_pins == 1:
-        #             self.INIT_COMMANDS[i][1] = 0x22
-        #         elif num_data_pins == 4:
-        #             self.INIT_COMMANDS[i][1] = 0x42
-        #         else:
-        #             self.INIT_COMMANDS[i][1] = 0x02
+        super().__init__(i2c, i2c_address)
 
         self.soft_reset()
         self.send_init(num_data_pins)

cv2_drivers/cameras/hm01b0_pio.py

@@ -1,5 +1,6 @@
 from .hm01b0 import HM01B0
 from .dvp_rp2_pio import DVP_RP2_PIO
+from ulab import numpy as np
 
 class HM01B0_PIO(HM01B0, DVP_RP2_PIO):
     def __init__(
@@ -10,13 +11,34 @@ def __init__(
         pin_hsync,
         pin_pclk,
         pin_xclk = None,
+        xclk_freq = 25_000_000,
         sm_id = 0,
         num_data_pins = 1,
         i2c_address = 0x24,
     ):
+        # Create the frame buffer
+        self.buffer = np.zeros((244, 324), dtype=np.uint8)
+
         # Call both parent constructors
-        HM01B0.__init__(self, i2c, i2c_address, num_data_pins)
-        DVP_RP2_PIO.__init__(self, pin_d0, pin_vsync, pin_hsync, pin_pclk, pin_xclk, sm_id, num_data_pins)
+        DVP_RP2_PIO.__init__(
+            self,
+            pin_d0,
+            pin_vsync,
+            pin_hsync,
+            pin_pclk,
+            pin_xclk,
+            xclk_freq,
+            sm_id,
+            num_data_pins,
+            bytes_per_frame = self.buffer.size,
+            byte_swap = True
+        )
+        HM01B0.__init__(
+            self,
+            i2c,
+            i2c_address,
+            num_data_pins
+        )
 
     def open(self):
         self.active(True)

cv2_drivers/cameras/ov5640.py

@@ -355,15 +355,15 @@ class OV5640(DVP_Camera):
 
     _ratio_table = [
         #  mw,   mh,  sx,  sy,   ex,   ey, ox, oy,   tx,   ty
-        [2560, 1920, 0, 0, 2623, 1951, 32, 16, 2844, 1968],  # 4x3
-        [2560, 1704, 0, 110, 2623, 1843, 32, 16, 2844, 1752],  # 3x2
-        [2560, 1600, 0, 160, 2623, 1791, 32, 16, 2844, 1648],  # 16x10
-        [2560, 1536, 0, 192, 2623, 1759, 32, 16, 2844, 1584],  # 5x3
-        [2560, 1440, 0, 240, 2623, 1711, 32, 16, 2844, 1488],  # 16x9
-        [2560, 1080, 0, 420, 2623, 1531, 32, 16, 2844, 1128],  # 21x9
-        [2400, 1920, 80, 0, 2543, 1951, 32, 16, 2684, 1968],  # 5x4
-        [1920, 1920, 320, 0, 2543, 1951, 32, 16, 2684, 1968],  # 1x1
-        [1088, 1920, 736, 0, 1887, 1951, 32, 16, 1884, 1968],  # 9x16
+        [2560, 1920,   0,   0, 2623, 1951, 32, 16, 2844, 1968],  # 4x3
+        [2560, 1704,   0, 110, 2623, 1843, 32, 16, 2844, 1752],  # 3x2
+        [2560, 1600,   0, 160, 2623, 1791, 32, 16, 2844, 1648],  # 16x10
+        [2560, 1536,   0, 192, 2623, 1759, 32, 16, 2844, 1584],  # 5x3
+        [2560, 1440,   0, 240, 2623, 1711, 32, 16, 2844, 1488],  # 16x9
+        [2560, 1080,   0, 420, 2623, 1531, 32, 16, 2844, 1128],  # 21x9
+        [2400, 1920,  80,   0, 2543, 1951, 32, 16, 2684, 1968],  # 5x4
+        [1920, 1920, 320,   0, 2543, 1951, 32, 16, 2684, 1968],  # 1x1
+        [1088, 1920, 736,   0, 1887, 1951, 32, 16, 1884, 1968],  # 9x16
     ]
 
     _pll_pre_div2x_factors = [1, 1, 2, 3, 4, 1.5, 6, 2.5, 8]
@@ -875,13 +875,10 @@ class OV5640(DVP_Camera):
     def __init__(
         self,
         i2c,
-        i2c_address = 0x3C,
-        num_data_pins = 1
+        i2c_address = 0x3C
     ):
-        super().__init__(i2c, i2c_address, (240, 320, 2))
+        super().__init__(i2c, i2c_address)
 
-        # self.soft_reset()
-        # sleep_us(1_000_000)
         self.write_list(self._sensor_default_regs)
 
         self._colorspace = self.OV5640_COLOR_RGB
@@ -950,7 +947,6 @@ def write_list(self, data):
         for i in range(len(data) // 2):
             reg = data[i * 2]
             value = data[i * 2 + 1]
-            print(i, reg, value)
             if reg == self._REG_DLY:
                 sleep_us(value)
             else:
@@ -999,16 +995,15 @@ def _set_size_and_colorspace(self) -> None:
 
         self._set_image_options()
 
-        # if self._colorspace == self.OV5640_COLOR_JPEG:
-        #     sys_mul = 200
-        #     if size < self.OV5640_SIZE_QVGA:
-        #         sys_mul = 160
-        #     if size < self.OV5640_SIZE_XGA:
-        #         sys_mul = 180
-        #     self._set_pll(False, sys_mul, 4, 2, False, 2, True, 4)
-        # else:
-        #     self._set_pll(False, 32, 1, 1, False, 1, True, 4)
-        self._set_pll(False, 32, 1, 1, False, 1, True, 4)
+        if self._colorspace == self.OV5640_COLOR_JPEG:
+            sys_mul = 200
+            if size < self.OV5640_SIZE_QVGA:
+                sys_mul = 160
+            if size < self.OV5640_SIZE_XGA:
+                sys_mul = 180
+            self._set_pll(False, sys_mul, 4, 2, False, 2, True, 4)
+        else:
+            self._set_pll(False, 32, 1, 1, False, 1, True, 4)
 
         self._set_colorspace()
 
@@ -1121,4 +1116,11 @@ def read(self, image = None):
         Returns:
             tuple: (success, frame)
         """
-        return (True, cv2.cvtColor(self.buffer, cv2.COLOR_BayerRG2BGR, image))
+        if self._colorspace == self.OV5640_COLOR_RGB:
+            return (True, cv2.cvtColor(self.buffer, cv2.COLOR_BGR5652BGR, image))
+        elif self._colorspace == self.OV5640_COLOR_GRAYSCALE:
+            return (True, cv2.cvtColor(self.buffer, cv2.COLOR_GRAY2BGR, image))
+        else:
+            NotImplementedError(
+                f"OV5640:Reading images in colorspace {self._colorspace} is not yet implemented."
+            )

cv2_drivers/cameras/ov5640_pio.py

@@ -1,5 +1,6 @@
 from .ov5640 import OV5640
 from .dvp_rp2_pio import DVP_RP2_PIO
+from ulab import numpy as np
 
 class OV5640_PIO(OV5640, DVP_RP2_PIO):
     def __init__(
@@ -10,13 +11,32 @@ def __init__(
         pin_hsync,
         pin_pclk,
         pin_xclk = None,
+        xclk_freq = 5_000_000,
         sm_id = 0,
-        num_data_pins = 8,
-        i2c_address = 0x3c,
+        i2c_address = 0x3c
     ):
+        # Create the frame buffer
+        self.buffer = np.zeros((240, 320, 2), dtype=np.uint8)
+
         # Call both parent constructors
-        DVP_RP2_PIO.__init__(self, pin_d0, pin_vsync, pin_hsync, pin_pclk, pin_xclk, sm_id, num_data_pins)
-        OV5640.__init__(self, i2c, i2c_address, num_data_pins)
+        DVP_RP2_PIO.__init__(
+            self,
+            pin_d0,
+            pin_vsync,
+            pin_hsync,
+            pin_pclk,
+            pin_xclk,
+            xclk_freq,
+            sm_id,
+            num_data_pins = 8,
+            bytes_per_frame = self.buffer.size,
+            byte_swap = False
+        )
+        OV5640.__init__(
+            self,
+            i2c,
+            i2c_address
+        )
 
     def open(self):
         self.active(True)