Refactor RGBLEDBoard tests and update docstring

bennuttall · bennuttall · commit 781ef0aa7261 · 2018-09-24T21:51:03.000+01:00
diff --git a/gpiozero/boards.py b/gpiozero/boards.py
@@ -1057,11 +1057,11 @@ def __init__(self, pwm=False, initial_value=False, pin_factory=None):
 
 class RGBLEDBoard(CompositeOutputDevice):
     """
-    Extends :class:`CompositeOutputDevice`. and represents a generic RGBLED
-    board or collection of LEDs.
+    Extends :class:`CompositeOutputDevice`. and represents a generic RGB LED
+    board or collection of RGB LEDs.
 
-    The following example turns on all the LEDs on a board containing 5 LEDs
-    attached to GPIO pins 2 through 6::
+    The following example creates a device containing three RGB LEDs and turns
+    them on red, green and blue respectfully:
 
         from gpiozero import RGBLEDBoard
 
@@ -1080,7 +1080,7 @@ def __init__(self, *args, **kwargs):
         order = kwargs.pop('_order', None)
 
         if len(args) == 0 and len(kwargs) == 0:
-            raise ValueError(
+            raise GPIOPinMissing(
                 'at least 1 sequence of RGB pins must be provided')
         else:
             try:
@@ -1089,7 +1089,7 @@ def __init__(self, *args, **kwargs):
                 for arg in kwargs.values():
                     r, g, b = arg
             except:
-                raise ValueError(
+                raise GPIOPinMissing(
                     'pins must be provided as a tuple of 3-tuples '
                     'e.g. RGBLEDBoard((2, 3, 4), (5, 6, 7)')
 
@@ -1152,6 +1152,60 @@ def toggle(self, *args):
         else:
             super(RGBLEDBoard, self).toggle()
 
+    def blink(self, on_time=1, off_time=1, fade_in_time=0, fade_out_time=0,
+              on_color=(1, 1, 1), off_color=(0, 0, 0), n=None, background=True):
+        """
+        Make all the LEDs in the device turn on and off repeatedly.
+
+        :param float on_time:
+            Number of seconds on. Defaults to 1 second.
+
+        :param float off_time:
+            Number of seconds off. Defaults to 1 second.
+
+        :param float fade_in_time:
+            Number of seconds to spend fading in. Defaults to 0. Must be 0 if
+            ``pwm`` was ``False`` when the class was constructed
+            (:exc:`ValueError` will be raised if not).
+
+        :param float fade_out_time:
+            Number of seconds to spend fading out. Defaults to 0. Must be 0 if
+            ``pwm`` was ``False`` when the class was constructed
+            (:exc:`ValueError` will be raised if not).
+
+        :param tuple on_color:
+            The color to use when the LED is "on". Defaults to white.
+
+        :param tuple off_color:
+            The color to use when the LED is "off". Defaults to black.
+
+        :param int n:
+            Number of times to blink; ``None`` (the default) means forever.
+
+        :param bool background:
+            If ``True`` (the default), start a background thread to continue
+            blinking and return immediately. If ``False``, only return when the
+            blink is finished (warning: the default value of *n* will result in
+            this method never returning).
+        """
+        if isinstance(self._leds[0]._leds[0], LED):
+            if fade_in_time:
+                raise ValueError('fade_in_time must be 0 with non-PWM RGBLEDs')
+            if fade_out_time:
+                raise ValueError('fade_out_time must be 0 with non-PWM RGBLEDs')
+        self._stop_blink()
+        self._blink_thread = GPIOThread(
+            target=self._blink_device,
+            args=(
+                on_time, off_time, fade_in_time, fade_out_time,
+                on_color, off_color, n
+            )
+        )
+        self._blink_thread.start()
+        if not background:
+            self._blink_thread.join()
+            self._blink_thread = None
+
     def _stop_blink(self, led=None):
         if led is None:
             if self._blink_thread:
@@ -1161,6 +1215,40 @@ def _stop_blink(self, led=None):
             with self._blink_lock:
                 self._blink_leds.remove(led)
 
+    def _blink_device(self, on_time, off_time, fade_in_time, fade_out_time, n,
+                      on_color, off_color, nfps=25):
+        sequence = []
+        if fade_in_time > 0:
+            sequence += [
+                (i * (1 / fps) / fade_in_time, 1 / fps)
+                for i in range(int(fps * fade_in_time))
+                ]
+        sequence.append((1, on_time))
+        if fade_out_time > 0:
+            sequence += [
+                (1 - (i * (1 / fps) / fade_out_time), 1 / fps)
+                for i in range(int(fps * fade_out_time))
+                ]
+        sequence.append((0, off_time))
+        sequence = (
+                cycle(sequence) if n is None else
+                chain.from_iterable(repeat(sequence, n))
+                )
+        with self._blink_lock:
+            self._blink_leds = list(self.leds)
+            for led in self._blink_leds:
+                if led._controller not in (None, self):
+                    led._controller._stop_blink(led)
+                led._controller = self
+        for value, delay in sequence:
+            with self._blink_lock:
+                if not self._blink_leds:
+                    break
+                for led in self._blink_leds:
+                    led._write(value)
+            if self._blink_thread.stopping.wait(delay):
+                break
+
 
 class TrafficLightsBuzzer(CompositeOutputDevice):
     """
