Picam 2 usage

FRC_Fiducial_Tracking/April_PNP_Live.py

@@ -12,7 +12,8 @@ import math
 from networktables import NetworkTables
 import argparse
 from TagObj import TagObj
-from PiVid import PiVid
+#from PiVid import PiVid
+from Picam2Vid import Picam2Vid
 
 RAD2DEG = 180*pi
 
@@ -96,7 +97,7 @@ field_tag_coords = tag_corners(tag_coords)
 def getTagCoords(tag_id):
     return tag_coords[tag_id]
 
-cam = PiVid(camera_res).start()
+cam = Picam2Vid(camera_res)
 
 def connectionListener(connected, info):
     print(info, "; Connected=%s" % connected)

FRC_Fiducial_Tracking/Picam2Vid.py

@@ -0,0 +1,31 @@
+from picamera2 import Picamera2
+from libcamera import Transform
+from threading import Thread
+
+# class for allocating a thread to only updating the camera stream,
+# the other thread is used for detection processing
+class Picam2Vid:
+
+    def __init__(self, camera_res):
+        # RPi camera recording setup with threading crap.
+        #  For specs - https://www.raspberrypi.com/documentation/accessories/camera.html
+        self.camera = Picamera2()
+        self.resolution = camera_res
+        config = self.camera.create_video_configuration(main={"size": self.resolution, "format": "RGB888"}, transform=Transform(hflip=True), lores={"size": (640,480)}, encode='main')
+        self.camera.configure(config)
+        self.camera.set_controls({"FrameRate": 120})
+        self.frame = None
+        self.stopped = False
+        self.camera.start()
+
+    # output the frame we want
+    def read(self):
+        self.frame=self.camera.capture_array()
+        if self.stopped:
+            self.camera.stop()
+            return
+        return self.frame
+
+    # end threading
+    def stop(self):
+        self.stopped = True

FRC_Fiducial_Tracking/PicamThreadingtest

@@ -0,0 +1,10 @@
+from Picam2Vid import Picam2Vid
+import cv2
+
+camera = Picam2Vid((640, 480))
+
+while True:
+    frame = camera.read()
+
+    cv2.imshow("worknow", frame)
+    cv2.waitKey(1)

FRC_Fiducial_Tracking/picam2video.py

@@ -0,0 +1,21 @@
+from picamera2 import Picamera2
+from libcamera import Transform
+import cv2
+import time
+
+picam = Picamera2(camera_num=0)
+config = picam.create_video_configuration(main={"size": (640, 480), "format": "RGB888"},
+                                        transform=Transform(hflip=True),
+                                        lores={"size": (640,480)},
+                                        encode='main',
+                                        )
+picam.configure(config)
+picam.set_controls({"FrameRate": 120})
+picam.start()
+while True:
+    start_frame = time.time()
+    cv2.imshow("frame", picam.capture_array())
+    cv2.waitKey(1)
+    frame = 1/(time.time()-start_frame)
+    print(frame)
+

FRC_Fiducial_Tracking/picamera2test.py

@@ -0,0 +1,40 @@
+from picamera2 import Picamera2
+from libcamera import Transform
+from threading import Thread
+
+# class for allocating a thread to only updating the camera stream,
+# the other thread is used for detection processing
+class PiVid:
+
+    def __init__(self, camera_res):
+        # RPi camera recording setup with threading crap.
+        #  For specs - https://www.raspberrypi.com/documentation/accessories/camera.html
+        self.camera = Picamera2()
+        self.resolution = camera_res
+        config = self.camera.create_video_configuration(main={"size": self.resolution}, transform=Transform(hflip=True), lores={"size": (640,480)}, encode='main')
+        self.camera.configure(config)
+        self.camera.controls({"FrameRate": 120})
+        self.frame = None
+        self.stopped = False
+
+    # Start camera thread
+    def start(self):
+        self.camera.start()
+        
+        Thread(target=self.update, args=()).start()
+        return self
+
+    # update camera stream threading
+    def update(self):
+        self.frame=self.camera.capture_array()
+        if self.stopped:
+            self.camera.stop()
+            return
+
+    # output the frame we want
+    def read(self):
+        return self.frame
+
+    # end threading
+    def stop(self):
+        self.stopped = True