import uasyncio as asyncio
from machine import Pin
#from aiohttp import ClientSession, web
#import accelstepper
from accelstepper import AccelStepper
import display

# Set up the stepper motor
dirn = 19 #Stepper direction pin
step = 18 #Stepper step pin
en = 5 #stepper enable pin
STEPPER_PINS = [Pin(dirn, Pin.OUT), Pin(step, Pin.OUT)]
stepper = AccelStepper(interface = 1, *STEPPER_PINS)
AccelStepper.set_enable_pin(en)

# Set the maximum speed and acceleration
controller.setMaxSpeed(1000)
controller.setAcceleration(500)

motor_running = False # Flag to track whether the motor is running
RPM = 0 # Flag to track currently-set RPM

# Set up the physical buttons
slow_button = Pin(25, Pin.IN, Pin.PULL_UP)
medium_button = Pin(33, Pin.IN, Pin.PULL_UP)
fast_button = Pin(32, Pin.IN, Pin.PULL_UP)
encoder_button = Pin(13, Pin.IN, Pin.PULL_UP)
go_button = Pin(35, Pin.IN, Pin.PULL_UP)
stop_button = Pin(34, Pin.IN, Pin.PULL_UP)
#restart_button = Pin(..., Pin.IN, Pin.PULL_UP)  # Add a new button to restart the event loop

# Create an event to signal when the program should exit
exit_event = asyncio.Event()

# Define the interrupt handler functions
async def slow_handler(pin):
    global RPM, lastpressed
    RPM = slowRPM
    stepper.set_speed(100)
    speed = 100
    #display.text("Stepper Motor Speed: {} steps/sec".format(stepper.speed()), 0, 0)
    display.show()
    async with ClientSession() as session:
        async with session.post('http://localhost:8080/forward', json={'speed': 100}) as resp:
            print(await resp.text())
    # Reset the exit event to prevent the sleep task from being triggered
    exit_event.clear()

async def backward_handler(pin):
    stepper.move(-100)
    speed = -100
    display.text("Stepper Motor Speed: {} steps/sec".format(speed), 0, 0)
    display.show()
    async with ClientSession() as session:
        async with session.post('http://localhost:8080/backward', json={'speed': 100}) as resp:
            print(await resp.text())
    # Reset the exit event to prevent the sleep task from being triggered
    exit_event.clear()

async def stop_handler(pin):
    stepper.move(0)
    speed = 0
    display.text("Stepper Motor Stopped", 0, 0)
    display.show()
    async with ClientSession() as session:
        async with session.post('http://localhost:8080/stop') as resp:
            print(await resp.text())
    # Reset the exit event to prevent the sleep task from being triggered
    exit_event.clear()

async def restart_handler(pin):
    # Reset the exit event to restart the event loop
    exit_event.clear()

# Initialise the display
display.draw_screen()

# Attach the interrupt handler functions to the physical buttons
slow_button.irq(trigger=Pin.IRQ_FALLING, handler=slow_handler)
medium_button.irq(trigger=Pin.IRQ_FALLING, handler=medium_handler)
fast_button.irq(trigger=Pin.IRQ_FALLING, handler=fast_handler)
encoder_button.irq(trigger=Pin.IRQ_FALLING, handler=encoder_handler)
stop_button.irq(trigger=Pin.IRQ_FALLING, handler=stop_handler)
#restart_button.irq(trigger=Pin.IRQ_FALLING, handler=restart_handler)

# Start the event loop
# while not exit_event.is_set():
#     asyncio.sleep(1)

# Create a sleep task that waits for 10 minutes of inactivity before displaying a message and setting the exit event
async def sleep_task():
    await asyncio.sleep(600)  # Sleep for 10 minutes
    display.text("Controller Going to Sleep", 0, 16)
    display.show()
    exit_event.set()

# Start the sleep task
asyncio.create_task(sleep_task())

# Run the event loop indefinitely
while True:
    asyncio.run_until_complete(exit_event.wait())



main.py

diagram.json

from math import sqrt, fabs
from machine import Pin
from utime import ticks_us, sleep_ms


def constrain(val, min_val, max_val):
    return min(max_val, max(min_val, val))


DIRECTION_CCW = 0,  # Counter-Clockwise
DIRECTION_CW = 1  # Clockwise

