# main.py
# === Test: MPX5050DP + ADS1115 + TCA9548A -> Stepper an/aus nach mbar ===
# ESP32-S3-DevKitC-1:
# STEP -> IO1 (GPIO1)
# EN -> IO2 (GPIO2) (active-low)
#
# Ablauf:
# - Sensor lesen (Median-of-5 + EMA)
# - p_mbar berechnen (LinReg aus u_list -> p_list_template)
# - Hysterese:
# p <= THRESH_ON -> Motor START
# p >= THRESH_OFF -> Motor STOP
from machine import Pin, I2C
from time import sleep_ms, ticks_ms, ticks_diff
from ads1115 import ADS1115
from stepper import StepperPump
# ================== Hardware Setup ==================
I2C_ID = 0
SDA = 8
SCL = 9
I2C_FREQ = 100_000
TCA_ADDR = 0x70 # TCA9548A
# ================== Stepper Pins (ESP32-S3-DevKitC-1) ==================
STEPPER_STEP_PIN = 1 # IO1/GPIO1
STEPPER_EN_PIN = 2 # IO2/GPIO2 (active-low)
# ================== ADS1115 / PGA ==================
PGA_CODE = 2 # ±2.048 V
DR = 32 # 32 SPS
# ================== Timing / Filter ==================
SAMPLE_PERIOD_MS = 1000
EMA_M = 10
ALPHA = 1.0 / EMA_M
STATUS_PERIOD_MS = 5000
SENSOR_TIMEOUT_MS = 8000
MEDIAN_N = 5
MEDIAN_READ_DELAY_MS = 35 # ~1/DR
# ================== Steuer-Grenzen ==================
THRESH_ON = -22.0
THRESH_OFF = -20.0
# Stepper Geschwindigkeit (Steps pro Sekunde)
STEP_SPS = 5000
# ================== Sensor Setup ==================
TCA_PORT = 0
ADS_ADDR = 0x48
ADS_CHANNEL = 0 # A0
# ================== Kalibrierung ==================
CM_TO_MBAR = -0.980665
u_list = [
0.173117, 0.258452, 0.345842, 0.429504, 0.515623,
0.599765, 0.687427, 0.773606, 0.857606, 0.944102,
1.031952, 1.116262, 1.200392
]
cm_list = [0, 10, 20, 30, 40, 50, 60, 70, 80, 90, 100, 110, 120]
p_list_template = [CM_TO_MBAR * cm for cm in cm_list]
# ================== I2C / TCA ==================
i2c = I2C(I2C_ID, sda=Pin(SDA), scl=Pin(SCL), freq=I2C_FREQ)
def tca_select(port: int):
"""Aktiviert genau einen Port (0..7) auf dem TCA9548A."""
if not (0 <= port <= 7):
raise ValueError("TCA port must be 0..7")
i2c.writeto(TCA_ADDR, bytes([1 << port]))
# ================== Mathe & Filter ==================
def linreg(u, p):
n = len(u)
su = sum(u)
sp = sum(p)
suu = sum(ui*ui for ui in u)
sup = sum(ui*pi for ui, pi in zip(u, p))
denom = n*suu - su*su
if denom == 0:
return 0.0, 0.0, 0.0
a = (n*sup - su*sp) / denom
b = (sp - a*su) / n
mean_p = sp / n
ss_tot = sum((pi-mean_p)**2 for pi in p)
ss_res = sum((pi-(a*ui + b))**2 for ui, pi in zip(u, p))
r2 = 1.0 - (ss_res / ss_tot if ss_tot != 0 else 0.0)
return a, b, r2
def median_of_n(read_fn, n=5, delay_ms=0):
vals = []
for _ in range(n):
vals.append(read_fn())
if delay_ms:
sleep_ms(delay_ms)
vals.sort()
return vals[n // 2]
def ema_update(prev, x, alpha):
return x if prev is None else (alpha * x + (1.0 - alpha) * prev)
# ================== ADS ==================
ads = ADS1115(i2c, addr=ADS_ADDR, pga=PGA_CODE, dr=DR)
def read_channel_median():
def _read_once():
tca_select(TCA_PORT)
return ads.read_voltage(ADS_CHANNEL)
return median_of_n(_read_once, n=MEDIAN_N, delay_ms=MEDIAN_READ_DELAY_MS)
# ================== Main ==================
def main():
# Kalibrierung bestimmen
m = min(len(u_list), len(p_list_template))
uu = u_list[:m]
pp = p_list_template[:m]
A, B, R = linreg(uu, pp)
print("=== Stepper-Test nach mbar ===")
print("TCA{} ADS{:02X} CH{}".format(TCA_PORT, ADS_ADDR, ADS_CHANNEL))
print("Kalibrierung: A={:.6f} mbar/V | B={:.6f} mbar | R²={:.6f}".format(A, B, R))
print("Grenzen: ON<={:.1f} OFF>={:.1f} | STEP_SPS={}".format(THRESH_ON, THRESH_OFF, STEP_SPS))
print("Pins: STEP=IO{} EN=IO{} (active-low)".format(STEPPER_STEP_PIN, STEPPER_EN_PIN))
print()
# Stepper-Objekt
motor = StepperPump(step_pin=STEPPER_STEP_PIN, en_pin=STEPPER_EN_PIN)
motor.stop() # sicher aus
last_status_ms = ticks_ms()
last_u_ema = None
last_p = None
last_ts = 0
sensor_ok = False
try:
while True:
now = ticks_ms()
# --- Sensor lesen ---
try:
u_raw = read_channel_median()
u_ema = ema_update(last_u_ema, u_raw, ALPHA)
p_mbar = A * u_ema + B
last_u_ema = u_ema
last_p = p_mbar
last_ts = now
sensor_ok = True
except OSError:
sensor_ok = False
fresh = sensor_ok and (ticks_diff(now, last_ts) <= SENSOR_TIMEOUT_MS)
# --- Motorsteuerung ---
if not fresh:
# Sensor weg -> Motor aus Sicherheitsgründen aus
motor.stop()
else:
# Hysterese
if (not motor.is_enabled()) and (last_p <= THRESH_ON):
motor.start(STEP_SPS)
elif motor.is_enabled() and (last_p >= THRESH_OFF):
motor.stop()
# --- Status ---
if ticks_diff(now, last_status_ms) >= STATUS_PERIOD_MS:
last_status_ms = now
if not fresh:
print("[STATUS] SENSOR_FAIL | Motor={}".format("ON" if motor.is_enabled() else "OFF"))
else:
print("[STATUS] U_ema={:.6f} V | p={:+7.2f} mbar | Motor={}".format(
last_u_ema, last_p, "ON" if motor.is_enabled() else "OFF"
))
sleep_ms(SAMPLE_PERIOD_MS)
except KeyboardInterrupt:
print("\n[INFO] KeyboardInterrupt -> Motor aus")
except Exception as e:
print("\n[ERROR]", e)
finally:
motor.stop()
print("[SAFE] Motor AUS.")
main()
. Aber wenn es tatsächlich komplett gekapselt, und keines der oftmals hier gezeigten OpenFrame Netzteile ( die dann auch einfach auf dem Tisch liegen 
) ist, ok.