How to Filament Runout Sensor A Step-by-Step Guide

import machine
import time
import ubinascii
from machine import Pin, Timer
from micropython import const

# Constants for optical interrupted beam sensor (Omron EE-SX671)
const(TRIGGER_THRESHOLD = 0.5)  # 50% duty cycle threshold for valid signal
const(DEBOUNCE_MS = 50)  # 50ms debounce to filter mechanical bounce
const(SENSOR_PIN = 18)  # GPIO18 for optical sensor input
const(LED_PIN = 2)  # Onboard LED for status indication
const(ERROR_PIN = 19)  # GPIO19 for error output to mainboard
const(WATCHDOG_TIMEOUT_MS = 1000)  # 1s watchdog to detect stuck signals

class FilamentRunoutSensor:
    def __init__(self):
        # Initialize sensor pin with pull-up resistor (sensor is active low)
        self.sensor_pin = Pin(SENSOR_PIN, Pin.IN, Pin.PULL_UP)
        self.led_pin = Pin(LED_PIN, Pin.OUT)
        self.error_pin = Pin(ERROR_PIN, Pin.OUT)
        self.last_state = 1  # Default: filament present (sensor not triggered)
        self.last_change_ms = time.ticks_ms()
        self.watchdog = Timer(0)
        self.watchdog.init(period=WATCHDOG_TIMEOUT_MS, mode=Timer.PERIODIC, callback=self._watchdog_callback)
        self.sensor_pin.irq(trigger=Pin.IRQ_FALLING | Pin.IRQ_RISING, handler=self._sensor_irq)
        self.error_flag = False
        self.runout_flag = False

    def _sensor_irq(self, pin):
        # Disable IRQ during processing to prevent re-entry
        self.sensor_pin.irq(handler=None)
        try:
            current_state = pin.value()
            current_ms = time.ticks_ms()
            # Check for debounce: ignore changes within DEBOUNCE_MS
            if time.ticks_diff(current_ms, self.last_change_ms) < DEBOUNCE_MS:
                return
            # Validate signal: active low, so 0 = filament absent (sensor triggered)
            if current_state != self.last_state:
                self.last_state = current_state
                self.last_change_ms = current_ms
                if current_state == 0:
                    self.runout_flag = True
                    self.led_pin.value(1)  # LED on = runout
                    self.error_pin.value(0)  # Active low error to mainboard
                else:
                    self.runout_flag = False
                    self.led_pin.value(0)  # LED off = filament present
                    self.error_pin.value(1)  # Inactive
        except Exception as e:
            # Log error to serial, set error flag
            print(f"IRQ Error: {ubinascii.hexlify(bytes(str(e), 'utf-8'))}")
            self.error_flag = True
            self.error_pin.value(0)
        finally:
            # Re-enable IRQ
            self.sensor_pin.irq(trigger=Pin.IRQ_FALLING | Pin.IRQ_RISING, handler=self._sensor_irq)

    def _watchdog_callback(self, timer):
        # Check if sensor pin is stuck (no change in 2x watchdog period)
        if time.ticks_diff(time.ticks_ms(), self.last_change_ms) > 2 * WATCHDOG_TIMEOUT_MS:
            print("Watchdog: Sensor signal stuck, possible hardware failure")
            self.error_flag = True
            self.error_pin.value(0)

    def read_state(self):
        # Return current state with error handling
        if self.error_flag:
            raise RuntimeError("Sensor hardware error detected")
        return {
            "filament_present": not self.runout_flag,
            "raw_pin_state": self.sensor_pin.value(),
            "last_change_ms": self.last_change_ms,
            "error_flag": self.error_flag
        }

    def clear_error(self):
        self.error_flag = False
        self.error_pin.value(1)
        self.last_change_ms = time.ticks_ms()

if __name__ == "__main__":
    sensor = FilamentRunoutSensor()
    print("Filament Runout Sensor Initialized. Monitoring GPIO18...")
    while True:
        try:
            state = sensor.read_state()
            print(f"State: {state['filament_present']}, Raw: {state['raw_pin_state']}")
            time.sleep(1)
        except RuntimeError as e:
            print(f"Fatal Error: {e}")
            sensor.clear_error()
            time.sleep(5)
        except Exception as e:
            print(f"Unexpected error: {e}")
            time.sleep(1)

// Marlin 2.1.2.4 Filament Runout Sensor Handler with Dual Redundancy
// Target: STM32F103RET6 (Creality Ender 3 S1 Pro Mainboard)
// Includes: Dual sensor support, debounce, error logging to SD card

