import time import os import wifi import socketpool import microosc import asyncio import rotaryio import digitalio import board import pwmio # ── OSC / WiFi Config ──────────────────────────────────────────────────────── UDP_HOST = "192.168.0.5" # ← Your Zynthian IP UDP_PORT = 1370 ssid = os.getenv("WIFI_SSID") password = os.getenv("WIFI_PASSWORD") print("Connecting to WiFi:", ssid) wifi.radio.connect(ssid, password) print("My IP address:", wifi.radio.ipv4_address) socket_pool = socketpool.SocketPool(wifi.radio) osc_client = microosc.OSCClient(socket_pool, UDP_HOST, UDP_PORT) print("OSC client ready →", UDP_HOST, UDP_PORT) # ── RGB LED Setup (GP9=R, GP10=G, GP11=B) ──────────────────────────────── RED = pwmio.PWMOut(board.GP9, frequency=1000, duty_cycle=0) GREEN = pwmio.PWMOut(board.GP10, frequency=1000, duty_cycle=0) BLUE = pwmio.PWMOut(board.GP11, frequency=1000, duty_cycle=0) def set_rgb(r, g, b): """Set 0–65535 duty cycle (0=off, 65535=full bright)""" RED.duty_cycle = min(65535, max(0, int(round(r)))) GREEN.duty_cycle = min(65535, max(0, int(round(g)))) BLUE.duty_cycle = min(65535, max(0, int(round(b)))) # Encoder-specific flash colors (for indices 0–3) ENCODER_FLASH_COLORS = [ (65535, 0, 0), # 0: Red ( 0, 65535, 0), # 1: Green (65535, 10000, 0), # 2: Yellow (65535, 0, 65535), # 3: Purple/Magenta ] FLASH_DURATION = 0.40 # seconds – how long the LED stays lit per message # ── Hardware Config ────────────────────────────────────────────────────────── encoders_cfg = [ [0, "Chain", board.GP3, board.GP4, board.GP5 ], [1, "Back", board.GP0, board.GP1, board.GP2 ], [2, "Snapshot", board.GP17, board.GP18, board.GP19], [3, "Select", board.GP14, board.GP15, board.GP16], ] switches_cfg = [ [4, "S1", board.GP6 ], [5, "S2", board.GP7 ], [6, "S3", board.GP21], [7, "S4", board.GP20], ] SHORT_THRESHOLD = 0.30 BOLD_THRESHOLD = 1.00 LONG_THRESHOLD = 2.00 DEBOUNCE_TIME = 0.025 # ── LED Flash Helpers (now also for ZYNPOT) ────────────────────────────────── async def flash_led_for_encoder(enc_idx: int, duration=FLASH_DURATION): """Briefly light LED in encoder's color""" if not (0 <= enc_idx <= 3): return r, g, b = ENCODER_FLASH_COLORS[enc_idx] set_rgb(r, g, b) await asyncio.sleep(duration) set_rgb(0, 0, 0) # ── Button Base Class ──────────────────────────────────────────────────────── class ButtonBase: def __init__(self, idx, name, sw_pin): self.idx = idx self.name = name self.switch = digitalio.DigitalInOut(sw_pin) self.switch.switch_to_input(pull=digitalio.Pull.UP) self.press_start = None self.long_sent = False self.last_value = self.switch.value # initial read def _send_action(self, action_type: str): try: msg = microosc.OscMsg("/CUIA/ZYNSWITCH", [self.idx, action_type], ("i", "s")) osc_client.send(msg) print(f"ZYNSWITCH {self.name} ({self.idx}): {action_type}") if 0 <= self.idx <= 3: asyncio.create_task(flash_led_for_encoder(self.idx, FLASH_DURATION)) except OSError as e: print(f"OSC send failed: {e}") def check_button(self, now: float): """One-shot button check – call every loop iteration""" pressed = not self.switch.value if pressed != self.last_value: self.last_value = pressed # Simple debounce delay would block; we rely on fast polling + threshold logic if pressed: if self.press_start is None: self.press_start = now self.long_sent = False self._send_action("P") elif not self.long_sent and (now - self.press_start) >= LONG_THRESHOLD: self._send_action("L") self.long_sent = True else: if self.press_start is not None: duration = now - self.press_start if duration < SHORT_THRESHOLD: self._send_action("S") elif duration < LONG_THRESHOLD: self._send_action("B") else: self._send_action("L") # optional repeat self.press_start = None self.long_sent = False # ── Encoder ────────────────────────────────────────────────────────────────── class Encoder(ButtonBase): def __init__(self, idx, name, sw_pin, a_pin, b_pin): super().__init__(idx, name, sw_pin) self.encoder = rotaryio.IncrementalEncoder(a_pin, b_pin) self.last_pos = self.encoder.position async def monitor(self): while True: now = time.monotonic() # Rotation check – do this first for best responsiveness pos = self.encoder.position if pos != self.last_pos: delta = pos - self.last_pos # direction = 1 if delta > 0 else -1 # or keep sign for Zynthian # steps = abs(delta) # But Zynthian ZYNPOT usually expects signed steps (positive = clockwise) try: msg = microosc.OscMsg( "/CUIA/ZYNPOT", [self.idx, delta], # send signed delta directly ("i", "i") ) osc_client.send(msg) print(f"ZYNPOT {self.name} ({self.idx}): {delta:+d}") # Brief LED flash on rotation (shorter than button flash to not overwhelm) asyncio.create_task(flash_led_for_encoder(self.idx, 0.12)) # quick 120 ms flash except OSError as e: print(f"ZYNPOT send failed: {e}") self.last_pos = pos # Button check (one iteration) self.check_button(now) await asyncio.sleep(0.008) # ~125 Hz unified polling # ── SimpleSwitch stays the same, but update to use check_button too ────────── class SimpleSwitch(ButtonBase): async def monitor(self): while True: now = time.monotonic() self.check_button(now) await asyncio.sleep(0.008) # ── Main ───────────────────────────────────────────────────────────────────── async def main(): encoders = [Encoder(*cfg) for cfg in encoders_cfg] simple_switches = [SimpleSwitch(*cfg) for cfg in switches_cfg] tasks = [] for enc in encoders: tasks.append(asyncio.create_task(enc.monitor())) for sw in simple_switches: tasks.append(asyncio.create_task(sw.monitor())) await asyncio.gather(*tasks) print("Starting main loop (LED flashes ONLY on encoder ZYNSWITCH messages)...") try: asyncio.run(main()) except KeyboardInterrupt: print("\nStopped by user") except Exception as e: print("Error:", e) finally: set_rgb(0, 0, 0) print("LEDs off")