LibStepper.py

#!/usr/bin/env python3

import serial
import time

params = {
    "speed": 3000,
    "accel": 10000,
    "offset0": 140,
    "offset1": 0,
}


class StepperError(Exception):
    """Custom exception for motor hardware/protocol errors."""
    pass


class ControllerAPI:
    def __init__(self, port, baudrate=9600):
        self.ser = serial.Serial(port, baudrate, timeout=0.1)
        time.sleep(2)  # Bootloader delay

    def _execute(self, cmd, wait=True, timeout=10.0):
        self.ser.reset_input_buffer()
        self.ser.write(f"{cmd}\n".encode())

        if not wait:
            return None

        start = time.time()
        while (time.time() - start) < timeout:
            if self.ser.in_waiting:
                line = self.ser.readline().decode().strip()
                if not line:
                    continue

                # --- THE MAGIC HAPPENS HERE ---
                if line.startswith("ERR"):
                    raise StepperError(f"Hardware error on '{cmd}': {line}")

                return line
            time.sleep(0.01)
        raise TimeoutError(f"Command '{cmd}' timed out.")

    def poll_response(self):
        """Checks if the Arduino has sent a response yet. Returns None if still moving."""
        if self.ser.in_waiting:
            line = self.ser.readline().decode().strip()
            if not line:
                return None

            if line.startswith("ERR"):
                raise StepperError(f"Deferred Hardware Error: {line}")

            return line  # Returns "OK GOTO_A", etc.
        return None

    def await_response(self, timeout=5):
        """awaits for response and returns."""
        start = time.time()
        while (time.time() - start) < timeout:
            response = self.poll_response()
            if response is not None:
                return response
            time.sleep(0.001)

        raise TimeoutError(f"await timed out.")

    def reset(self, timeout=20.0):
        raw = self._execute("R", wait=True, timeout=timeout)
        assert raw is not None, "raw was none."
        parts = raw.split()
        # Even if it returns "OK", we might want to fail on "BROKEN"
        if parts[1] == "BROKEN":
            raise StepperError(f"Mechanical Failure: {raw}")

        return {"status": parts[1], "steps": int(parts[2])}

    def set_speeds(self, speed, accel):
        return self._execute(f"S {speed} {accel}", wait=True)

    def set_offsets(self, ofset0, ofset1, timeout=2.0):
        return self._execute(f"O {ofset0} {ofset1}", True)

    def disable(self, timeout=2.0):
        return self._execute("D", True)

    def goto_a(self, wait=True):
        return self._execute("A", wait=wait)

    def goto_b(self, wait=True):
        return self._execute("B", wait=wait)

    def goto(self, position, wait=True):
        return self._execute(f"G {position}", wait=wait)

    def LED_on(self, timeout=2.0):
        return self._execute("L", True)

    def LED_off(self, timeout=2.0):
        return self._execute("F", wait=True)

    def applysettings(self, params=params):
        try:
            self.LED_off()
            self.set_speeds(params["speed"], params["accel"])
            self.set_offsets(params["offset0"], params["offset1"])

            # Reset returns a dict, but will raise StepperError if 'BROKEN'
            res = self.reset()
            print(
                f"Homing complete: {res['status']} ({res['steps']} steps off)")

            self.goto(0)
            time.sleep(.5)

        except (StepperError, TimeoutError) as e:
            print(f"Failed to apply settings: {e}")
            # Handle emergency stop or logging here
            raise

    def cleanup(self):
        self.disable()
        self.LED_off()