#include "MarlinCore.h"
#include "feature/pause.h"
#include "sd/cardreader.h"
#include "module/printcounter.h"
#include 
#include 
#include 

// Configuration: Uncomment to enable dual sensor redundancy
#define DUAL_SENSOR_ENABLE
#define PRIMARY_SENSOR_PIN PC15  // Optical sensor 1 (Omron EE-SX671)
#define SECONDARY_SENSOR_PIN PC14  // Optical sensor 2 (redundant)
#define RUNOUT_DEBOUNCE_MS 50  // Match hardware debounce
#define SENSOR_ACTIVE_LOW true  // Sensors are active low (triggered = 0)
#define ERROR_LOG_FILE "runout_errors.txt"

// Global state variables
bool primary_runout_flag = false;
bool secondary_runout_flag = false;
bool sensor_error_flag = false;
uint32_t last_primary_change_ms = 0;
uint32_t last_secondary_change_ms = 0;
uint32_t total_runout_events = 0;

// Initialize sensor pins
void filament_runout_init() {
  #ifdef DUAL_SENSOR_ENABLE
    pinMode(PRIMARY_SENSOR_PIN, INPUT_PULLUP);
    pinMode(SECONDARY_SENSOR_PIN, INPUT_PULLUP);
    // Set initial state
    last_primary_change_ms = millis();
    last_secondary_change_ms = millis();
    SERIAL_ECHOLNPGM("Dual Filament Runout Sensor Initialized");
  #else
    pinMode(PRIMARY_SENSOR_PIN, INPUT_PULLUP);
    last_primary_change_ms = millis();
    SERIAL_ECHOLNPGM("Single Filament Runout Sensor Initialized");
  #endif
  sensor_error_flag = false;
}

// Read primary sensor with debounce
bool read_primary_sensor() {
  uint32_t current_ms = millis();
  bool raw_state = digitalRead(PRIMARY_SENSOR_PIN);
  // Apply active low logic
  bool triggered = SENSOR_ACTIVE_LOW ? !raw_state : raw_state;
  // Debounce check
  if (triggered != primary_runout_flag) {
    if (current_ms - last_primary_change_ms >= RUNOUT_DEBOUNCE_MS) {
      primary_runout_flag = triggered;
      last_primary_change_ms = current_ms;
      // Log state change to serial
      SERIAL_ECHOPAIR("Primary Sensor State: ", triggered ? "RUNOUT" : "PRESENT");
      SERIAL_EOL();
    }
  } else {
    last_primary_change_ms = current_ms;
  }
  return primary_runout_flag;
}

#ifdef DUAL_SENSOR_ENABLE
// Read secondary sensor with debounce
bool read_secondary_sensor() {
  uint32_t current_ms = millis();
  bool raw_state = digitalRead(SECONDARY_SENSOR_PIN);
  bool triggered = SENSOR_ACTIVE_LOW ? !raw_state : raw_state;
  if (triggered != secondary_runout_flag) {
    if (current_ms - last_secondary_change_ms >= RUNOUT_DEBOUNCE_MS) {
      secondary_runout_flag = triggered;
      last_secondary_change_ms = current_ms;
      SERIAL_ECHOPAIR("Secondary Sensor State: ", triggered ? "RUNOUT" : "PRESENT");
      SERIAL_EOL();
    }
  } else {
    last_secondary_change_ms = current_ms;
  }
  return secondary_runout_flag;
}
#endif

// Main runout check function called from Marlin's main loop
void filament_runout_check() {
  #ifndef DUAL_SENSOR_ENABLE
    bool runout_detected = read_primary_sensor();
  #else
    // Redundant check: both sensors must trigger to confirm runout (reduce false positives)
    bool primary = read_primary_sensor();
    bool secondary = read_secondary_sensor();
    bool runout_detected = primary && secondary;
    // Check for sensor mismatch (possible hardware failure)
    if (primary != secondary) {
      sensor_error_flag = true;
      log_sensor_error(primary, secondary);
    }
  #endif

  if (runout_detected && !sensor_error_flag) {
    total_runout_events++;
    SERIAL_ECHOLNPGM("Filament Runout Confirmed. Pausing print...");
    // Call Marlin's pause routine
    pause_print();
    // Log event to SD card
    log_runout_event();
  }

  if (sensor_error_flag) {
    SERIAL_ECHOLNPGM("Sensor Mismatch Error. Check hardware.");
    // Optional: Pause print on sensor error to prevent false runout
    // pause_print();
  }
}

