Precision A Practical Guide to For Makers

#include <Arduino.h>
#include <Grbl_ESP32.h>  // Grbl 1.1 port for ESP32, https://github.com/mitcdh/Grbl_Esp32
#include <AS5600.h>      // AMS AS5600 magnetic encoder library, https://github.com/robTillaart/AS5600
#include <Wire.h>
#include <SPI.h>

// Configuration constants
#define STEP_PIN 12       // GPIO12 for step pulses
#define DIR_PIN 13        // GPIO13 for direction
#define EN_PIN 14         // GPIO14 for motor enable
#define ENCODER_ADDR 0x36 // I2C address for AS5600
#define STEPS_PER_REV 200 // 1.8° stepper motor (200 steps/rev)
#define MICROSTEPPING 16  // 16x microstepping
#define TOLERANCE 2       // Allowed encoder deviation (raw counts, ~0.175°)

AS5600 encoder;
GrblEsp32 grbl;

// Track last encoder position for drift detection
volatile int32_t lastEncoderPos = 0;
volatile int32_t targetSteps = 0;
volatile bool motorEnabled = false;

void IRAM_ATTR stepInterrupt() {
  // Increment/decrement target steps based on direction pin state
  if (digitalRead(DIR_PIN) == HIGH) {
    targetSteps++;
  } else {
    targetSteps--;
  }
}

void setup() {
  Serial.begin(115200);
  while (!Serial) delay(10); // Wait for serial connection on native USB

  // Initialize I2C for encoder
  Wire.begin(21, 22); // SDA=21, SCL=22 on ESP32-S3
  if (!encoder.begin(ENCODER_ADDR, &Wire)) {
    Serial.println("[ERROR] AS5600 encoder not detected. Check wiring.");
    while (1) delay(100); // Halt execution on critical failure
  }
  encoder.setDirection(AS5600_CLOCK_WISE); // Match motor rotation direction

  // Configure stepper pins
  pinMode(STEP_PIN, INPUT_PULLDOWN); // Step pin as input for interrupt
  pinMode(DIR_PIN, OUTPUT);
  pinMode(EN_PIN, OUTPUT);
  digitalWrite(EN_PIN, HIGH); // Disable motor by default (active low)

  // Attach step interrupt to count commanded steps
  attachInterrupt(digitalPinToInterrupt(STEP_PIN), stepInterrupt, RISING);

  // Initialize Grbl with custom config
  grbl.begin();
  grbl.setMicrostepping(MICROSTEPPING);
  grbl.setStepsPerMm(100.0); // 100 steps/mm for leadscrew with 2mm pitch

  Serial.println("[INFO] Closed-loop controller initialized. Waiting for commands...");
}

void loop() {
  // Process Grbl commands (non-blocking)
  grbl.run();

  // Read current encoder position (12-bit raw value: 0-4095)
  int32_t currentEncoderPos = encoder.rawAngle();

  // Calculate expected encoder position based on commanded steps
  // 4096 raw counts per full encoder revolution, 200*16=3200 steps/rev
  float stepsPerEncoderRev = STEPS_PER_REV * MICROSTEPPING;
  float encoderCountsPerStep = 4096.0 / stepsPerEncoderRev;
  int32_t expectedEncoderPos = (targetSteps * encoderCountsPerStep) % 4096;

  // Calculate deviation
  int32_t deviation = abs(currentEncoderPos - expectedEncoderPos);

  // Handle encoder wrap-around (4095 to 0 transition)
  if (deviation > 2048) deviation = 4096 - deviation;

  // Correct position if deviation exceeds tolerance
  if (deviation > TOLERANCE && motorEnabled) {
    Serial.printf("[WARN] Position deviation detected: %d counts. Correcting...\n", deviation);
    // Simple P-controller for correction (proportional gain = 0.5)
    int32_t correctionSteps = (deviation / encoderCountsPerStep) * 0.5;
    if (currentEncoderPos > expectedEncoderPos) {
      // Move reverse to correct
      digitalWrite(DIR_PIN, LOW);
      for (int i = 0; i < correctionSteps; i++) {
        digitalWrite(STEP_PIN, HIGH);
        delayMicroseconds(500);
        digitalWrite(STEP_PIN, LOW);
        delayMicroseconds(500);
      }
    } else {
      // Move forward to correct
      digitalWrite(DIR_PIN, HIGH);
      for (int i = 0; i < correctionSteps; i++) {
        digitalWrite(STEP_PIN, HIGH);
        delayMicroseconds(500);
        digitalWrite(STEP_PIN, LOW);
        delayMicroseconds(500);
      }
    }
    targetSteps += correctionSteps; // Update target to match corrected position
  }

  // Update motor enabled state from Grbl
  motorEnabled = !digitalRead(EN_PIN); // EN pin is active low

  delay(10); // 10ms loop interval for real-time correction
}

