blink and pulse

picozero/picozero.py

@@ -3,13 +3,75 @@
 
 class PWMChannelAlreadyInUse(Exception):
     pass
+
+class ValueChange:
+    """
+    Internal class to control the value of an output device 
+
+    :param OutputDevice output_device:
+        The OutputDevice object you wish to change the value of
+
+    :param generator:
+        A generator function which yields a 2d list of
+        ((value, seconds), *).
+
+        The output_device's value will be set for the number of
+        seconds.
 
+    :param int n:
+        The number of times to repeat the sequence. If None, the
+        sequence will repeat forever. 
+
+    :param bool wait:
+        If True the ValueChange object will block (wait) until
+        the sequence has completed.
+    """
+    def __init__(self, output_device, generator, n, wait):
+        self._output_device = output_device
+        self._generator = generator
+        self._n = n
+
+        self._gen = self._generator()
+
+        self._timer = Timer()
+        self._running = True
+        self._set_value()
+
+        while wait and self._running:
+            sleep(0.001)
+
+    def _set_value(self, timer_obj=None):
+
+        try:
+            if self._running:
+                next_seq = next(self._gen)
+                value, seconds = next_seq
+
+                self._output_device._write(value)            
+                self._timer.init(period=int(seconds * 1000), mode=Timer.ONE_SHOT, callback=self._set_value)
+
+        except StopIteration:
+
+            self._n = self._n - 1 if self._n is not None else None
+            if self._n == 0:
+                # its the end, set the value to 0 and stop running
+                self._output_device.value = 0
+                self._running = False
+            else:
+                # recreate the generator and start again
+                self._gen = self._generator()
+                self._set_value()
+
+    def stop(self):
+        self._running = False
+        self._timer.deinit()
+
 class OutputDevice:
 
     def __init__(self, active_high=True, initial_value=False):
         self.active_high = active_high
         self._write(initial_value)
-        self._timer = None
+        self._value_changer = None
 
     @property
     def active_high(self):
@@ -29,20 +91,19 @@ def value(self):
 
     @value.setter
     def value(self, value):
+        self._stop_change()
         self._write(value)
 
     def on(self):
         """
         Turns the device on.
         """
-        self._stop_blink()
         self.value = 1
 
     def off(self):
         """
         Turns the device off.
         """
-        self._stop_blink()
         self.value = 0
 
     @property
@@ -61,30 +122,36 @@ def toggle(self):
         else:
             self.on()
 
-    def blink(self, time=1):
+    def blink(self, on_time=1, off_time=None, n=None, wait=False):
         """
         Make the device turn on and off repeatedly.
 
-        :param float time:
-            The length of time in seconds between on and off. Defaults to 1.
+        :param float on_time:
+            The length of time in seconds the device will be on. Defaults to 1.
+
+        :param float off_time:
+            The length of time in seconds the device will be off. Defaults to 1.
+
+        :param int n:
+            The number of times to repeat the blink operation. If None is 
+            specified, the device will continue blinking forever. The default
+            is None.
         """
-        self._stop_blink()
-        self._timer = Timer()
-        self._timer.init(period=int(time * 1000), mode=Timer.PERIODIC, callback=self._blink_callback)
+        off_time = on_time if off_time is None else off_time
 
-    def _blink_callback(self, timer_obj):
-        if self.is_active:
-            self.value = 0
-        else:
-            self.value = 1
-
-    def _stop_blink(self):
-        if self._timer is not None:
-            self._timer.deinit()
-            self._timer = None
+        self.off()
+        self._start_change(lambda : iter([(1,on_time), (0,off_time)]), n, wait)
+
+    def _start_change(self, generator, n, wait):
+        self._value_changer = ValueChange(self, generator, n, wait)
+    
+    def _stop_change(self):
+        if self._value_changer is not None:
+            self._value_changer.stop()
+            self._value_changer = None
 
     def __del__(self):
-        self._stop_blink()
+        self._stop_change()
 
 class DigitalOutputDevice(OutputDevice):
     def __init__(self, pin, active_high=True, initial_value=False):
@@ -192,6 +259,88 @@ def _write(self, value):
 
     def _read(self):
         return 1 if super()._read() > 0 else 0
+
+    def blink(self, on_time=1, off_time=None, fade_in_time=0, fade_out_time=None, n=None, wait=False, fps=25):
+        """
+        Make the device turn on and off repeatedly.
+
+        :param float on_time:
+            The length of time in seconds the device will be on. Defaults to 1.
+
+        :param float off_time:
+            The length of time in seconds the device will be off. If `None`, 
+            it will be the same as ``on_time``. Defaults to `None`.
+
+        :param float fade_in_time:
+            The length of time in seconds to spend fading in. Defaults to 0.
+
+        :param float fade_out_time:
+            The length of time in seconds to spend fading out. If `None`,
+            it will be the same as ``fade_in_time``. Defaults to `None`.
+
+        :param int n:
+            The number of times to repeat the blink operation. If `None`, the 
+            device will continue blinking forever. The default is `None`.
+
+        :param int fps:
+           The frames per second that will be used to calculate the number of
+           steps between off/on states when fading. Defaults to 25.
+        """    
+        self.off()
+
+        off_time = on_time if off_time is None else off_time
+        fade_out_time = fade_in_time if fade_out_time is None else fade_out_time
+
+        def blink_generator():
+            if fade_in_time > 0:
+                for s in [
+                    (i * (1 / fps) / fade_in_time, 1 / fps)
+                    for i in range(int(fps * fade_in_time))
+                    ]:
+                    yield s
+
+            if on_time > 0:
+                yield (1, on_time)
+
+            if fade_out_time > 0:
+                for s in [
+                    (1 - (i * (1 / fps) / fade_out_time), 1 / fps)
+                    for i in range(int(fps * fade_out_time))
+                    ]:
+                    yield s
+
+            if off_time > 0:
+                 yield (0, off_time)
+
+        self._start_change(blink_generator, n, wait)
+
+    def pulse(self, fade_in_time=1, fade_out_time=None, n=None, wait=False, fps=25):
+        """
+        Make the device pulse on and off repeatedly.
+
+        :param float fade_in_time:
+            The length of time in seconds the device will take to turn on.
+            Defaults to 1.
+
+        :param float fade_out_time:
+           The length of time in seconds the device will take to turn off.
+           Defaults to 1.
+
+        :param int fps:
+           The frames per second that will be used to calculate the number of
+           steps between off/on states. Defaults to 25.
+
+        :param int n:
+           The number of times to pulse the LED. If None the LED will pulse
+           forever. Defaults to None.
+
+        :param bool wait:
+           If True the method will block until the LED stops pulsing. If False
+           the method will return and the LED is will pulse in the background.
+           Defaults to False.
+
+        """
+        self.blink(on_time=0, off_time=0, fade_in_time=fade_in_time, fade_out_time=fade_out_time, n=n, wait=wait, fps=fps)
 
 # factory for returning an LED
 def LED(pin, use_pwm=True, active_high=True, initial_value=False):