// Log runout event to SD card
void log_runout_event() {
  if (card.isMounted()) {
    card.openFileWrite("runout_log.txt");
    char log_entry[128];
    snprintf(log_entry, sizeof(log_entry), "Runout Event %lu: Time %lu ms, Total Events %lu\n",
             total_runout_events, print_job_timer.duration(), total_runout_events);
    card.write(log_entry, strlen(log_entry));
    card.closeFile();
  }
}

// Log sensor mismatch error to SD card
void log_sensor_error(bool primary, bool secondary) {
  if (card.isMounted()) {
    card.openFileWrite(ERROR_LOG_FILE);
    char error_entry[128];
    snprintf(error_entry, sizeof(error_entry), "Sensor Mismatch: Primary %d, Secondary %d, Time %lu ms\n",
             primary, secondary, millis());
    card.write(error_entry, strlen(error_entry));
    card.closeFile();
  }
}

#!/usr/bin/env python3
"""
Filament Runout Sensor Validation Benchmark
Tests sensor accuracy, false positive rate, and response time over 1000 print cycles.
Requires: Python 3.10+, pyserial, pandas, matplotlib
"""

import serial
import time
import pandas as pd
import matplotlib.pyplot as plt
from dataclasses import dataclass
from typing import List, Optional
import logging
from pathlib import Path

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('sensor_benchmark.log'),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger(__name__)

@dataclass
class SensorReading:
    timestamp_ms: int
    filament_present: bool
    raw_voltage: float
    is_valid: bool
    error_msg: Optional[str] = None

class SensorValidator:
    def __init__(self, port: str = '/dev/ttyUSB0', baud_rate: int = 115200):
        self.port = port
        self.baud_rate = baud_rate
        self.ser: Optional[serial.Serial] = None
        self.readings: List[SensorReading] = []
        self.total_reads = 0
        self.false_positives = 0
        self.false_negatives = 0
        self.response_times_ms: List[int] = []

    def connect(self) -> bool:
        """Connect to ESP32 sensor serial port with retry logic."""
        max_retries = 3
        for attempt in range(max_retries):
            try:
                self.ser = serial.Serial(self.port, self.baud_rate, timeout=1)
                time.sleep(2)  # Wait for ESP32 reset
                logger.info(f"Connected to sensor on {self.port}")
                return True
            except serial.SerialException as e:
                logger.error(f"Connection attempt {attempt+1} failed: {e}")
                time.sleep(1)
        logger.error("Failed to connect to sensor after max retries")
        return False

    def read_sensor(self) -> Optional[SensorReading]:
        """Read a single sensor reading with error handling."""
        if not self.ser or not self.ser.is_open:
            raise ConnectionError("Serial port not open")
        try:
            line = self.ser.readline().decode('utf-8').strip()
            if not line:
                return None
            # Parse line format: "State: 1, Raw: 0" (1=present, 0=runout)
            parts = line.split(',')
            filament_present = parts[0].split(':')[1].strip() == '1'
            raw_voltage = float(parts[1].split(':')[1].strip()) * 3.3 / 1024  # 10-bit ADC
            timestamp = int(time.time() * 1000)
            self.total_reads += 1
            return SensorReading(
                timestamp_ms=timestamp,
                filament_present=filament_present,
                raw_voltage=raw_voltage,
                is_valid=True
            )
        except (IndexError, ValueError) as e:
            logger.warning(f"Invalid reading: {line}, Error: {e}")
            return SensorReading(
                timestamp_ms=int(time.time() * 1000),
                filament_present=False,
                raw_voltage=0.0,
                is_valid=False,
                error_msg=str(e)
            )
        except Exception as e:
            logger.error(f"Unexpected read error: {e}")
            return None

    def run_benchmark(self, duration_sec: int = 3600, simulate_runout_every: int = 300):
        """Run benchmark for specified duration, simulate runout every N seconds."""
        logger.info(f"Starting benchmark for {duration_sec} seconds")
        start_time = time.time()
        last_simulated_runout = start_time
        simulated_runout_active = False

        while (time.time() - start_time) < duration_sec:
            # Simulate filament runout every simulate_runout_every seconds
            if (time.time() - last_simulated_runout) > simulate_runout_every:
                simulated_runout_active = not simulated_runout_active
                logger.info(f"Simulated runout state: {simulated_runout_active}")
                last_simulated_runout = time.time()
                # Measure response time
                resp_start = time.time()
                time.sleep(0.1)  # Wait for sensor to react

            reading = self.read_sensor()
            if reading:
                # Check for false positives/negatives
                if simulated_runout_active and reading.filament_present:
                    self.false_negatives += 1
                    logger.warning("False negative detected: Simulated runout but sensor reports present")
                elif not simulated_runout_active and not reading.filament_present:
                    self.false_positives += 1
                    logger.warning("False positive detected: No runout simulated but sensor reports runout")
                self.readings.append(reading)
                # Calculate response time if state changed
                if reading.filament_present != simulated_runout_active:
                    resp_time = (time.time() - resp_start) * 1000
                    self.response_times_ms.append(int(resp_time))

            time.sleep(0.1)  # 10Hz sample rate

        logger.info("Benchmark complete")

    def generate_report(self):
        """Generate CSV report and matplotlib plots."""
        if not self.readings:
            logger.warning("No readings to generate report")
            return
        # Save to CSV
        df = pd.DataFrame([r.__dict__ for r in self.readings])
        report_path = Path('sensor_benchmark_report.csv')
        df.to_csv(report_path, index=False)
        logger.info(f"Report saved to {report_path}")

        # Calculate metrics
        false_positive_rate = (self.false_positives / self.total_reads) * 100 if self.total_reads > 0 else 0
        avg_response_ms = sum(self.response_times_ms) / len(self.response_times_ms) if self.response_times_ms else 0

        # Plot results
        plt.figure(figsize=(12, 6))
        plt.plot(df['timestamp_ms'], df['filament_present'], label='Filament Present')
        plt.xlabel('Timestamp (ms)')
        plt.ylabel('State (1=Present, 0=Runout)')
        plt.title('Sensor State Over Benchmark Duration')
        plt.legend()
        plot_path = Path('sensor_state_plot.png')
        plt.savefig(plot_path)
        logger.info(f"Plot saved to {plot_path}")

        # Print summary
        print("\n=== Benchmark Summary ===")
        print(f"Total Reads: {self.total_reads}")
        print(f"False Positive Rate: {false_positive_rate:.2f}%")
        print(f"False Negative Rate: { (self.false_negatives / self.total_reads) * 100:.2f}%")
        print(f"Average Response Time: {avg_response_ms:.2f} ms")
        print(f"Total Valid Readings: {len([r for r in self.readings if r.is_valid])}")

    def disconnect(self):
        if self.ser and self.ser.is_open:
            self.ser.close()
            logger.info("Disconnected from sensor")