import numpy as np
import scipy.optimize as opt
import serial
import time
import csv
from typing import List, Tuple

# Configuration
SERIAL_PORT = "/dev/ttyUSB0"
BAUD_RATE = 115200
CALIBRATION_POINTS = 10 # Number of points to measure per axis
AXIS_RANGE = 100.0 # Calibration range in mm (0-100mm)
TOLERANCE = 0.005 # Acceptable error before flagging

class CNCProbe:
    def __init__(self, port: str, baud: int):
        self.ser = None
        self.port = port
        self.baud = baud
        self.connect()

    def connect(self) -> None:
        """Establish serial connection to Grbl controller with retry logic"""
        max_retries = 3
        for attempt in range(max_retries):
            try:
                self.ser = serial.Serial(self.port, self.baud, timeout=2)
                time.sleep(2) # Wait for Grbl to initialize
                self.ser.flushInput()
                self.ser.write(b"\r\n\r\n") # Wake up Grbl
                time.sleep(1)
                response = self.ser.read_all().decode()
                if "Grbl" in response:
                    print(f"[INFO] Connected to Grbl on {self.port}")
                    return
                else:
                    raise ConnectionError(f"Unexpected response: {response}")
            except Exception as e:
                print(f"[WARN] Connection attempt {attempt+1} failed: {e}")
                if attempt == max_retries -1:
                    raise RuntimeError(f"Failed to connect after {max_retries} attempts")
                time.sleep(1)

    def send_command(self, cmd: str, wait_for_ok: bool = True) -> str:
        """Send G-code command to Grbl with error handling"""
        if not self.ser or not self.ser.is_open:
            raise ConnectionError("Serial port not open")
        self.ser.write(f"{cmd}\n".encode())
        response = ""
        if wait_for_ok:
            while "ok" not in response:
                response += self.ser.readline().decode()
                if "error" in response:
                    raise ValueError(f"Grbl error: {response}")
        return response

    def probe_position(self, axis: str, target: float) -> float:
        """Move to target position and return actual measured position via touch probe"""
        # Rapid move to 2mm above target to avoid crashing
        self.send_command(f"G0 {axis}{target - 2.0}")
        # Slow probe to target
        self.send_command(f"G38.2 {axis}{target} F100") # Probe at 100mm/min
        # Read probe position from Grbl
        self.send_command("M114") # Get current position
        response = self.ser.readline().decode()
        # Parse position (format: X:10.000 Y:20.000 Z:30.000)
        for part in response.split():
            if part.startswith(axis.upper()):
                return float(part.split(":")[1])
        raise ValueError(f"Failed to parse position from response: {response}")

    def close(self) -> None:
        if self.ser and self.ser.is_open:
            self.ser.close()

def collect_calibration_data(probe: CNCProbe, axis: str) -> Tuple[List[float], List[float]]:
    """Collect commanded vs actual position data for calibration"""
    commanded = []
    actual = []
    step = AXIS_RANGE / (CALIBRATION_POINTS - 1)
    for i in range(CALIBRATION_POINTS):
        target = i * step
        commanded.append(target)
        print(f"[INFO] Probing {axis} axis at {target:.3f}mm...")
        try:
            measured = probe.probe_position(axis, target)
            actual.append(measured)
            error = abs(measured - target)
            if error > TOLERANCE:
                print(f"[WARN] High error at {target}mm: {error:.4f}mm")
        except Exception as e:
            print(f"[ERROR] Failed to probe {target}mm: {e}")
            actual.append(target) # Fallback to commanded if probe fails
    return commanded, actual

def calculate_calibration(commanded: List[float], actual: List[float]) -> np.poly1d:
    """Fit linear calibration model (y = mx + b) to correct for systematic error"""
    # Use linear regression to find best fit
    coeffs = np.polyfit(actual, commanded, 1) # Fit actual -> commanded
    return np.poly1d(coeffs)