FUNCTION = 0  # Use the functional interface, implementing your own driver functions (internal use only)
DRIVER = 1  # Stepper Driver, 2 driver pins required
FULL2WIRE = 2  # 2 wire stepper, 2 motor pins required
FULL3WIRE = 3  # 3 wire stepper, such as HDD spindle, 3 motor pins required
FULL4WIRE = 4  # 4 wire full stepper, 4 motor pins required
HALF3WIRE = 6  # 3 wire half stepper, such as HDD spindle, 3 motor pins required
HALF4WIRE = 8  # 4 wire half stepper, 4 motor pins required


class AccelStepper:
    def __init__(self, *args):
        self._currentPos = 0
        self._targetPos = 0
        self._speed = 0.0
        self._maxSpeed = 1.0
        self._acceleration = 0.0
        self._sqrt_twoa = 1.0
        self._stepInterval = 0
        self._minPulseWidth = 1
        self._enablePin = 0xff
        self._lastStepTime = 0
        self._pin = [0, 0, 0, 0]
        self._enableInverted = False
        self._n = 0
        self._c0 = 0.0
        self._cn = 0.0
        self._cmin = 1.0
        self._direction = DIRECTION_CCW
        self._pinInverted = [0, 0, 0, 0]

        if len(args) == 6:
            self._interface = args[0]
            self._pin[0] = args[1]
            self._pin[1] = args[2]
            self._pin[2] = args[3]
            self._pin[3] = args[4]
            if args[5]:
                self.enable_outputs()
        elif len(args) == 2:
            self._interface = 0
            self._forward = args[0]
            self._backward = args[1]

        self.set_acceleration(1)

    def move_to(self, absolute: int) -> None:
        if self._targetPos != absolute:
            self._targetPos = absolute
            self.compute_new_speed()

    def move(self, relative: int) -> None:
        self.move_to(self._currentPos + relative)

    def run_speed(self) -> bool:
        if not self._stepInterval:
            return False
        time = ticks_us()
        if (time - self._lastStepTime) >= self._stepInterval:
            if self._direction == DIRECTION_CW:
                self._currentPos += 1
            else:
                self._currentPos -= 1
            self.step(self._currentPos)
            self._lastStepTime = time
            return True
        else:
            return False

    def distance_to_go(self) -> int:
        return self._targetPos - self._currentPos

    def target_position(self) -> int:
        return self._targetPos

    def current_position(self) -> int:
        return self._currentPos

    def set_current_position(self, position: int) -> None:
        self._targetPos = self._currentPos = position
        self._n = 0
        self._stepInterval = 0
        self._speed = 0.0

    def compute_new_speed(self) -> None:
        distance_to = self.distance_to_go()
        steps_to_stop = int((self._speed * self._speed) / (2.0 * self._acceleration))
        if distance_to == 0 and steps_to_stop <= 1:
            self._stepInterval = 0
            self._speed = 0.0
            self._n = 0
            return
        if distance_to > 0:
            if self._n > 0:
                if (steps_to_stop >= distance_to) or self._direction == DIRECTION_CCW:
                    self._n = -steps_to_stop
            elif self._n < 0:
                if (steps_to_stop < distance_to) and self._direction == DIRECTION_CW:
                    self._n = -self._n
        elif distance_to < 0:
            if self._n > 0:
                if (steps_to_stop >= -distance_to) or self._direction == DIRECTION_CW:
                    self._n = -steps_to_stop
            elif self._n < 0:
                if (steps_to_stop < -distance_to) and self._direction == DIRECTION_CCW:
                    self._n = -self._n
        if self._n == 0:
            self._cn = self._c0
            self._direction = DIRECTION_CW if distance_to > 0 else DIRECTION_CCW
        else:
            self._cn = self._cn - ((2.0 * self._cn) / ((4.0 * self._n) + 1))
            self._cn = max(self._cn, self._cmin)
        self._n += 1
        self._stepInterval = self._cn
        self._speed = 1000000.0 / self._cn
        if self._direction == DIRECTION_CCW:
            self._speed = -self._speed

    def run(self) -> bool:
        if self.run_speed():
            self.compute_new_speed()
        return self._speed != 0.0 or self.distance_to_go() != 0

    def set_max_speed(self, speed: float) -> None:
        if speed < 0.0:
            speed = -speed
        if self._maxSpeed != speed:
            self._maxSpeed = speed
            self._cmin = 1000000.0 / speed
            if self._n > 0:
                self._n = int((self._speed * self._speed) / (2.0 * self._acceleration))
                self.compute_new_speed()

    def max_speed(self):
        return self._maxSpeed

    def set_acceleration(self, acceleration: float) -> None:
        if acceleration == 0.0:
            return
        if acceleration < 0.0:
            acceleration = -acceleration
        if self._acceleration != acceleration:
            self._n = self._n * (self._acceleration / acceleration)
            self._c0 = 0.676 * sqrt(2.0 / acceleration) * 1000000.0
            self._acceleration = acceleration
            self.compute_new_speed()

    def set_speed(self, speed: float) -> None:
        if speed == self._speed:
            return
        speed = constrain(speed, -self._maxSpeed, self._maxSpeed)
        if speed == 0.0:
            self._stepInterval = 0
        else:
            self._stepInterval = fabs(1000000.0 / speed)
            self._direction = DIRECTION_CW if speed > 0.0 else DIRECTION_CCW
        self._speed = speed

    def speed(self) -> float:
        return self._speed

    def step(self, step: int) -> None:
        if self._interface is FUNCTION:
            self.step0(step)
        elif self._interface is DRIVER:
            self.step1(step)
        elif self._interface is FULL2WIRE:
            self.step2(step)
        elif self._interface is FULL3WIRE:
            self.step3(step)
        elif self._interface is FULL4WIRE:
            self.step4(step)
        elif self._interface is HALF3WIRE:
            self.step6(step)
        elif self._interface is HALF4WIRE:
            self.step8(step)

    def set_output_pins(self, mask: int) -> None:
        num_pins = 2
        if self._interface is FULL4WIRE or self._interface is HALF4WIRE:
            num_pins = 4
        elif self._interface is FULL3WIRE or self._interface is HALF3WIRE:
            num_pins = 3
        for i in range(num_pins):
            self._pin[i].value(True ^ self._pinInverted[i] if mask & (1 << i) else False ^ self._pinInverted[i])

    def step0(self, step: int) -> None:
        if self._speed > 0:
            self._forward()
        else:
            self._backward()

    def step1(self, step: int) -> None:
        self.set_output_pins(0b10 if self._direction else 0b00)
        self.set_output_pins(0b11 if self._direction else 0b01)
        sleep_ms(self._minPulseWidth)
        self.set_output_pins(0b10 if self._direction else 0b00)

    def step2(self, step: int) -> None:
        aux = step & 0x3
        if aux == 0:
            self.set_output_pins(0b10)
        elif aux == 1:
            self.set_output_pins(0b11)
        elif aux == 2:
            self.set_output_pins(0b01)
        elif aux == 3:
            self.set_output_pins(0b00)

    def step3(self, step: int) -> None:
        aux = step % 3
        if aux == 0:
            self.set_output_pins(0b100)
        elif aux == 1:
            self.set_output_pins(0b001)
        elif aux == 2:
            self.set_output_pins(0b010)

    def step4(self, step: int) -> None:
        aux = step & 0x3
        if aux == 0:
            self.set_output_pins(0b0101)
        elif aux == 1:
            self.set_output_pins(0b0110)
        elif aux == 2:
            self.set_output_pins(0b1010)
        elif aux == 3:
            self.set_output_pins(0b1001)

    def step6(self, step: int) -> None:
        aux = step % 6
        if aux == 0:
            self.set_output_pins(0b100)
        elif aux == 1:
            self.set_output_pins(0b101)
        elif aux == 2:
            self.set_output_pins(0b001)
        elif aux == 3:
            self.set_output_pins(0b011)
        elif aux == 4:
            self.set_output_pins(0b010)
        elif aux == 5:
            self.set_output_pins(0b110)

    def step8(self, step: int) -> None:
        aux = step & 0x7
        if aux == 0:
            self.set_output_pins(0b0001)
        elif aux == 1:
            self.set_output_pins(0b0101)
        elif aux == 2:
            self.set_output_pins(0b0100)
        elif aux == 3:
            self.set_output_pins(0b0110)
        elif aux == 4:
            self.set_output_pins(0b0010)
        elif aux == 5:
            self.set_output_pins(0b1010)
        elif aux == 6:
            self.set_output_pins(0b1000)
        elif aux == 7:
            self.set_output_pins(0b1001)

    def disable_outputs(self) -> None:
        if not self._interface:
            return
        self.set_output_pins(0)
        if self._enablePin != 0xff:
            self._enablePin = Pin(self._enablePin, Pin.OUT)
            self._enablePin.value(False ^ self._enableInverted)

    def enable_outputs(self) -> None:
        if not self._interface:
            return
        self._pin[0] = Pin(self._pin[0], Pin.OUT)
        self._pin[1] = Pin(self._pin[1], Pin.OUT)
        if self._interface is FULL4WIRE or self._interface is HALF4WIRE:
            self._pin[2] = Pin(self._pin[2], Pin.OUT)
            self._pin[3] = Pin(self._pin[3], Pin.OUT)
        elif self._interface is FULL3WIRE or self._interface is HALF3WIRE:
            self._pin[2] = Pin(self._pin[2], Pin.OUT)
        if self._enablePin != 0xff:
            self._enablePin = Pin(self._enablePin, Pin.OUT)
            self._enablePin.value(True ^ self._enableInverted)

    def set_min_pulse_width(self, min_width: int) -> None:
        self._minPulseWidth = min_width

    def set_enable_pin(self, enable_pin: int) -> None:
        self._enablePin = enable_pin
        if self._enablePin != 0xff:
            self._enablePin = Pin(self._enablePin, Pin.OUT)
            self._enablePin.value(True ^ self._enableInverted)

    def set_pins_inverted(self, *args) -> None:
        if len(args) == 3:
            self.set_2_pins(args[0], args[1], args[2])
        elif len(args) == 5:
            self.set_4_pins(args[0], args[1], args[2], args[3], args[4])

    def set_2_pins(self, direction_invert: bool, step_invert: bool, enable_invert: bool) -> None:
        self._pinInverted[0] = step_invert
        self._pinInverted[1] = direction_invert
        self._enableInverted = enable_invert

    def set_4_pins(self, pin_1_invert: bool, pin_2_invert: bool, pin_3_invert: bool, pin_4_invert: bool,
                   enable_invert: bool) -> None:
        self._pinInverted[0] = pin_1_invert
        self._pinInverted[1] = pin_2_invert
        self._pinInverted[2] = pin_3_invert
        self._pinInverted[3] = pin_4_invert
        self._enableInverted = enable_invert

    def run_to_position(self) -> None:
        while self.run():
            pass

    def run_speed_to_position(self) -> bool:
        if self._targetPos == self._currentPos:
            return False
        if self._targetPos > self._currentPos:
            self._direction = DIRECTION_CW
        else:
            self._direction = DIRECTION_CCW
        return self.run_speed()

    def run_to_new_position(self, position: int) -> None:
        self.move_to(position)
        self.run_to_position()

    def stop(self) -> None:
        if self._speed != 0.0:
            steps_to_stop = int((self._speed * self._speed) / (2.0 * self._acceleration)) + 1
            if self._speed > 0:
                self.move(steps_to_stop)
            else:
                self.move(-steps_to_stop)

    def is_running(self) -> bool:
        return not (self._speed == 0 and self._targetPos == self._currentPos)