if __name__ == "__main__":
    validator = SensorValidator(port='/dev/ttyUSB0', baud_rate=115200)
    if not validator.connect():
        exit(1)
    try:
        validator.run_benchmark(duration_sec=3600, simulate_runout_every=300)
        validator.generate_report()
    except KeyboardInterrupt:
        logger.info("Benchmark interrupted by user")
    except Exception as e:
        logger.error(f"Benchmark failed: {e}")
    finally:
        validator.disconnect()

import machine
import time

# Unconditioned sensor pin (GPIO18)
unconditioned = machine.Pin(18, machine.Pin.IN)
# Conditioned sensor pin (GPIO19, after Schmitt trigger)
conditioned = machine.Pin(19, machine.Pin.IN)

while True:
    print(f"Unconditioned: {unconditioned.value()}, Conditioned: {conditioned.value()}")
    time.sleep(0.1)

[filament_switch_sensor sensor_1]
pin: PA15
runout_distance: 0.5
pause_on_runout: False

[filament_switch_sensor sensor_2]
pin: PA14
runout_distance: 0.5
pause_on_runout: False

[gcode_macro CHECK_FILAMENT]
gcode:
  {% if printer["filament_switch_sensor sensor_1"].filament_detected == False and printer["filament_switch_sensor sensor_2"].filament_detected == False %}
    PAUSE
    M117 FILAMENT RUNOUT
  {% endif %}

import serial

ser = serial.Serial('/dev/ttyUSB0', 115200, timeout=1)
ser.write(b'M20\n')  # List SD card files
files = ser.readlines()
for f in files:
    if b'runout' in f:
        ser.write(f'M23 {f.strip()}\n'.encode())  # Select log file
        ser.write(b'M24\n')  # Start reading
        print(ser.readlines())

filament-runout-sensor/
├── firmware/
│   ├── marlin/
│   │   ├── Configuration.h
│   │   ├── Configuration_adv.h
│   │   └── filament_runout.cpp
│   ├── klipper/
│   │   └── printer.cfg
│   └── micropython/
│       └── sensor_reader.py
├── validation/
│   ├── benchmark.py
│   └── requirements.txt
├── hardware/
│   ├── schematic.pdf
│   └── bom.csv
└── README.md