def save_calibration(calib: np.poly1d, axis: str) -> None:
    """Save calibration coefficients to CSV for Grbl to load"""
    with open(f"calibration_{axis}.csv", "w", newline="") as f:
        writer = csv.writer(f)
        writer.writerow(["coefficient_m", "coefficient_b"])
        writer.writerow([calib.coeffs[0], calib.coeffs[1]])
    print(f"[INFO] Saved calibration for {axis} axis to calibration_{axis}.csv")

if __name__ == "__main__":
    probe = None
    try:
        probe = CNCProbe(SERIAL_PORT, BAUD_RATE)
        # Home all axes before calibration
        probe.send_command("G28") # Home all axes
        time.sleep(5) # Wait for homing to complete

        for axis in ["x", "y", "z"]:
            print(f"\n=== Calibrating {axis.upper()} Axis ===")
            commanded, actual = collect_calibration_data(probe, axis)
            calib = calculate_calibration(commanded, actual)
            save_calibration(calib, axis)
            # Print calibration stats
            errors = np.array(commanded) - calib(np.array(actual))
            print(f"Max error after calibration: {np.max(np.abs(errors)):.4f}mm")
            print(f"Mean error after calibration: {np.mean(errors):.4f}mm")
    except Exception as e:
        print(f"[CRITICAL] Calibration failed: {e}")
    finally:
        if probe:
            probe.close()

import paho.mqtt.client as mqtt
import influxdb_client
from influxdb_client.client.write_api import SYNCHRONOUS
import statistics
import json
import logging
from typing import Dict, List
from datetime import datetime

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s"
)
logger = logging.getLogger(__name__)

# Configuration
MQTT_BROKER = "192.168.1.100"
MQTT_PORT = 1883
MQTT_TOPIC = "precision/encoder/+/deviation"
INFLUX_URL = "http://localhost:8086"
INFLUX_TOKEN = "your-influx-token-here"
INFLUX_ORG = "maker-precision"
INFLUX_BUCKET = "precision-metrics"
CONTROL_LIMIT = 3.0 # 3-sigma control limit for deviation (mm)
WINDOW_SIZE = 30 # Number of samples for moving average

# InfluxDB client setup
influx_client = influxdb_client.InfluxDBClient(
    url=INFLUX_URL,
    token=INFLUX_TOKEN,
    org=INFLUX_ORG
)
write_api = influx_client.write_api(write_options=SYNCHRONOUS)

# Store moving window of deviations per axis
deviation_windows: Dict[str, List[float]] = {
    "x": [],
    "y": [],
    "z": []
}
out_of_control_events: List[Dict] = []

def on_connect(client, userdata, flags, rc):
    if rc == 0:
        logger.info("Connected to MQTT broker")
        client.subscribe(MQTT_TOPIC)
    else:
        logger.error(f"MQTT connection failed with code {rc}")

def on_message(client, userdata, msg):
    try:
        # Parse incoming deviation data
        payload = json.loads(msg.payload.decode())
        axis = msg.topic.split("/")[2] # Topic: precision/encoder/x/deviation
        deviation = float(payload["deviation_mm"])
        timestamp = datetime.utcnow()

        # Update moving window
        window = deviation_windows[axis]
        window.append(deviation)
        if len(window) > WINDOW_SIZE:
            window.pop(0)

        # Calculate SPC metrics if window is full
        if len(window) == WINDOW_SIZE:
            mean = statistics.mean(window)
            stdev = statistics.stdev(window) if len(window) > 1 else 0.0
            ucl = mean + (CONTROL_LIMIT * stdev) # Upper Control Limit
            lcl = mean - (CONTROL_LIMIT * stdev) # Lower Control Limit

            # Check if current deviation is out of control
            if deviation > ucl or deviation < lcl:
                event = {
                    "timestamp": timestamp.isoformat(),
                    "axis": axis,
                    "deviation": deviation,
                    "mean": mean,
                    "stdev": stdev,
                    "ucl": ucl,
                    "lcl": lcl
                }
                out_of_control_events.append(event)
                logger.warning(f"Out of control! Axis {axis}: {deviation:.4f}mm (UCL: {ucl:.4f}, LCL: {lcl:.4f})")

            # Write metrics to InfluxDB
            point = influxdb_client.Point("precision_deviation") \
                .tag("axis", axis) \
                .field("deviation", deviation) \
                .field("mean", mean) \
                .field("stdev", stdev) \
                .field("ucl", ucl) \
                .field("lcl", lcl) \
                .time(timestamp)

            write_api.write(bucket=INFLUX_BUCKET, org=INFLUX_ORG, record=point)
            logger.debug(f"Wrote {axis} deviation: {deviation:.4f}mm")

    except json.JSONDecodeError:
        logger.error("Failed to decode MQTT payload")
    except KeyError as e:
        logger.error(f"Missing key in payload: {e}")
    except Exception as e:
        logger.error(f"Message processing failed: {e}")

