astra/nema17-stepper-motor-controller-spring-clamp-rp2040

firmware/thermal_guard.py

"""MicroPython RP2040 adapter. No automatic start or automatic fault restart."""
from machine import Pin, I2C, disable_irq, enable_irq
from time import sleep_ms
from tmp102 import TMP102
from thermal_policy import ThermalPolicy
from stepper_driver import StepperDriver


class ThermalGuard:
    def __init__(self):
        self.enable = Pin(22, Pin.OUT, value=0)
        self.driver = StepperDriver()
        self.alert = Pin(24, Pin.IN)
        self.driver_fault = Pin(23, Pin.IN)
        self.policy = ThermalPolicy()
        self.irq_trip = 0
        self.configured = False
        self.sensor = TMP102(I2C(1, sda=Pin(26), scl=Pin(27), freq=100_000))
        self.alert.irq(handler=self._temperature_irq, trigger=Pin.IRQ_FALLING, hard=True)
        try:
            self.sensor.configure()
            # Exceed the 125 ms conversion interval plus conversion time.
            sleep_ms(300)
            self.configured = True
        except (OSError, ValueError):
            self.policy.trip("temperature sensor initialization failed")
        self.poll()

    def _temperature_irq(self, _pin):
        # Hardware gate already holds nSLEEP low; capture it until manual rearm.
        self.enable.value(0)
        self.irq_trip = 1

    def stop(self, reason="user stop"):
        self.enable.value(0)
        self.policy.trip(reason)
        self.driver.disable()

    def poll(self):
        try:
            if not self.configured:
                raise OSError("Sensor not configured; reboot to retry")
            temp = self.sensor.temperature()
            self.policy.sample(temp, bool(self.alert.value()),
                               driver_fault=self.policy.enabled and not self.driver_fault.value())
        except (OSError, ValueError):
            self.policy.sample(None, sensor_ok=False)
        if self.irq_trip:
            self.policy.trip("hardware overtemperature alert")
        if not self.policy.enabled:
            self.enable.value(0)
            self.driver.disable()
        return self.policy.status()

    def arm(self):
        # Always take a fresh verified reading. A stale last sample cannot arm.
        self.poll()
        self.driver.disable()
        # Avoid an I2C transaction with interrupts disabled. The critical section
        # protects the final check and enable write against a concurrent alert.
        irq_state = disable_irq()
        try:
            if not self.alert.value() or not self.policy.arm():
                return False
            self.irq_trip = 0
            self.enable.value(1)
        finally:
            enable_irq(irq_state)
        sleep_ms(3)  # DRV8825 wake requirement is 1.7 ms; outputs remain disabled.
        self.poll()
        irq_state = disable_irq()
        try:
            if self.policy.enabled and not self.irq_trip and self.alert.value() and self.driver_fault.value():
                self.driver.enable()
            else:
                self.stop("driver not ready after wake")
        finally:
            enable_irq(irq_state)
        return self.policy.enabled

    def step(self, forward=True):
        """One guarded 1/16 microstep. No movement is generated automatically."""
        return self.driver.pulse(forward, self._motion_permitted)

    def _motion_permitted(self):
        return (self.policy.enabled and not self.irq_trip and
                bool(self.alert.value()) and bool(self.driver_fault.value()))