accelstepper.py

#MicroPython SSD1306 OLED driver, I2C and SPI interfaces created by Adafruit

import time
import framebuf

# register definitions
SET_CONTRAST        = const(0x81)
SET_ENTIRE_ON       = const(0xa4)
SET_NORM_INV        = const(0xa6)
SET_DISP            = const(0xae)
SET_MEM_ADDR        = const(0x20)
SET_COL_ADDR        = const(0x21)
SET_PAGE_ADDR       = const(0x22)
SET_DISP_START_LINE = const(0x40)
SET_SEG_REMAP       = const(0xa0)
SET_MUX_RATIO       = const(0xa8)
SET_COM_OUT_DIR     = const(0xc0)
SET_DISP_OFFSET     = const(0xd3)
SET_COM_PIN_CFG     = const(0xda)
SET_DISP_CLK_DIV    = const(0xd5)
SET_PRECHARGE       = const(0xd9)
SET_VCOM_DESEL      = const(0xdb)
SET_CHARGE_PUMP     = const(0x8d)


class SSD1306:
    def __init__(self, width, height, external_vcc):
        self.width = width
        self.height = height
        self.external_vcc = external_vcc
        self.pages = self.height // 8
        # Note the subclass must initialize self.framebuf to a framebuffer.
        # This is necessary because the underlying data buffer is different
        # between I2C and SPI implementations (I2C needs an extra byte).
        self.poweron()
        self.init_display()

    def init_display(self):
        for cmd in (
            SET_DISP | 0x00, # off
            # address setting
            SET_MEM_ADDR, 0x00, # horizontal
            # resolution and layout
            SET_DISP_START_LINE | 0x00,
            SET_SEG_REMAP | 0x01, # column addr 127 mapped to SEG0
            SET_MUX_RATIO, self.height - 1,
            SET_COM_OUT_DIR | 0x08, # scan from COM[N] to COM0
            SET_DISP_OFFSET, 0x00,
            SET_COM_PIN_CFG, 0x02 if self.height == 32 else 0x12,
            # timing and driving scheme
            SET_DISP_CLK_DIV, 0x80,
            SET_PRECHARGE, 0x22 if self.external_vcc else 0xf1,
            SET_VCOM_DESEL, 0x30, # 0.83*Vcc
            # display
            SET_CONTRAST, 0xff, # maximum
            SET_ENTIRE_ON, # output follows RAM contents
            SET_NORM_INV, # not inverted
            # charge pump
            SET_CHARGE_PUMP, 0x10 if self.external_vcc else 0x14,
            SET_DISP | 0x01): # on
            self.write_cmd(cmd)
        self.fill(0)
        self.show()

    def poweroff(self):
        self.write_cmd(SET_DISP | 0x00)

    def contrast(self, contrast):
        self.write_cmd(SET_CONTRAST)
        self.write_cmd(contrast)

    def invert(self, invert):
        self.write_cmd(SET_NORM_INV | (invert & 1))

    def show(self):
        x0 = 0
        x1 = self.width - 1
        if self.width == 64:
            # displays with width of 64 pixels are shifted by 32
            x0 += 32
            x1 += 32
        self.write_cmd(SET_COL_ADDR)
        self.write_cmd(x0)
        self.write_cmd(x1)
        self.write_cmd(SET_PAGE_ADDR)
        self.write_cmd(0)
        self.write_cmd(self.pages - 1)
        self.write_framebuf()

    def fill(self, col):
        self.framebuf.fill(col)

    def pixel(self, x, y, col):
        self.framebuf.pixel(x, y, col)

    def scroll(self, dx, dy):
        self.framebuf.scroll(dx, dy)

    def text(self, string, x, y, col=1):
        self.framebuf.text(string, x, y, col)