@@ -358,6 +507,8 @@ def __del__(self):
         super().__del__()
 
     def _write(self, value):
+        if type(value) is not tuple:
+            value = (value, ) * 3       
         for led, v in zip(self._leds, value):
             led.brightness = v
 
@@ -367,6 +518,7 @@ def value(self):
 
     @value.setter
     def value(self, value):
+        self._stop_change()
         self._write(value)
 
     @property
@@ -419,9 +571,6 @@ def blue(self, value):
     def on(self):
         self.value = (1, 1, 1)
 
-    def off(self):
-        self.value = (0, 0, 0)
-
     def invert(self):
         r, g, b = self.value
         self.value = (1 - r, 1 - g, 1 - b)
@@ -432,6 +581,83 @@ def toggle(self):
         else:
             self._last = self.value 
             self.value = (0, 0, 0)
+
+    def blink(self, on_times=1, fade_times=0, colors=((1, 0, 0), (0, 1, 0), (0, 0, 1)), n=None, wait=False, fps=25):
+
+        self.off()
+
+        if type(on_times) is not tuple:
+            on_times = (on_times, ) * len(colors)
+        if type(fade_times) is not tuple:
+            fade_times = (fade_times, ) * len(colors)
+        # If any value is above zero then treat all as 0-255 values
+        if any(v > 1 for v in sum(colors, ())):
+            colors = tuple(tuple(self._from_255(v) for v in t) for t in colors)
+
+        def blink_generator():
+
+            # Define a linear interpolation between
+            # off_color and on_color
+
+            lerp = lambda t, fade_in, color1, color2: tuple(
+                (1 - t) * off + t * on
+                if fade_in else
+                (1 - t) * on + t * off
+                for off, on in zip(color2, color1)
+                )
+
+            for c in range(len(colors)):
+                if fade_times[c] > 0:
+                    for i in range(int(fps * fade_times[c])):
+                        v = lerp(i * (1 / fps) / fade_times[c], True, colors[(c + 1) % len(colors)], colors[c])
+                        t = 1 / fps       
+                        yield (v, t)
+
+                if on_times[c] > 0:
+                    yield (colors[c], on_times[c])
+
+        self.off()
+        self._start_change(blink_generator, n, wait)
+
+    def pulse(self, fade_times=1, colors=((0, 0, 0), (1, 0, 0), (0, 0, 0), (0, 1, 0), (0, 0, 0), (0, 0, 1)), n=None, wait=False, fps=25):
+        """
+        Make the device fade in and out repeatedly.
+        :param float fade_in_times:
+            Number of seconds to spend fading in. Defaults to 1.
+        :param float fade_out_time:
+            Number of seconds to spend fading out. Defaults to 1.
+        :type on_color: ~colorzero.Color or tuple
+        :param on_color:
+            The color to use when the LED is "on". Defaults to white.
+        :type off_color: ~colorzero.Color or tuple
+        :param off_color:
+            The color to use when the LED is "off". Defaults to black.
+        :type n: int or None
+        :param n:
+            Number of times to pulse; :data:`None` (the default) means forever.
+        """
+        on_times = 0
+        self.blink(on_times, fade_times, colors, n, wait, fps)
+
+    def cycle(self, fade_times=1, colors=((1, 0, 0), (0, 1, 0), (0, 0, 1)), n=None, wait=False, fps=25):
+        """
+        Make the device fade in and out repeatedly.
+        :param float fade_in_time:
+            Number of seconds to spend fading in. Defaults to 1.
+        :param float fade_out_time:
+            Number of seconds to spend fading out. Defaults to 1.
+        :type on_color: ~colorzero.Color or tuple
+        :param on_color:
+            The color to use when the LED is "on". Defaults to white.
+        :type off_color: ~colorzero.Color or tuple
+        :param off_color:
+            The color to use when the LED is "off". Defaults to black.
+        :type n: int or None
+        :param n:
+            Number of times to pulse; :data:`None` (the default) means forever.
+        """
+        on_times = 0
+        self.blink(on_times, fade_times, colors, n, wait, fps)
 
 class AnalogInputDevice():
     def __init__(self, pin, active_high=True, threshold=0.5):
@@ -552,3 +778,4 @@ def Speaker(pin, use_tones=True, active_high=True, initial_value=False):
         return PWMBuzzer(pin, freq=440, active_high=active_high, initial_value=initial_value)
     else:
         return Buzzer(pin, active_high=active_high, initial_value=initial_value)
+