diff --git a/tests/test_boards.py b/tests/test_boards.py
@@ -261,9 +261,6 @@ def test_led_board_nested():
         assert pin3.state
 
 def test_led_board_bad_blink():
-    pin1 = Device.pin_factory.pin(2)
-    pin2 = Device.pin_factory.pin(3)
-    pin3 = Device.pin_factory.pin(4)
     with LEDBoard(2, LEDBoard(3, 4)) as board:
         with pytest.raises(ValueError):
             board.blink(fade_in_time=1, fade_out_time=1)
@@ -846,13 +843,179 @@ def test_rgbledboard_initial_value_no_pwm():
             board.value = ((0.5, 0.5, 0.5), (0.5, 0.5, 0.5))
         assert board.value == ((0, 0, 0), (0, 0, 0))
 
-def test_rgbledboard_blink():
-    with RGBLEDBoard((2, 3, 4), (5, 6, 7)) as board:
-        board.blink()
-        assert board.value == ((1, 1, 1), (1, 1, 1))
-        board.off()
-        board[0].blink()
-        assert board.value == ((1, 1, 1), (0, 0, 0))
+# start
+def test_rgbledboard_bad_blink():
+    with RGBLEDBoard((2, 3, 4), (5, 6, 7), pwm=False) as board:
+        with pytest.raises(ValueError):
+            board.blink(fade_in_time=1, fade_out_time=1)
+        with pytest.raises(ValueError):
+            board.blink(fade_out_time=1)
+        with pytest.raises(ValueError):
+            board.pulse()
+
+@pytest.mark.skipif(hasattr(sys, 'pypy_version_info'),
+                    reason='timing is too random on pypy')
+def test_rgbledboard_blink_background():
+    pins = [Device.pin_factory.pin(n) for n in (2, 3, 4, 5, 6, 7)]
+    with RGBLEDBoard((2, 3, 4), (5, 6, 7), pwm=False) as board:
+        # Instantiation takes a long enough time that it throws off our timing
+        # here!
+        for pin in pins:
+            pin.clear_states()
+        board.blink(0.1, 0.1, n=2)
+        board._blink_thread.join() # naughty, but ensures no arbitrary waits in the test
+        test = [
+            (0.0, False),
+            (0.0, True),
+            (0.1, False),
+            (0.1, True),
+            (0.1, False)
+            ]
+        for pin in pins:
+            pin.assert_states_and_times(test)
+
+@pytest.mark.skipif(hasattr(sys, 'pypy_version_info'),
+                    reason='timing is too random on pypy')
+def test_rgbledboard_blink_foreground():
+    pins = [Device.pin_factory.pin(n) for n in (2, 3, 4, 5, 6, 7)]
+    with RGBLEDBoard((2, 3, 4), (5, 6, 7), pwm=False) as board:
+        for pin in pins:
+            pin.clear_states()
+        board.blink(0.1, 0.1, n=2, background=False)
+        test = [
+            (0.0, False),
+            (0.0, True),
+            (0.1, False),
+            (0.1, True),
+            (0.1, False)
+            ]
+        for pin in pins:
+            pin.assert_states_and_times(test)
+
+@pytest.mark.skipif(hasattr(sys, 'pypy_version_info'),
+                    reason='timing is too random on pypy')
+def test_rgbledboard_blink_control():
+    pins = [Device.pin_factory.pin(n) for n in (2, 3, 4, 5, 6, 7)]
+    with RGBLEDBoard((2, 3, 4), (5, 6, 7), pwm=False) as board:
+        for pin in pins:
+            pin.clear_states()
+        board.blink(0.1, 0.1, n=2)
+        # make sure the blink thread's started
+        while not board._blink_leds:
+            sleep(0.00001) # pragma: no cover
+        board[1][0].off() # immediately take over the second LED
+        board._blink_thread.join() # naughty, but ensures no arbitrary waits in the test
+        test = [
+            (0.0, False),
+            (0.0, True),
+            (0.1, False),
+            (0.1, True),
+            (0.1, False)
+            ]
+        test2 = [(0.0, False), (0.0, True), (0.0, False)]
+        for pin in pins[:3]:
+            pin.assert_states_and_times(test)
+        for pin in pins[3:]:
+            pin.assert_states_and_times(test2)
+
+@pytest.mark.skipif(hasattr(sys, 'pypy_version_info'),
+                    reason='timing is too random on pypy')
+def test_rgbledboard_blink_take_over():
+    pins = [Device.pin_factory.pin(n) for n in (2, 3, 4, 5, 6, 7)]
+    with RGBLEDBoard((2, 3, 4), (5, 6, 7), pwm=False) as board:
+        for pin in pins:
+            pin.clear_states()
+        board[1].blink(0.1, 0.1, n=2)
+        board.blink(0.1, 0.1, n=2) # immediately take over blinking
+        board[1]._blink_thread.join()
+        board._blink_thread.join()
+        test = [
+            (0.0, False),
+            (0.0, True),
+            (0.1, False),
+            (0.1, True),
+            (0.1, False)
+            ]
+        for pin in pins:
+            pin.assert_states_and_times(test)
+
+@pytest.mark.skipif(hasattr(sys, 'pypy_version_info'),
+                    reason='timing is too random on pypy')
+def test_rgbledboard_blink_control_all():
+    pins = [Device.pin_factory.pin(n) for n in (2, 3, 4, 5, 6, 7)]
+    with RGBLEDBoard((2, 3, 4), (5, 6, 7), pwm=False) as board:
+        for pin in pins:
+            pin.clear_states()
+        board.blink(0.1, 0.1, n=2)
+        # make sure the blink thread's started
+        while not board._blink_leds:
+            sleep(0.00001) # pragma: no cover
+        board[0].off() # immediately take over all LEDs
+        board[1].off()
+        board._blink_thread.join() # blink should terminate here anyway
+        test = [
+            (0.0, False),
+            (0.0, True),
+            (0.0, False),
+            ]
+        for pin in pins:
+            pin.assert_states_and_times(test)
+
+def test_rgbledboard_blink_interrupt_on():
+    pins = [Device.pin_factory.pin(n) for n in (2, 3, 4, 5, 6, 7)]
+    with RGBLEDBoard((2, 3, 4), (5, 6, 7), pwm=False) as board:
+        board.blink(1, 0.1)
+        sleep(0.2)
+        board.off() # should interrupt while on
+        for pin in pins:
+            pin.assert_states([False, True, False])
+
+def test_rgbledboard_blink_interrupt_off():
+    pins = [Device.pin_factory.pin(n) for n in (2, 3, 4, 5, 6, 7)]
+    with RGBLEDBoard((2, 3, 4), (5, 6, 7), pwm=False) as board:
+        for pin in pins:
+            pin.clear_states()
+        board.blink(0.1, 1)
+        sleep(0.2)
+        board.off() # should interrupt while off
+        for pin in pins:
+            pin.assert_states([False, True, False])
+
+@pytest.mark.skipif(hasattr(sys, 'pypy_version_info'),
+                    reason='timing is too random on pypy')
+def test_rgbledboard_fade_background():
+    pins = [Device.pin_factory.pin(n) for n in (2, 3, 4, 5, 6, 7)]
+    with RGBLEDBoard((2, 3, 4), (5, 6, 7), pwm=False) as board:
+        for pin in pins:
+            pin.clear_states()
+        board.blink(0, 0, 0.2, 0.2, n=2)
+        board._blink_thread.join()
+        test = [
+            (0.0, 0),
+            (0.04, 0.2),
+            (0.04, 0.4),
+            (0.04, 0.6),
+            (0.04, 0.8),
+            (0.04, 1),
+            (0.04, 0.8),
+            (0.04, 0.6),
+            (0.04, 0.4),
+            (0.04, 0.2),
+            (0.04, 0),
+            (0.04, 0.2),
+            (0.04, 0.4),
+            (0.04, 0.6),
+            (0.04, 0.8),
+            (0.04, 1),
+            (0.04, 0.8),
+            (0.04, 0.6),
+            (0.04, 0.4),
+            (0.04, 0.2),
+            (0.04, 0),
+            ]
+        for pin in pins:
+            pin.assert_states_and_times(test)
+# end
 
 def test_traffic_lights_buzzer():
     red_pin = Device.pin_factory.pin(2)
