Add Traffic script

software/contrib/README.md

@@ -172,6 +172,13 @@ Users have the x, y, and z values of the output of each attractor model availabl
 <i>Author: [seanbechhofer](https://github.com/seanbechhofer)</i>
 <br><i>Labels: gates, triggers, randomness</i>
 
+### Traffic \[ [documentation](/software/contrib/traffic.md) | [script](/software/contrib/traffic.py) \]
+A re-imagining of [Jasmine and Olive Tree's Traffic](https://jasmineandolivetrees.com/products/traffic) module. Triggers are sent to both inputs
+generating CV signals based on which trigger fired most recently and a pair of gains per channel.
+
+<i>Author: [chrisib](http://github.com/chrisib)</i>
+<br><i>Labels: sequencer, gate, triggers</i>
+
 ### Turing Machine \[ [documentation](/software/contrib/turing_machine.md) | [script](/software/contrib/turing_machine.py) \]
 A script meant to recreate the [Music Thing Modular Turning Machine Random Sequencer](https://musicthing.co.uk/pages/turing.html)
 as faithfully as possible on the EuroPi hardware.

software/contrib/menu.py

@@ -53,6 +53,7 @@
     ["Seq. Switch",       "contrib.sequential_switch.SequentialSwitch"],
     ["Smooth Rnd Volts",  "contrib.smooth_random_voltages.SmoothRandomVoltages"],
     ["StrangeAttractor",  "contrib.strange_attractor.StrangeAttractor"],
+    ["Traffic",           "contrib.traffic.Traffic"],
     ["Turing Machine",    "contrib.turing_machine.EuroPiTuringMachine"],
 
     ["_Calibrate",        "calibrate.Calibrate"],              # this one should always be second to last!

software/contrib/traffic.md

@@ -0,0 +1,58 @@
+# Traffic
+
+This script is inspired by [Jasmine & Olive Tree's Traffic](https://jasmineandolivetrees.com/products/traffic).
+Instead of 3 trigger inputs, this version only has 2.
+
+Traffic has 3 output channels (`cv1`-`cv3`), which output CVs signals. The value of the output signal depends on:
+1. which input the trigger was received most recently and
+2. the gains for that trigger on each channel.
+
+For example, suppose the gains for channel A are set to `[0.25, 0.6]`. Whenever a trigger on `din` (trigger input 1) is
+received, channel A (`cv1`) will output `MAX_VOLTAGE * 0.25 = 2.5V`. Whenever a trigger on `ain` (trigger input 2) in
+received, channel A will output `MAX_VOLTAGE * 0.6 = 6V`.
+
+The same occurs for channels B and C on `cv2` and `cv3`, each with their own pair of gains for the two inputs.
+
+Changing the gains will immediately update the output value, if the gain for that input is active. i.e. if `din` last
+detected a trigger, changing the gains for `din` on channel A/B/C will affect the voltage on `cv1/2/3` immediately.
+
+`cv6` outputs a 10ms, 5V gate every time a trigger is received on _either_ input.
+
+`cv4` and `cv5` have no equivalent on the original Traffic module, but are used here as difference channels:
+- `cv4` is the absolute difference between `cv1` and `cv2`
+- `cv4` is the absolute difference between `cv1` and `cv3`
+
+For a video tutorial on how the original Traffic module works, please see
+https://youtu.be/Pn7_NCCKcJc?si=OJ78FRa9PvjD8oSd. The functionality of this script is very much the same, but limited
+to two input triggers.
+
+
+## Setting the gains
+
+Turning `k1` and `k2` will set the gains for channel A. Pressing and holding `b1` while rotating the knobs will set the
+gains for channel B. Pressing and holding `b2` while rotating the knobs will set the gains for channel C.
+
+The gains for channels B and C are saved to the module's onboard memory, and will persist across power-cycles.  The
+gains for channel A are always read from the current knob positions on startup.
+
+Note that this each channel makes used of "locked knobs."  This means that when changing the active channel by pressing
+or releasing `b1` or `b2` it may be necessary to sweep the knob to its prior position before the gain can be changed.
+This helps prevent accidentally changing the gains by pressing the buttons.
+
+
+## I/O Mapping
+
+| I/O           | Usage
+|---------------|-------------------------------------------------------------------|
+| `din`         | Trigger input 1                                                   |
+| `ain`         | Trigger input 2                                                   |
+| `b1`          | Hold to adjust gains for channel B                                |
+| `b2`          | Hold to adjust gains for channel C                                |
+| `k1`          | Input 1 gain for channel A/B/C                                    |
+| `k2`          | Input 2 gain for channel A/B/C                                    |
+| `cv1`         | Channel A output                                                  |
+| `cv2`         | Channel B output                                                  |
+| `cv3`         | Channel C output                                                  |
+| `cv4`         | Channel A minus channel B (absolute value)                        |
+| `cv5`         | Channel A minus channel C (absolute value)                        |
+| `cv6`         | 10ms, 5V trigger whenever a rising edge occurs on `ain` or `din`  |

software/contrib/traffic.py

@@ -0,0 +1,152 @@
+#!/usr/bin/env python3
+"""A EuroPi re-imagining of Traffic by Jasmine & Olive Trees
+
+Two gate inputs are sent to AIN and DIN, their values multiplied by gains controlled by K1 and K2,
+and the summed & differenced outputs sent to the outputs
+"""
+
+from europi import *
+from europi_script import EuroPiScript
+
+from experimental.a_to_d import AnalogReaderDigitalWrapper
+from experimental.knobs import KnobBank
+from experimental.screensaver import OledWithScreensaver
+
+ssoled = OledWithScreensaver()
+
+class Traffic(EuroPiScript):
+    def __init__(self):
+        super().__init__()
+
+        state = self.load_state_json()
+
+        # Button handlers to change the active channel for setting gains
+        self.channel_markers = ['>', ' ', ' ']    # used to indicate the editable channel on the screen
+        @b1.handler
+        def b1_rising():
+            """Activate channel b controls while b1 is held
+            """
+            ssoled.notify_user_interaction()
+            self.k1_bank.set_current("channel_b")
+            self.k2_bank.set_current("channel_b")
+            self.channel_markers[0] = ' '
+            self.channel_markers[1] = '>'
+            self.channel_markers[2] = ' '
+
+        @b2.handler
+        def b2_rising():
+            """Activate channel c controls while b1 is held
+            """
+            ssoled.notify_user_interaction()
+            self.k1_bank.set_current("channel_c")
+            self.k2_bank.set_current("channel_c")
+            self.channel_markers[0] = ' '
+            self.channel_markers[1] = ' '
+            self.channel_markers[2] = '>'
+
+        def either_button_falling():
+            """Revert to channel a when the button is released
+            """
+            self.k1_bank.set_current("channel_a")
+            self.k2_bank.set_current("channel_a")
+            self.channel_markers[0] = '>'
+            self.channel_markers[1] = ' '
+            self.channel_markers[2] = ' '
+            self.save_state()
+        b1.handler_falling(either_button_falling)
+        b2.handler_falling(either_button_falling)
+
+
+        # Input trigger handlers
+        self.input1_trigger_at = time.ticks_ms()
+        self.input2_trigger_at = self.input1_trigger_at
+        @din.handler
+        def din1_rising():
+            self.input1_trigger_at = time.ticks_ms()
+
+            # Set the all-triggers output high
+            self.last_output_trigger_at = self.input1_trigger_at
+            cv6.on()
+
+        def din2_rising():
+            self.input2_trigger_at = time.ticks_ms()
+
+            # Set the all-triggers output high
+            self.last_output_trigger_at = self.input2_trigger_at
+            cv6.on()
+
+        self.k1_bank = (
+            KnobBank.builder(k1) \
+            .with_unlocked_knob("channel_a") \
+            .with_locked_knob("channel_b", initial_percentage_value=state.get("gain_b1", 0.5)) \
+            .with_locked_knob("channel_c", initial_percentage_value=state.get("gain_c1", 0.5)) \
+            .build()
+        )
+
+        self.k2_bank = (
+            KnobBank.builder(k2) \
+            .with_unlocked_knob("channel_a") \
+            .with_locked_knob("channel_b", initial_percentage_value=state.get("gain_b2", 0.5)) \
+            .with_locked_knob("channel_c", initial_percentage_value=state.get("gain_c2", 0.5)) \
+            .build()
+        )
+
+        self.din1 = din
+        self.din2 = AnalogReaderDigitalWrapper(ain, cb_rising=din2_rising)
+
+        # we fire a trigger on CV6, so keep track of the previous outputs so we know when something's changed
+        self.last_output_trigger_at = time.ticks_ms()
+
+    def save_state(self):
+        state = {
+            "gain_b1": self.k1_bank["channel_b"].percent(samples=1024),
+            "gain_b2": self.k2_bank["channel_b"].percent(samples=1024),
+            "gain_c1": self.k1_bank["channel_c"].percent(samples=1024),
+            "gain_c2": self.k2_bank["channel_c"].percent(samples=1024)
+        }
+        self.save_state_json(state)
+
+    def main(self):
+        TRIGGER_DURATION = 10   # 10ms triggers every time we get a rising edge on either input channel
+
+        while True:
+            self.din2.update()
+
+            gain_a1 = round(self.k1_bank["channel_a"].percent(samples=1024), 3)
+            gain_a2 = round(self.k2_bank["channel_a"].percent(samples=1024), 3)
+
+            gain_b1 = round(self.k1_bank["channel_b"].percent(samples=1024), 3)
+            gain_b2 = round(self.k2_bank["channel_b"].percent(samples=1024), 3)
+
+            gain_c1 = round(self.k1_bank["channel_c"].percent(samples=1024), 3)
+            gain_c2 = round(self.k2_bank["channel_c"].percent(samples=1024), 3)
+
+            # calculate the outputs
+            delta_t = time.ticks_diff(self.input1_trigger_at, self.input2_trigger_at)
+            if delta_t > 0 or abs(delta_t) <= 50:       # assume triggers within 50ms are simultaneous
+                # din received a trigger more recently than ain
+                out_a = MAX_OUTPUT_VOLTAGE * gain_a1
+                out_b = MAX_OUTPUT_VOLTAGE * gain_b1
+                out_c = MAX_OUTPUT_VOLTAGE * gain_c1
+            else:
+                # ain received a trigger more recently than din
+                out_a = MAX_OUTPUT_VOLTAGE * gain_a2
+                out_b = MAX_OUTPUT_VOLTAGE * gain_b2
+                out_c = MAX_OUTPUT_VOLTAGE * gain_c2
+
+            cv1.voltage(out_a)
+            cv2.voltage(out_b)
+            cv3.voltage(out_c)
+            cv4.voltage(abs(out_a - out_b))
+            cv5.voltage(abs(out_a - out_c))
+
+            now = time.ticks_ms()
+            if time.ticks_diff(now, self.last_output_trigger_at) > TRIGGER_DURATION:
+                cv6.off()
+
+            # show the current gains * outputs, marking the channel we're controlling via the knobs & buttons
+            ssoled.centre_text(f"{self.channel_markers[0]} A {gain_a1:0.3f} {gain_a2:0.3f}\n{self.channel_markers[1]} B {gain_b1:0.3f} {gain_b2:0.3f}\n{self.channel_markers[2]} C {gain_c1:0.3f} {gain_c2:0.3f}")
+
+
+if __name__ == "__main__":
+    Traffic().main()

software/firmware/experimental/a_to_d.py

@@ -0,0 +1,70 @@
+from europi import HIGH, LOW
+
+import utime
+
+
+class AnalogReaderDigitalWrapper:
+    """Wraps an AnalogReader to allow it to simulate a DigitalReader.
+
+    The EuroPiScript class using the AnalogReaderDigitalWrapper must call `.update()` to read the current state of
+    the analogue input and trigger any rising/falling edge callbacks.
+
+    The value returned by `.value()` is accurate to the last time `.update()` was called.
+    """
+
+    def __init__(
+        self, ain, debounce=1, high_low_cutoff=0.8, cb_rising=lambda: None, cb_falling=lambda: None
+    ):
+        """Constructor
+
+        @param ain  The AnalogReader we're wrapping
+        @param debounce  The number of consecutive high/low signals needed to flip the digital state
+        @param high_low_cutoff  The threshold at which the analog signal is considered high
+
+        @param cb_rising  A function to call on the rising edge of the signal
+        @param cb_falling  A function to call on the falling edge of the signal
+        """
+        self.ain = ain
+        self.debounce = debounce
+        self.high_low_cutoff = high_low_cutoff
+        self.last_rising_time = 0
+        self.last_falling_time = 0
+        self.debounce_counter = 0
+        self.state = False
+
+        if not callable(cb_rising) or not callable(cb_falling):
+            raise ValueError("Provided callback func is not callable")
+
+        self.cb_rising = cb_rising
+        self.cb_falling = cb_falling
+
+    def value(self):
+        """Returns europi.HIGH or europi.LOW depending on the state of the input"""
+        return HIGH if self.state else LOW
+
+    def update(self):
+        """Reads the current value of the analogue input and updates the internal state"""
+        volts = self.ain.read_voltage()
+
+        # count how many opposite-voltage readings we have
+        if (self.state and volts < self.high_low_cutoff) or (
+            not self.state and volts >= self.high_low_cutoff
+        ):
+            self.debounce_counter += 1
+
+        # change state if we've reached the debounce threshold
+        if self.debounce_counter >= self.debounce:
+            self.debounce_counter = 0
+            self.state = not self.state
+            if self.state:
+                self.last_rising_time = utime.ticks_ms()
+                self.cb_rising()
+            else:
+                self.last_falling_time = utime.ticks_ms()
+                self.cb_falling()
+
+    def last_rising_ms(self):
+        return self.last_rising_time
+
+    def last_falling_ms(self):
+        return self.last_falling_time

software/firmware/experimental/screensaver.py

@@ -129,9 +129,6 @@ def show(self):
     def fill(self, color):
         oled.fill(color)
 
-    def clear(self):
-        oled.clear()
-
     def text(self, string, x, y, color=1):
         oled.text(string, x, y, color)