class SSD1306_I2C(SSD1306):
    def __init__(self, width, height, i2c, addr=0x3c, external_vcc=False):
        self.i2c = i2c
        self.addr = addr
        self.temp = bytearray(2)
        # Add an extra byte to the data buffer to hold an I2C data/command byte
        # to use hardware-compatible I2C transactions.  A memoryview of the
        # buffer is used to mask this byte from the framebuffer operations
        # (without a major memory hit as memoryview doesn't copy to a separate
        # buffer).
        self.buffer = bytearray(((height // 8) * width) + 1)
        self.buffer[0] = 0x40  # Set first byte of data buffer to Co=0, D/C=1
        self.framebuf = framebuf.FrameBuffer1(memoryview(self.buffer)[1:], width, height)
        super().__init__(width, height, external_vcc)

    def write_cmd(self, cmd):
        self.temp[0] = 0x80 # Co=1, D/C#=0
        self.temp[1] = cmd
        self.i2c.writeto(self.addr, self.temp)

    def write_framebuf(self):
        # Blast out the frame buffer using a single I2C transaction to support
        # hardware I2C interfaces.
        self.i2c.writeto(self.addr, self.buffer)

    def poweron(self):
        pass


class SSD1306_SPI(SSD1306):
    def __init__(self, width, height, spi, dc, res, cs, external_vcc=False):
        self.rate = 10 * 1024 * 1024
        dc.init(dc.OUT, value=0)
        res.init(res.OUT, value=0)
        cs.init(cs.OUT, value=1)
        self.spi = spi
        self.dc = dc
        self.res = res
        self.cs = cs
        self.buffer = bytearray((height // 8) * width)
        self.framebuf = framebuf.FrameBuffer1(self.buffer, width, height)
        super().__init__(width, height, external_vcc)

    def write_cmd(self, cmd):
        self.spi.init(baudrate=self.rate, polarity=0, phase=0)
        self.cs.high()
        self.dc.low()
        self.cs.low()
        self.spi.write(bytearray([cmd]))
        self.cs.high()

    def write_framebuf(self):
        self.spi.init(baudrate=self.rate, polarity=0, phase=0)
        self.cs.high()
        self.dc.high()
        self.cs.low()
        self.spi.write(self.buffer)
        self.cs.high()

    def poweron(self):
        self.res.high()
        time.sleep_ms(1)
        self.res.low()
        time.sleep_ms(10)
        self.res.high()

ssd1306.py

import machine
from machine import Pin, SoftI2C
import ssd1306
import math

# ESP32 Pin assignment 
i2c = SoftI2C(scl=Pin(22), sda=Pin(21))

oled_width = 128
oled_height = 64
oled = ssd1306.SSD1306_I2C(oled_width, oled_height, i2c)

def draw_screen():
  oled.fill(0)
  oled.text("Clever. Props", 12, 57, 1)
  oled.text("Select speed", 10, 1, 1)
  oled.text("and direction", 10, 10, 1)
  draw_logo()
  oled.show()
  
def draw_logo():
    r = 12
    yc = 37
    for angle in range(0, 90, 2):
        y3 = int(r * math.sin(math.radians(angle)))
        x3 = int(r * math.cos(math.radians(angle)))
        oled.pixel(r + 4 - x3, yc + y3, 1)
        oled.pixel(r + 4 - x3, yc - y3, 1)
        oled.pixel(r + 4 + x3, yc + y3, 1)
        oled.pixel(r + 4 + x3, yc - y3, 1)
    for angle in range(0, 90, 2):
        y3 = int((r / 3 * 2) * math.sin(math.radians(angle)))
        x3 = int((r / 3 * 2) * math.cos(math.radians(angle)))
        oled.pixel(r + 4 - x3, yc + 4 + y3, 1)
        oled.pixel(r + 4 - x3, yc + 4 - y3, 1)
        oled.pixel(r + 4 + x3, yc + 4 + y3, 1)
        oled.pixel(r + 4 + x3, yc + 4 - y3, 1)
    for angle in range(0, 90, 2):
        y3 = int((r / 2) * math.sin(math.radians(angle)))
        x3 = int((r / 2) * math.cos(math.radians(angle)))
        oled.pixel(r + 4 - x3, yc + 6 + y3, 1)
        oled.pixel(r + 4 - x3, yc + 6 - y3, 1)
        oled.pixel(r + 4 + x3, yc + 6 + y3, 1)
        oled.pixel(r + 4 + x3, yc + 6 - y3, 1)

def slow(spd):
  draw_screen()
  oled.text("{0} RPM".format(spd), 35, 25, 1)
  oled.text("Slow", 35, 38, 1)
  oled.show()

def medium(spd):
  draw_screen()
  oled.text("{0} RPM".format(spd), 35, 25, 1)
  oled.text("Medium", 35, 38, 1)
  oled.show()
  
def fast(spd):
  draw_screen()
  oled.text("{0} RPM".format(spd), 35, 25, 1)
  oled.text("Fast", 35, 38, 1)
  oled.show()
  
def setCustom(spd):
  oled.fill(0)
  oled.text("Clever. Props", 22, 57, 1)
  oled.text("Use dial to adjust speed./nClick dial to set.", 1, 1, 1)
  oled.text("{0} RPM".format(spd), 35, 25, 1)
  draw_logo()
  oled.show()

def custom(spd):
  draw_screen()
  oled.text("{0} RPM".format(spd), 35, 25, 1)
  oled.text("Custom", 35, 38, 1)
  oled.show()

def finished():
  oled.fill(1)
  oled.text("Clever. Props", 5, 20, 0)
  oled.show()


display.py

# The 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


class RotaryIRQ(Rotary):

    def __init__(self, pin_num_clk, pin_num_dt, min_val=0, max_val=10,
                 reverse=False, range_mode=Rotary.RANGE_UNBOUNDED, pull_up=False, half_step=False, invert=False):

        super().__init__(min_val, max_val, 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()

rotaryirq.py

# The 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, reverse, range_mode, half_step, invert):
        self._min_val = min_val
        self._max_val = max_val
        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,
            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 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 = 1
        elif direction == _DIR_CCW:
            incr = -1

        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:
                micropython.schedule(_trigger, self)
        except:
            pass