Ethan/refactor

VERSION

@@ -1 +1 @@
-0.4.13
+0.6.0-alpha

picar/ParallelTask.py

@@ -1,45 +1,128 @@
-from multiprocessing import Process, Manager, Queue
+from multiprocessing import Process, Event
+import redis
 
+import RPi.GPIO as GPIO
+from time import time, sleep
+from picamera.array import PiRGBArray
+from picamera import PiCamera
 
-class ParallelTask:
+from typing import Tuple
 
-    _process = None
 
-    _queue = None
+class UltrasonicProcess(Process):
+    """
+    The process function to read ultrasonic values
+    queue: multiprocessing.Manager.Queue = queue to store ultrasonic data in
+    ECHO_PIN: GPIO.BOARD Pin = attached to ultrasonic echo pin
+    TRIG_PIN: GPIO.BOARD Pin = attached to ultrasonic trigger pin
+    target_sample_rate: Int = target sample rate
+    """
 
-    @staticmethod
-    def get_from_queue(queue):
-        """
-        The method to flush the queues and return the most recent result
-        Args:
-        queue: multiprocessing.Manager.Queue
-        return: (data value, time read, datapoints in queue when called)
-        """
-        v = (None, None)
-        tossed_readings = 0
-        while not queue.empty():
-            tossed_readings += 1
-            # don't block if the queue is busy
-            v = queue.get_nowait()
-        # expand the queue 2-tuple (reading, time read) into a flat 3-tuple
-        return (*v, tossed_readings)
+    _db = redis.Redis(host="localhost", port=6379, db=0)
 
-    def __init__(self, funct, funct_args):
+    TRIG_TIME = 0.00001
 
-        manager = Manager()
+    sample_rate: int = 10
+    trig_pin: int = None
+    echo_pin: int = None
 
-        self._queue = manager.Queue()
+    def __init__(
+        self,
+        echo_pin: int,
+        trig_pin: int,
+        target_sample_rate: int = 10,
+    ):
+        Process.__init__(self)
 
-        self._process = Process(target=funct, args=(self._queue, *funct_args))
+        self.sample_rate = target_sample_rate
+        self.trig_pin = trig_pin
+        self.echo_pin = echo_pin
 
-        self._process.start()
+        self.kill = Event()
 
-    def get_result(self):
-        """
-        Method to return process result
-        data takes the format (reading, time read, values in queue when read)
-        """
-        # note that get_from_queue empties the queue
-        data = ParallelTask.get_from_queue(self._queue)
+    def run(self):
 
-        return data
+        start_time, end_time = (0, 0)
+
+        nextTime = time()
+
+        while not self.kill.is_set():
+            if time() > nextTime:
+                # trigger a reading
+                GPIO.output(self.trig_pin, True)
+                sleep(self.TRIG_TIME)
+                GPIO.output(self.trig_pin, False)
+
+                # find the start and end of the ultrasonic pulse
+                while GPIO.input(self.echo_pin) == 0:
+                    start_time = time()
+                while GPIO.input(self.echo_pin) == 1:
+                    end_time = time()
+
+                # Speed of sound 34300 cm/sec
+                dist = (end_time - start_time) * 34300 / 2
+                dist_time = time()
+                self._db.set("dist_ps_result", dist, time())
+                self._db.set("dist_update_time", dist_time)
+
+                nextTime += 1 / self.sample_rate
+
+    def stop(self):
+        self.kill.set()
+
+
+class CameraProcess(Process):
+    """
+    The process function to read ultrasonic values
+    queue: multiprocessing.Manager.Queue = queue to store ultrasonic data in
+    ECHO_PIN: GPIO.BOARD Pin = attached to ultrasonic echo pin
+    TRIG_PIN: GPIO.BOARD Pin = attached to ultrasonic trigger pin
+    target_sample_rate: Int = target sample rate
+    """
+
+    _db = redis.Redis(host="localhost", port=6379, db=0)
+
+    resolution: Tuple[int, int] = None
+    framerate: int = None
+    task = None
+
+    def __init__(
+        self,
+        cam_task=None,
+        resolution: Tuple[int, int] = (1280, 720),
+        framerate: int = 30,
+    ):
+        Process.__init__(self)
+
+        self.resolution = resolution
+        self.framerate = framerate
+
+        self.task = cam_task
+
+        self.kill = Event()
+
+    def run(self):
+
+        camera = PiCamera()
+        camera.resolution = self.resolution
+        camera.framerate = self.framerate
+        rawCapture = PiRGBArray(camera, size=self.resolution)
+        for frame in camera.capture_continuous(
+            rawCapture, format="bgr", use_video_port=True
+        ):
+
+            # TODO test this to make sure it sends fine
+            res = frame.array
+            if self.task is not None:
+                res = self.task(frame.array)
+            else:
+                res = bytes(res)
+            cam_time = time()
+            self._db.set("cam_ps_result", res)
+            self._db.set("cam_update_time", cam_time)
+            rawCapture.truncate(0)
+            if self.kill.is_set():
+                break
+
+    def stop(self):
+        self.kill.set()