From 3b3e4d9f34c22c67aa56b543f851e4220f835a9f Mon Sep 17 00:00:00 2001 From: Correl Date: Sat, 27 May 2023 17:36:27 -0400 Subject: [PATCH] Add rotary knob volume control --- main.py | 45 +++++++++--- rotary.py | 173 ++++++++++++++++++++++++++++++++++++++++++++++ rotary_irq_esp.py | 76 ++++++++++++++++++++ 3 files changed, 285 insertions(+), 9 deletions(-) create mode 100644 rotary.py create mode 100644 rotary_irq_esp.py diff --git a/main.py b/main.py index ed2d7ac..5cb6443 100644 --- a/main.py +++ b/main.py @@ -10,6 +10,7 @@ from umqtt.simple import MQTTClient import ssd1306 import mcp4 +from rotary_irq_esp import RotaryIRQ from statetree import StateTree VOLUME_MAX = const(128) @@ -29,12 +30,24 @@ state = StateTree( ) last_update = 0 -i2c = SoftI2C(sda=Pin(2), scl=Pin(16)) -oled_width = const(128) -oled_height = const(32) -oled = ssd1306.SSD1306_I2C(oled_width, oled_height, i2c) -buf = bytearray((oled_height // 8) * oled_width) -fbuf = framebuf.FrameBuffer1(buf, oled_width, oled_height) +rotary = RotaryIRQ( + 33, + 32, + 0, + max_val=128, + range_mode=RotaryIRQ.RANGE_BOUNDED, + pull_up=True, +) +rotary_value = rotary.value() + +try: + i2c = SoftI2C(sda=Pin(21), scl=Pin(22)) + oled_width = const(128) + oled_height = const(32) + oled = ssd1306.SSD1306_I2C(oled_width, oled_height, i2c) +except Exception as e: + print("WARNING: OLED unavailable:", e) + oled = None sta_if = network.WLAN(network.STA_IF) @@ -65,11 +78,25 @@ def on_message(topic, msg): def loop(): - global mqtt, state, last_update + global mqtt, state, last_update, rotary, rotary_value state["volume"]["left"] = pot.read(0) state["volume"]["right"] = pot.read(1) + if state.changed: + # Volume changed externally + rotary.set(value=max(state["volume"]["left"], state["volume"]["right"])) + rotary_value = rotary.value() + + new_value = rotary.value() + if rotary_value != new_value: + print("Rotary:", new_value) + state["volume"]["left"] = new_value + state["volume"]["right"] = new_value + pot.write(0, new_value) + pot.write(1, new_value) + rotary_value = new_value + if not sta_if.active(): print("Connecting to WiFi") sta_if.active(True) @@ -170,7 +197,7 @@ def loop(): mqtt.publish(topic, payload, retain=True) last_update = utime.time() mqtt.check_msg() - if state.changed: + if oled and state.changed: oled.fill(0) oled.framebuf.rect(10, 0, 92, 8, 1) oled.framebuf.rect( @@ -191,4 +218,4 @@ def loop(): while True: loop() - utime.sleep(1) + utime.sleep(0.1) diff --git a/rotary.py b/rotary.py new file mode 100644 index 0000000..b599e74 --- /dev/null +++ b/rotary.py @@ -0,0 +1,173 @@ +# MIT License (MIT) +# Copyright (c) 2022 Mike Teachman +# https://opensource.org/licenses/MIT + +# Platform-independent MicroPython code for the rotary encoder module + +# Documentation: +# https://github.com/MikeTeachman/micropython-rotary + +import micropython + +_DIR_CW = const(0x10) # Clockwise step +_DIR_CCW = const(0x20) # Counter-clockwise step + +# Rotary Encoder States +_R_START = const(0x0) +_R_CW_1 = const(0x1) +_R_CW_2 = const(0x2) +_R_CW_3 = const(0x3) +_R_CCW_1 = const(0x4) +_R_CCW_2 = const(0x5) +_R_CCW_3 = const(0x6) +_R_ILLEGAL = const(0x7) + +_transition_table = [ + + # |------------- NEXT STATE -------------| |CURRENT STATE| + # CLK/DT CLK/DT CLK/DT CLK/DT + # 00 01 10 11 + [_R_START, _R_CCW_1, _R_CW_1, _R_START], # _R_START + [_R_CW_2, _R_START, _R_CW_1, _R_START], # _R_CW_1 + [_R_CW_2, _R_CW_3, _R_CW_1, _R_START], # _R_CW_2 + [_R_CW_2, _R_CW_3, _R_START, _R_START | _DIR_CW], # _R_CW_3 + [_R_CCW_2, _R_CCW_1, _R_START, _R_START], # _R_CCW_1 + [_R_CCW_2, _R_CCW_1, _R_CCW_3, _R_START], # _R_CCW_2 + [_R_CCW_2, _R_START, _R_CCW_3, _R_START | _DIR_CCW], # _R_CCW_3 + [_R_START, _R_START, _R_START, _R_START]] # _R_ILLEGAL + +_transition_table_half_step = [ + [_R_CW_3, _R_CW_2, _R_CW_1, _R_START], + [_R_CW_3 | _DIR_CCW, _R_START, _R_CW_1, _R_START], + [_R_CW_3 | _DIR_CW, _R_CW_2, _R_START, _R_START], + [_R_CW_3, _R_CCW_2, _R_CCW_1, _R_START], + [_R_CW_3, _R_CW_2, _R_CCW_1, _R_START | _DIR_CW], + [_R_CW_3, _R_CCW_2, _R_CW_3, _R_START | _DIR_CCW], + [_R_START, _R_START, _R_START, _R_START], + [_R_START, _R_START, _R_START, _R_START]] + +_STATE_MASK = const(0x07) +_DIR_MASK = const(0x30) + + +def _wrap(value, incr, lower_bound, upper_bound): + range = upper_bound - lower_bound + 1 + value = value + incr + + if value < lower_bound: + value += range * ((lower_bound - value) // range + 1) + + return lower_bound + (value - lower_bound) % range + + +def _bound(value, incr, lower_bound, upper_bound): + return min(upper_bound, max(lower_bound, value + incr)) + + +def _trigger(rotary_instance): + for listener in rotary_instance._listener: + listener() + + +class Rotary(object): + + RANGE_UNBOUNDED = const(1) + RANGE_WRAP = const(2) + RANGE_BOUNDED = const(3) + + def __init__(self, min_val, max_val, incr, reverse, range_mode, half_step, invert): + self._min_val = min_val + self._max_val = max_val + self._incr = incr + self._reverse = -1 if reverse else 1 + self._range_mode = range_mode + self._value = min_val + self._state = _R_START + self._half_step = half_step + self._invert = invert + self._listener = [] + + def set(self, value=None, min_val=None, incr=None, + max_val=None, reverse=None, range_mode=None): + # disable DT and CLK pin interrupts + self._hal_disable_irq() + + if value is not None: + self._value = value + if min_val is not None: + self._min_val = min_val + if max_val is not None: + self._max_val = max_val + if incr is not None: + self._incr = incr + if reverse is not None: + self._reverse = -1 if reverse else 1 + if range_mode is not None: + self._range_mode = range_mode + self._state = _R_START + + # enable DT and CLK pin interrupts + self._hal_enable_irq() + + def value(self): + return self._value + + def reset(self): + self._value = 0 + + def close(self): + self._hal_close() + + def add_listener(self, l): + self._listener.append(l) + + def remove_listener(self, l): + if l not in self._listener: + raise ValueError('{} is not an installed listener'.format(l)) + self._listener.remove(l) + + def _process_rotary_pins(self, pin): + old_value = self._value + clk_dt_pins = (self._hal_get_clk_value() << + 1) | self._hal_get_dt_value() + + if self._invert: + clk_dt_pins = ~clk_dt_pins & 0x03 + + # Determine next state + if self._half_step: + self._state = _transition_table_half_step[self._state & + _STATE_MASK][clk_dt_pins] + else: + self._state = _transition_table[self._state & + _STATE_MASK][clk_dt_pins] + direction = self._state & _DIR_MASK + + incr = 0 + if direction == _DIR_CW: + incr = self._incr + elif direction == _DIR_CCW: + incr = -self._incr + + incr *= self._reverse + + if self._range_mode == self.RANGE_WRAP: + self._value = _wrap( + self._value, + incr, + self._min_val, + self._max_val) + elif self._range_mode == self.RANGE_BOUNDED: + self._value = _bound( + self._value, + incr, + self._min_val, + self._max_val) + else: + self._value = self._value + incr + + try: + if old_value != self._value and len(self._listener) != 0: + _trigger(self) + except: + pass diff --git a/rotary_irq_esp.py b/rotary_irq_esp.py new file mode 100644 index 0000000..3cf0615 --- /dev/null +++ b/rotary_irq_esp.py @@ -0,0 +1,76 @@ +# MIT License (MIT) +# Copyright (c) 2020 Mike Teachman +# https://opensource.org/licenses/MIT + +# Platform-specific MicroPython code for the rotary encoder module +# ESP8266/ESP32 implementation + +# Documentation: +# https://github.com/MikeTeachman/micropython-rotary + +from machine import Pin +from rotary import Rotary +from sys import platform + +_esp8266_deny_pins = [16] + + +class RotaryIRQ(Rotary): + + def __init__(self, pin_num_clk, pin_num_dt, min_val=0, max_val=10, incr=1, + reverse=False, range_mode=Rotary.RANGE_UNBOUNDED, pull_up=False, half_step=False, invert=False): + + if platform == 'esp8266': + if pin_num_clk in _esp8266_deny_pins: + raise ValueError( + '%s: Pin %d not allowed. Not Available for Interrupt: %s' % + (platform, pin_num_clk, _esp8266_deny_pins)) + if pin_num_dt in _esp8266_deny_pins: + raise ValueError( + '%s: Pin %d not allowed. Not Available for Interrupt: %s' % + (platform, pin_num_dt, _esp8266_deny_pins)) + + super().__init__(min_val, max_val, incr, reverse, range_mode, half_step, invert) + + if pull_up == True: + self._pin_clk = Pin(pin_num_clk, Pin.IN, Pin.PULL_UP) + self._pin_dt = Pin(pin_num_dt, Pin.IN, Pin.PULL_UP) + else: + self._pin_clk = Pin(pin_num_clk, Pin.IN) + self._pin_dt = Pin(pin_num_dt, Pin.IN) + + self._enable_clk_irq(self._process_rotary_pins) + self._enable_dt_irq(self._process_rotary_pins) + + def _enable_clk_irq(self, callback=None): + self._pin_clk.irq( + trigger=Pin.IRQ_RISING | Pin.IRQ_FALLING, + handler=callback) + + def _enable_dt_irq(self, callback=None): + self._pin_dt.irq( + trigger=Pin.IRQ_RISING | Pin.IRQ_FALLING, + handler=callback) + + def _disable_clk_irq(self): + self._pin_clk.irq(handler=None) + + def _disable_dt_irq(self): + self._pin_dt.irq(handler=None) + + def _hal_get_clk_value(self): + return self._pin_clk.value() + + def _hal_get_dt_value(self): + return self._pin_dt.value() + + def _hal_enable_irq(self): + self._enable_clk_irq(self._process_rotary_pins) + self._enable_dt_irq(self._process_rotary_pins) + + def _hal_disable_irq(self): + self._disable_clk_irq() + self._disable_dt_irq() + + def _hal_close(self): + self._hal_disable_irq()