[FireBeetle 2 ESP32 P4 Development Kit] Home Assistant IoT Stepper Motor Control

import json
import time
import random
import network
from machine import Pin
import uasyncio as asyncio
from umqtt.simple import MQTTClient

# ------------- User Configuration -------------
WIFI_SSID = "Xiaomi_8ABD"
WIFI_PWD = "15836035036"
MQTT_SERVER = "192.168.31.160"
MQTT_PORT   = 32768
MQTT_CLIENT_ID = 'micropython-client-{id}'.format(id=random.getrandbits(8))
MQTT_USER   = "LJL"
MQTT_PWD   = "4421989g"
SUB_TOPIC = "esp/stepper/cmd"
PUB_TOPIC = "esp/stepper/status"
# Motor pins (A B C D)
MOTOR_PINS = [4, 5, 20, 21]
# ------------------------------------

def wifi_connect(ssid, pwd):
    sta = network.WLAN(network.STA_IF)
    sta.active(True)
    if not sta.isconnected():
        print("Connecting Wi-Fi...")
        sta.connect(ssid, pwd)
        for _ in range(20):
            if sta.isconnected():
                break
            time.sleep(1)
    print("Wi-Fi connected:", sta.ifconfig())

# --------------Stepper-------------------------------------
class Stepper28BYJ48:
    STEP_SEQ = [
        [1,0,0,1], [1,0,0,0], [1,1,0,0], [0,1,0,0],
        [0,1,1,0], [0,0,1,0], [0,0,1,1], [0,0,0,1]
    ]
    
    def __init__(self, pins):
        self.coils = [Pin(p, Pin.OUT) for p in pins]
        self.release()

    def _write(self, pattern):
        for coil, v in zip(self.coils, pattern):
            coil.value(v)

    def step(self, steps, delay_ms=2):
        direction = 1 if steps > 0 else -1
        for _ in range(abs(steps)):
            for phase in range(8)[::direction]:
                self._write(self.STEP_SEQ[phase])
                time.sleep_ms(delay_ms)
        self.release()

    def angle(self, deg, delay_ms=2):
        steps = int(deg * (509 / 360))
        self.step(steps, delay_ms)

    def release(self):
        for c in self.coils:
            c.value(0)

motor = Stepper28BYJ48(MOTOR_PINS)

# Connect MQTT
def mqtt_connect():
    c = MQTTClient(MQTT_CLIENT_ID, MQTT_SERVER, MQTT_PORT, MQTT_USER, MQTT_PWD)
    c.set_callback(on_msg)
    c.connect()
    c.subscribe(SUB_TOPIC)
    print("[MQTT] connected & subscribed")
    return c

# Receive control command
def on_msg(topic, msg):
    try:
        cmd = json.loads(msg)
        angle = cmd.get("angle", 0)
        speed = max(1, min(10, cmd.get("speed", 2)))  # Limit speed range 1-10
        delay_ms = 11 - speed  # Higher value = faster speed (less delay)
        motor.angle(angle, delay_ms)

        # Return status
        stat = {"angle_done": angle, "speed_set": speed}
        client.publish(PUB_TOPIC, json.dumps(stat))
    except Exception as e:
        print("Bad msg:", e)

# Main coroutine
async def main():
    wifi_connect(WIFI_SSID, WIFI_PWD)
    global client
    client = mqtt_connect()

    while True:
        client.check_msg()
        await asyncio.sleep_ms(100)

asyncio.run(main())

mqtt:
  number:
    - name: "Step Motor"
      unique_id: stepper_angle
      command_topic: "esp/stepper/cmd"
      state_topic: "esp/stepper/status"
      value_template: "{{ value_json.angle_done }}"
      min: -360
      max: 360
      step: 1
      unit_of_measurement: "°"
      optimistic: false
  button:
    - name: "CW 90°"
      command_topic: "esp/stepper/cmd"
      payload_press: '{"angle":90,"speed":10}'
      unique_id: stepper_cw_90
    - name: "CCW 90°"
      command_topic: "esp/stepper/cmd"
      payload_press: '{"angle":-90,"speed":10}'
      unique_id: stepper_ccw_90