Add rotary knob volume control

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))
+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)
-buf = bytearray((oled_height // 8) * oled_width)
+except Exception as e:
-fbuf = framebuf.FrameBuffer1(buf, oled_width, oled_height)
+    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)

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

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()