def generate_spc_report():
    """Generate daily SPC report from out of control events"""
    if not out_of_control_events:
        logger.info("No out of control events today.")
        return
    report = "=== Daily SPC Report ===\n"
    for event in out_of_control_events:
        report += f"Time: {event['timestamp']}\n"
        report += f"Axis: {event['axis']}\n"
        report += f"Deviation: {event['deviation']:.4f}mm\n"
        report += f"Mean: {event['mean']:.4f}mm, Stdev: {event['stdev']:.4f}mm\n"
        report += "---\n"
    with open(f"spc_report_{datetime.now().strftime('%Y%m%d')}.txt", "w") as f:
        f.write(report)
    logger.info(f"Generated SPC report with {len(out_of_control_events)} events")

if __name__ == "__main__":
    # MQTT client setup
    mqtt_client = mqtt.Client()
    mqtt_client.on_connect = on_connect
    mqtt_client.on_message = on_message

    try:
        mqtt_client.connect(MQTT_BROKER, MQTT_PORT, 60)
        logger.info("Starting SPC monitoring loop...")
        mqtt_client.loop_forever()
    except KeyboardInterrupt:
        logger.info("Stopping monitoring...")
        generate_spc_report()
    except Exception as e:
        logger.error(f"Fatal error: {e}")
    finally:
        mqtt_client.disconnect()
        influx_client.close()

// Read current encoder position (12-bit raw value: 0-4095)
int32_t currentEncoderPos = encoder.rawAngle();

// Calculate expected encoder position based on commanded steps
float stepsPerEncoderRev = STEPS_PER_REV * MICROSTEPPING;
float encoderCountsPerStep = 4096.0 / stepsPerEncoderRev;
int32_t expectedEncoderPos = (targetSteps * encoderCountsPerStep) % 4096;

def calculate_calibration(commanded: List[float], actual: List[float]) -> np.poly1d:
    """Fit linear calibration model (y = mx + b) to correct for systematic error"""
    # Use linear regression to find best fit
    coeffs = np.polyfit(actual, commanded, 1) # Fit actual -> commanded
    return np.poly1d(coeffs)

if len(window) == WINDOW_SIZE:
    mean = statistics.mean(window)
    stdev = statistics.stdev(window) if len(window) > 1 else 0.0
    ucl = mean + (CONTROL_LIMIT * stdev) # Upper Control Limit
    lcl = mean - (CONTROL_LIMIT * stdev) # Lower Control Limit

    # Check if current deviation is out of control
    if deviation > ucl or deviation < lcl:
        event = {
            "timestamp": timestamp.isoformat(),
            "axis": axis,
            "deviation": deviation,
            "mean": mean,
            "stdev": stdev,
            "ucl": ucl,
            "lcl": lcl
        }
        out_of_control_events.append(event)

maker-precision-guide/
├── firmware/
│   ├── esp32-closed-loop/       # ESP32 Grbl closed-loop firmware
│   │   ├── src/
│   │   │   └── main.cpp         # Code from Example 1
│   │   ├── platformio.ini       # PlatformIO config for ESP32-S3
│   │   └── README.md            # Flashing instructions
├── calibration/
│   ├── cnc-calibrate.py         # Code from Example 2
│   ├── requirements.txt        # Python dependencies
│   └── sample-data/            # Pre-collected calibration data
├── monitoring/
│   ├── spc-monitor.py           # Code from Example 3
│   ├── node-red-flow.json       # Node-RED dashboard flow
│   └── grafana-dashboard.json   # Pre-built Grafana dashboard
├── case-studies/
│   └── pcb-milling/             # Case study data and configs
└── README.md                    # Full setup instructions