94 lines
2.9 KiB
Python
94 lines
2.9 KiB
Python
"""
|
|
Example application entry point.
|
|
|
|
Replace this with the Magnifica logic. The launcher provides connectivity,
|
|
configuration, activity tracking and OTA in ctx.
|
|
"""
|
|
try:
|
|
import uasyncio as asyncio
|
|
except ImportError:
|
|
import asyncio
|
|
|
|
|
|
def as_bool(value, default=False):
|
|
if value is None:
|
|
return default
|
|
return str(value).strip().lower() in ("1", "true", "yes", "on", "enabled")
|
|
|
|
|
|
def as_int(value, default=0, minimum=None):
|
|
try:
|
|
result = int(str(value).strip())
|
|
except Exception:
|
|
result = default
|
|
if minimum is not None:
|
|
result = max(minimum, result)
|
|
return result
|
|
|
|
|
|
async def run_probe_mode(ctx):
|
|
"""Continuously collect read-only probe data while the app keeps running."""
|
|
env = ctx["env"]
|
|
pin_text = env.get("PROBE_GPIO", "").strip()
|
|
if not pin_text:
|
|
raise RuntimeError("PROBE_ENABLED=true requires PROBE_GPIO")
|
|
try:
|
|
pin_id = int(pin_text)
|
|
except Exception:
|
|
raise RuntimeError("PROBE_GPIO must be an integer GPIO number")
|
|
|
|
mode = env.get("PROBE_MODE", "frequency").strip().lower()
|
|
if mode not in ("state", "frequency", "pulse", "watch"):
|
|
raise RuntimeError("Unknown PROBE_MODE=" + mode)
|
|
|
|
from app.digital_probe import open_probe
|
|
probe = open_probe(pin_id)
|
|
frequency_seconds = as_int(env.get("PROBE_FREQUENCY_SECONDS"), 2, 1)
|
|
watch_seconds = as_int(env.get("PROBE_WATCH_SECONDS"), 5, 1)
|
|
watch_events = as_int(env.get("PROBE_WATCH_MAX_EVENTS"), 200, 1)
|
|
pulse_timeout_us = as_int(env.get("PROBE_PULSE_TIMEOUT_US"), 1_000_000, 1)
|
|
interval_seconds = as_int(env.get("PROBE_INTERVAL_SECONDS"), 3, 1)
|
|
activity = ctx["activity"]
|
|
|
|
print("[probe] mode=%s interval=%ss" % (mode, interval_seconds))
|
|
while True:
|
|
# Measurements are data collection, not a machine action, but marking
|
|
# activity prevents an OTA reboot from interrupting a capture.
|
|
activity.touch()
|
|
if mode == "state":
|
|
probe.state()
|
|
elif mode == "frequency":
|
|
probe.frequency(frequency_seconds)
|
|
elif mode == "pulse":
|
|
probe.pulse(pulse_timeout_us)
|
|
else:
|
|
probe.watch(watch_seconds, watch_events)
|
|
activity.touch()
|
|
await asyncio.sleep_ms(interval_seconds * 1000)
|
|
|
|
|
|
async def setup(ctx):
|
|
# Configure GPIOs, state machines, etc. here.
|
|
# If this raises immediately after an OTA, boot.py can roll back.
|
|
ctx["system"].ready()
|
|
|
|
|
|
async def main(ctx):
|
|
if as_bool(ctx["env"].get("PROBE_ENABLED"), False):
|
|
await run_probe_mode(ctx)
|
|
return
|
|
|
|
activity = ctx["activity"]
|
|
uart = ctx["uart"]
|
|
|
|
while True:
|
|
# Example: UART traffic counts as activity.
|
|
if uart is not None and uart.any():
|
|
activity.touch()
|
|
data = uart.read()
|
|
if data:
|
|
print("[app] UART:", data)
|
|
|
|
# This local sleep is app responsiveness only; it is NOT OTA polling.
|
|
await asyncio.sleep_ms(250)
|