frarol96/Magnifica_S-Pro
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add ESP32-C3 launcher and digital probe

Browse Source
This commit is contained in:
Franz Rolfsvaag 2026-08-28 10:50:16 +02:00
commit 935395797a
13 changed files with 1606 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

108
.env.example Normal file
View File

@ -0,0 +1,108 @@
# ================================================================
# ESP32-C3 universal launcher configuration
# Copy to /.env on the ESP32. NEVER commit the real secrets file.
# ================================================================
PRINT_CONFIG_ON_BOOT=false
# ---------- Wi-Fi ----------
WIFI_ENABLED=true
WIFI_SSID=YourSSID
WIFI_PASSWORD=YourWiFiPassword
WIFI_HOSTNAME=magnifica-s-pro
WIFI_CONNECT_ON_BOOT=true
WIFI_REQUIRED=true
WIFI_TIMEOUT_SECONDS=20
WIFI_POWER_SAVE=true
# false = Wi-Fi is switched off after updater network tasks when the app
# itself does not need permanent Wi-Fi.
WIFI_KEEP_CONNECTED=true
NTP_SYNC_ENABLED=true
NTP_HOST=pool.ntp.org
# ---------- Bluetooth ----------
# ESP32-C3 supports BLE only, not Bluetooth Classic.
BLUETOOTH_ENABLED=false
BLUETOOTH_NAME=Coffee
BLUETOOTH_ADVERTISE=true
BLUETOOTH_ADV_INTERVAL_MS=2500
# ---------- UART ----------
UART_ENABLED=false
UART_ID=1
UART_BAUD=115200
UART_BITS=8
UART_PARITY=none
UART_STOP_BITS=1
UART_TX_GPIO=20
UART_RX_GPIO=21
# ---------- Self-update ----------
UPDATE_ENABLED=true
# Remote manifest requests are intentionally infrequent.
UPDATE_CHECK_INTERVAL_HOURS=24
# A due update will not start until the application has been idle this long.
UPDATE_MIN_IDLE_SECONDS=900
# If due but busy, this is only a LOCAL sleep before reconsidering idleness.
# It does NOT contact the repository every 15 minutes unless the device is idle.
UPDATE_BUSY_RETRY_SECONDS=900
# Failed network/repository attempts wait before another remote attempt.
UPDATE_FAILURE_RETRY_SECONDS=3600
# Boot checks still obey the boot delay and idle requirement.
UPDATE_CHECK_ON_BOOT=true
UPDATE_BOOT_DELAY_SECONDS=180
UPDATE_HTTP_TIMEOUT_SECONDS=20
UPDATE_MAX_FILE_BYTES=262144
UPDATE_ALLOW_HTTP=false
# ---------- Repository ----------
# auto | gitea | forgejo | github | gitlab | bitbucket | generic
UPDATE_PROVIDER=gitea
UPDATE_REPO_URL=https://git.rolfsvaag.no/frarol96/Magnifica_S-Pro
UPDATE_BRANCH=main
UPDATE_MANIFEST_PATH=firmware/manifest.json
# Optional overrides for unusual/self-hosted raw URL layouts:
# UPDATE_MANIFEST_URL=https://host/specific/path/manifest.json
# UPDATE_RAW_BASE_URL=https://host/specific/raw/base/
# ---------- Authentication ----------
# none/public | auto | basic | bearer | token | header
#
# Private Gitea/Forgejo:
# UPDATE_AUTH_MODE=auto
# UPDATE_TOKEN=...
#
# Private GitLab:
# UPDATE_AUTH_MODE=auto
# UPDATE_TOKEN=...
#
# Private GitHub:
# UPDATE_AUTH_MODE=auto
# UPDATE_TOKEN=...
#
# Bitbucket app password / generic Basic auth:
# UPDATE_AUTH_MODE=basic
# UPDATE_USERNAME=...
# UPDATE_PASSWORD=...
#
# Custom auth header:
# UPDATE_AUTH_MODE=header
# UPDATE_AUTH_HEADER=X-Api-Key
# UPDATE_AUTH_PREFIX=
# UPDATE_TOKEN=...
UPDATE_AUTH_MODE=none
UPDATE_TOKEN=
UPDATE_USERNAME=
UPDATE_PASSWORD=
UPDATE_AUTH_HEADER=Authorization
UPDATE_AUTH_PREFIX=

8
.gitignore vendored Normal file
View File

@ -0,0 +1,8 @@
.env
*.env
!.env.example
__pycache__/
*.pyc
.DS_Store
# Local Codex project log; intentionally never version-controlled.
AGENTS.md

115
README.md Normal file
View File

@ -0,0 +1,115 @@
# ESP32-C3 Universal MicroPython Launcher
A fault-tolerant launcher intended for the Magnifica ESP32-C3 project, but kept generic enough to reuse elsewhere.
## Core behavior
- `/.env` on the device holds all secrets and per-device options.
- Wi-Fi, BLE and UART are independent startup options.
- One Git repository can hold both the launcher and main application.
- The ESP32 does **not** need Git. It downloads a tiny JSON manifest and only changed raw files.
- `/launcher.py`, `/app/main.py`, and other manifest-listed files can update in one staged transaction.
- SHA-256 is verified before any file is committed.
- `.bak` rollback files plus a stable `boot.py` protect against interrupted or broken OTA.
- Secrets are never updated and are never in the manifest.
## Energy / update behavior
There is no 5-second repository polling loop.
Defaults are:
- remote manifest check: **once every 24 hours**
- boot delay before a possible check: **3 minutes**
- minimum inactivity before OTA may begin: **5 minutes**
- if busy when due: reconsider locally after **15 minutes**
- network failure: wait **1 hour** before another remote attempt
- BLE disabled by default
- Wi-Fi power saving enabled best-effort
If Wi-Fi exists only for OTA, set `WIFI_KEEP_CONNECTED=false`. If the app itself needs Wi-Fi for remote control, leave it `true`.
## Repository layout
```text
repo/
├─ README.md
├─ .gitignore
├─ .env.example
├─ firmware/
│ ├─ boot.py # stable recovery stub; install manually
│ ├─ launcher.py # OTA-updatable launcher
│ ├─ manifest.json # generated SHA-256 manifest
│ └─ app/
│ ├─ __init__.py
│ └─ main.py # actual Magnifica logic
├─ tools/
│ └─ build_manifest.py
└─ docs/
├─ DIGITAL_PROBE.md
├─ SETUP.md
└─ OTA_PROTOCOL.md
```
On the ESP32:
```text
/
├─ boot.py
├─ launcher.py
├─ .env
├─ update_state.json # generated automatically
└─ app/
├─ __init__.py
└─ main.py
```
## Safe inactivity-aware updates
The application receives `ctx["activity"]`.
Call this whenever a real interaction or machine-state change occurs:
```python
ctx["activity"].touch()
```
Bracket anything that must never be interrupted by OTA:
```python
ctx["activity"].begin()
try:
await brew_or_move_mechanism()
finally:
ctx["activity"].end()
```
OTA can start only after the long check interval is due **and** the device has been idle for `UPDATE_MIN_IDLE_SECONDS` **and** no busy lock is active.
## Gitea example
```env
UPDATE_PROVIDER=gitea
UPDATE_REPO_URL=https://git.example.com/user/repo
UPDATE_BRANCH=main
UPDATE_AUTH_MODE=auto
UPDATE_TOKEN=your_token
```
A public Gitea repo uses:
```env
UPDATE_AUTH_MODE=none
```
## Other platforms
Built-in URL/auth handling exists for Gitea/Forgejo/Codeberg, GitLab, GitHub and Bitbucket. For unusual self-hosted layouts, use `UPDATE_MANIFEST_URL` or `UPDATE_RAW_BASE_URL` and optionally a custom auth header.
See `docs/SETUP.md` and `docs/OTA_PROTOCOL.md`.
## Development probe
The first Magnifica S development tool is an input-only digital probe for
front-panel signal discovery. See `docs/DIGITAL_PROBE.md`; do not connect an
unknown machine wire directly to an ESP32 GPIO.

50
docs/DIGITAL_PROBE.md Normal file
View File

@ -0,0 +1,50 @@
# Read-only digital probe
`app.digital_probe` is the first Magnifica S development tool. It measures a
single already-conditioned digital signal; it does not control the machine and
never configures the ESP32-C3 pin as an output.
## Electrical boundary — required before connection
Do **not** attach an unknown wire from the coffee machine directly to an
ESP32-C3 GPIO. First identify the wire with the machine unplugged, then verify
its voltage, polarity, ground relationship, and behaviour at power-on with
appropriate test equipment. Connect it only through an interface that ensures
the ESP32 input stays between GND and 3.3 V at all times, including transients.
Use a board GPIO that is actually broken out and not occupied by the board.
Avoid GPIO2, GPIO8, and GPIO9 (ESP32-C3 boot strapping), GPIO12–GPIO17
(normally flash), and GPIO18–GPIO19 when USB-JTAG is needed. GPIO3 is a
candidate on boards that expose it, but confirm the board schematic first.
Keep the probe ground disconnected until it is known to be safe to share a
ground. If the machine signal is not confirmed low-voltage and ground-referred,
use a suitably rated isolated/conditioned interface instead of a direct link.
## Install and use
Deploy the whole `firmware/app/` directory. The probe is included in the OTA
manifest, but it is not auto-started, so normal launcher behaviour is unchanged.
At the MicroPython REPL:
```python
from app.digital_probe import open_probe
probe = open_probe(3) # Replace 3 after confirming your board pin and wiring.
probe.state()
probe.frequency(seconds=2)
probe.pulse(timeout_us=1_000_000)
probe.watch(seconds=5, max_events=200)
```
`frequency()` counts rising edges in its measurement window. It is useful for
signal discovery and approximate rates, but very fast or noisy signals can
overflow or exceed MicroPython IRQ handling capacity. `pulse()` reports one
high and one low width; negative results are timeouts (`-2` means the requested
level did not arrive, `-1` means its pulse did not finish). `watch()` is a
short, polling-based transition trace and can miss narrow pulses.
For every capture, note the machine state, wire identifier, wiring/interface,
measurement settings, and output. That context is necessary before mapping a
panel signal to a machine event.

67
docs/OTA_PROTOCOL.md Normal file
View File

@ -0,0 +1,67 @@
# OTA protocol
The updater uses a JSON manifest rather than Git operations on-device.
## Manifest schema 1
```json
{
"schema": 1,
"release": "1.2.3",
"files": [
{
"source": "firmware/launcher.py",
"target": "/launcher.py",
"sha256": "..."
},
{
"source": "firmware/app/main.py",
"target": "/app/main.py",
"sha256": "..."
}
]
}
```
A file is downloaded only when its local SHA-256 differs.
## Transaction
1. Wait for the long update interval to become due.
2. Wait for configured inactivity and no busy lock.
3. Connect Wi-Fi if needed.
4. Fetch the manifest.
5. Hash local target files.
6. Download only changed files into `/.update`.
7. Verify every SHA-256.
8. Create `/update_pending.json` with `commit_complete=false`.
9. Rename current files to `.bak`.
10. Rename staged files into their final locations.
11. Mark `commit_complete=true`.
12. Reboot.
13. New launcher/app performs setup.
14. App calls `ctx["system"].ready()`.
15. Backups and pending marker are removed.
A power loss during commit leaves `commit_complete=false`; stable `boot.py` restores the backups before normal startup.
If a completed update cannot import/start, `boot.py` restores the backups.
## Why boot.py is intentionally stable
The recovery component should not replace itself in the same transaction it is responsible for recovering. Update `/boot.py` manually and rarely.
## Authentication
Credentials exist only in `/.env`.
Supported modes:
- `none` / `public`
- `auto`
- `basic`
- `bearer`
- `token`
- `header` (arbitrary header + optional prefix)
HTTPS is required unless `UPDATE_ALLOW_HTTP=true` is explicitly set.

96
docs/SETUP.md Normal file
View File

@ -0,0 +1,96 @@
# Setup
## 1. Flash MicroPython
Use a current ESP32-C3 MicroPython build.
## 2. Initial device files
Copy:
```text
firmware/boot.py -> /boot.py
firmware/launcher.py -> /launcher.py
firmware/app/ -> /app/
.env.example -> /.env
```
Edit `/.env` on the ESP32 with real Wi-Fi/repository credentials.
## 3. Build the manifest before pushing a release
On your development machine:
```bash
python tools/build_manifest.py 1.0.0
```
Commit/push the resulting `firmware/manifest.json` together with the code.
The ESP32 compares SHA-256 hashes. A changed hash, not just the release label, determines whether a file updates.
## 4. Application interface
Recommended application shape:
```python
async def setup(ctx):
# initialize GPIOs and internal state
ctx["system"].ready()
async def main(ctx):
...
```
`ready()` is significant after OTA: it says the new launcher/app got through setup successfully, so rollback backups may be deleted.
### Context entries
- `ctx["env"]` — parsed local `.env`
- `ctx["wifi"]` — WiFiManager
- `ctx["ble"]` — BLE object or `None`
- `ctx["uart"]` — UART object or `None`
- `ctx["activity"]` — idle/busy tracker
- `ctx["updater"]` — UpdateManager
- `ctx["system"]` — ready/reboot helpers
## 5. Activity tracking
On user/device activity:
```python
ctx["activity"].touch()
```
For a protected operation:
```python
ctx["activity"].begin()
try:
await operation()
finally:
ctx["activity"].end()
```
A due update is postponed while the machine is active.
## 6. Energy-conscious configurations
OTA-only Wi-Fi:
```env
WIFI_ENABLED=true
WIFI_KEEP_CONNECTED=false
BLUETOOTH_ENABLED=false
UPDATE_CHECK_INTERVAL_HOURS=24
```
Always-online remote-control Wi-Fi:
```env
WIFI_ENABLED=true
WIFI_KEEP_CONNECTED=true
WIFI_POWER_SAVE=true
```
The ESP32-C3 supports BLE, not Bluetooth Classic.

1
firmware/app/__init__.py Normal file
View File

@ -0,0 +1 @@
"""Application package."""

125
firmware/app/digital_probe.py Normal file
View File

@ -0,0 +1,125 @@
"""Read-only digital logic probe for a conditioned ESP32-C3 GPIO input.
The ESP32 pin is always configured as an input with no internal pull. This is
for observation only: it must receive a signal already conditioned to the
ESP32-C3 input range. It cannot make an unknown coffee-machine wire safe.
"""
from machine import Pin, time_pulse_us
import time
class DigitalProbe:
"""Measure level, rising-edge frequency, pulse widths, and transitions."""
def __init__(self, pin_id):
self.pin_id = int(pin_id)
# Do not enable a pull resistor: the observed circuit must determine
# the level, and the probe must not intentionally bias it.
self.signal = Pin(self.pin_id, Pin.IN)
self._edge_count = 0
def _on_rising_edge(self, _pin):
# Keep IRQ work deliberately minimal. Do not print or allocate here.
self._edge_count += 1
def state(self):
"""Print and return the present digital level (0 or 1)."""
value = self.signal.value()
print("GPIO%d = %d" % (self.pin_id, value))
return value
def frequency(self, seconds=2):
"""Count rising edges during a fixed window and return approximate Hz."""
duration_ms = int(float(seconds) * 1000)
if duration_ms <= 0:
raise ValueError("seconds must be greater than zero")
self._edge_count = 0
start = time.ticks_ms()
self.signal.irq(trigger=Pin.IRQ_RISING, handler=self._on_rising_edge)
try:
time.sleep_ms(duration_ms)
finally:
self.signal.irq(handler=None)
elapsed_ms = max(1, time.ticks_diff(time.ticks_ms(), start))
count = self._edge_count
hz = (count * 1000.0) / elapsed_ms
print("Freq: %.2f Hz (%d rising edges / %.3fs)" % (
hz, count, elapsed_ms / 1000.0))
return hz
def pulse(self, timeout_us=1_000_000):
"""Measure one HIGH then one LOW pulse; return their timing data.
``time_pulse_us`` returns -2 when a requested level never arrives and
-1 when that level does not end before the timeout.
"""
timeout_us = int(timeout_us)
if timeout_us <= 0:
raise ValueError("timeout_us must be greater than zero")
high_us = time_pulse_us(self.signal, 1, timeout_us)
low_us = time_pulse_us(self.signal, 0, timeout_us)
if high_us < 0 or low_us < 0:
print("No complete pulse: HIGH=%d LOW=%d" % (high_us, low_us))
print("-2 = level did not arrive; -1 = pulse did not end in time")
return None
period_us = high_us + low_us
if period_us <= 0:
print("No complete pulse")
return None
hz = 1_000_000.0 / period_us
duty = (high_us * 100.0) / period_us
result = {
"high_us": high_us,
"low_us": low_us,
"period_us": period_us,
"frequency_hz": hz,
"duty_percent": duty,
}
print("HIGH: %dus LOW: %dus" % (high_us, low_us))
print("Freq: %.2f Hz Duty: %.1f%%" % (hz, duty))
return result
def watch(self, seconds=5, max_events=200):
"""Poll and print transitions with microsecond timestamps.
This is intended for short diagnostic captures. Narrow pulses can be
missed because output and polling run in Python; use ``frequency`` or
dedicated capture hardware for high-rate or timing-critical signals.
"""
duration_ms = int(float(seconds) * 1000)
max_events = int(max_events)
if duration_ms <= 0:
raise ValueError("seconds must be greater than zero")
if max_events <= 0:
raise ValueError("max_events must be greater than zero")
last = self.signal.value()
start_us = time.ticks_us()
deadline_ms = time.ticks_add(time.ticks_ms(), duration_ms)
events = 0
while time.ticks_diff(deadline_ms, time.ticks_ms()) > 0:
value = self.signal.value()
if value == last:
continue
now_us = time.ticks_us()
print("%9dus %d" % (time.ticks_diff(now_us, start_us), value))
last = value
events += 1
if events >= max_events:
print("Capture stopped after %d events" % max_events)
break
return events
def open_probe(pin_id):
"""Create a probe and print the interactive REPL commands."""
probe = DigitalProbe(pin_id)
print("ESP32-C3 read-only logic probe on GPIO%d" % probe.pin_id)
print("Use: probe.state(), probe.frequency(), probe.pulse(), probe.watch()")
return probe

32
firmware/app/main.py Normal file
View File

@ -0,0 +1,32 @@
"""
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
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):
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)

92
firmware/boot.py Normal file
View File

@ -0,0 +1,92 @@
"""
Stable recovery boot stub.
Install once as /boot.py. It is deliberately excluded from normal OTA updates.
If a staged launcher/application update is interrupted or cannot boot, this
restores the previous .bak files.
"""
try:
import ujson as json
except ImportError:
import json
try:
import uos as os
except ImportError:
import os
PENDING_FILE = "/update_pending.json"
def exists(path):
try:
os.stat(path)
return True
except OSError:
return False
def remove(path):
try:
os.remove(path)
except OSError:
pass
def read_pending():
if not exists(PENDING_FILE):
return None
try:
with open(PENDING_FILE, "r") as f:
return json.loads(f.read())
except Exception:
return None
def rollback(pending):
if not pending:
return False
restored = False
for item in reversed(pending.get("files", [])):
target = item.get("target")
backup = item.get("backup")
if not target or not backup:
continue
try:
if exists(backup):
remove(target)
os.rename(backup, target)
restored = True
except Exception as exc:
print("[boot] rollback failed:", target, exc)
remove(PENDING_FILE)
return restored
pending = read_pending()
# Never boot a half-committed file set after a power loss.
if pending and not pending.get("commit_complete", False):
print("[boot] interrupted OTA commit; rolling back")
rollback(pending)
try:
import machine
machine.reset()
except Exception:
raise SystemExit
try:
import launcher
launcher.run()
except Exception as exc:
print("[boot] launcher/application startup failed:", exc)
pending = read_pending()
if pending:
print("[boot] reverting pending OTA")
rollback(pending)
try:
import machine
machine.reset()
except Exception:
pass
raise

841
firmware/launcher.py Normal file
View File

@ -0,0 +1,841 @@
"""
Universal ESP32-C3 MicroPython launcher.
Features
--------
- Loads local /.env secrets/config.
- Independently enables/configures Wi-Fi, BLE and UART.
- Runs /app/main.py.
- Optional manifest-based self-update from Gitea/Forgejo/Codeberg,
GitHub, GitLab, Bitbucket, or a generic raw URL.
- Public, Basic, Bearer/token and custom-header authentication.
- Updates only begin after configured inactivity.
- Remote checks are infrequent (default: every 24 hours).
- Staged downloads, SHA-256 verification, backups and boot rollback.
This intentionally uses raw HTTP(S) files + a manifest rather than running Git
on the ESP32. MicroPython only downloads files that actually changed.
"""
try:
import ujson as json
except ImportError:
import json
try:
import uos as os
except ImportError:
import os
try:
import ubinascii as binascii
except ImportError:
import binascii
try:
import uhashlib as hashlib
except ImportError:
import hashlib
import time
try:
import uasyncio as asyncio
except ImportError:
import asyncio
ENV_FILE = "/.env"
PENDING_FILE = "/update_pending.json"
STATE_FILE = "/update_state.json"
STAGE_DIR = "/.update"
DEFAULT_MANIFEST_PATH = "firmware/manifest.json"
_GLOBAL_ENV = {}
def log(*parts):
print("[launcher]", *parts)
def exists(path):
try:
os.stat(path)
return True
except OSError:
return False
def remove(path):
try:
os.remove(path)
except OSError:
pass
def dirname(path):
if "/" not in path:
return ""
out = path.rsplit("/", 1)[0]
return out or "/"
def mkdir_p(path):
if not path or path == "/":
return
current = ""
for piece in path.split("/"):
if not piece:
continue
current += "/" + piece
if not exists(current):
try:
os.mkdir(current)
except OSError:
pass
def remove_tree_best_effort(path):
if not exists(path):
return
try:
for name in os.listdir(path):
child = path + "/" + name
try:
mode = os.stat(child)[0]
if mode & 0x4000:
remove_tree_best_effort(child)
else:
remove(child)
except Exception:
remove(child)
os.rmdir(path)
except Exception:
pass
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):
try:
return int(str(value).strip())
except Exception:
return default
def strip_quotes(value):
value = value.strip()
if len(value) >= 2 and value[0] == value[-1] and value[0] in ("'", '"'):
return value[1:-1]
return value
def load_env(path=ENV_FILE):
env = {}
if not exists(path):
log(path, "not found; built-in defaults will be used")
return env
with open(path, "r") as f:
for raw in f:
line = raw.strip()
if not line or line.startswith("#"):
continue
if line.startswith("export "):
line = line[7:].strip()
if "=" not in line:
continue
key, value = line.split("=", 1)
key = key.strip()
value = value.strip()
if value and value[0] not in ("'", '"') and " #" in value:
value = value.split(" #", 1)[0].rstrip()
env[key] = strip_quotes(value)
return env
def safe_config_log(env):
secret_words = ("PASSWORD", "TOKEN", "SECRET", "KEY")
log("configuration:")
for key in sorted(env):
if any(word in key.upper() for word in secret_words):
value = "***" if env.get(key) else ""
else:
value = env.get(key)
print(" %s=%s" % (key, value))
def sha256_file(path):
h = hashlib.sha256()
with open(path, "rb") as f:
while True:
chunk = f.read(1024)
if not chunk:
break
h.update(chunk)
return binascii.hexlify(h.digest()).decode().lower()
def write_json_atomic(path, obj):
tmp = path + ".tmp"
with open(tmp, "w") as f:
f.write(json.dumps(obj))
remove(path)
os.rename(tmp, path)
def read_json(path, default=None):
try:
with open(path, "r") as f:
return json.loads(f.read())
except Exception:
return default
def epoch_is_valid():
try:
return time.time() > 1700000000
except Exception:
return False
class ActivityTracker:
"""Tracks idleness and protected/busy operations for safe OTA timing."""
def __init__(self):
self._last_ms = time.ticks_ms()
self._busy_depth = 0
def touch(self):
self._last_ms = time.ticks_ms()
def begin(self):
self._busy_depth += 1
self.touch()
def end(self):
if self._busy_depth > 0:
self._busy_depth -= 1
self.touch()
def busy(self):
return self._busy_depth > 0
def idle_ms(self):
return max(0, time.ticks_diff(time.ticks_ms(), self._last_ms))
def idle_for(self, seconds):
return (not self.busy()) and self.idle_ms() >= int(seconds * 1000)
class WiFiManager:
def __init__(self, env):
self.env = env
self.wlan = None
def enabled(self):
return as_bool(self.env.get("WIFI_ENABLED"), False)
def connected(self):
return bool(self.wlan and self.wlan.isconnected())
def _ensure_interface(self):
if self.wlan:
return self.wlan
import network
self.wlan = network.WLAN(network.STA_IF)
self.wlan.active(True)
# Best effort: constants differ slightly between MicroPython builds.
if as_bool(self.env.get("WIFI_POWER_SAVE"), True):
try:
pm = getattr(network, "PM_POWERSAVE", None)
if pm is not None:
self.wlan.config(pm=pm)
except Exception:
pass
return self.wlan
def connect(self):
if not self.enabled():
return None
ssid = self.env.get("WIFI_SSID", "")
password = self.env.get("WIFI_PASSWORD", "")
if not ssid:
raise RuntimeError("WIFI_ENABLED=true but WIFI_SSID is empty")
wlan = self._ensure_interface()
if wlan.isconnected():
return wlan
hostname = self.env.get("WIFI_HOSTNAME", "")
if hostname:
try:
wlan.config(hostname=hostname)
except Exception:
pass
timeout_s = as_int(self.env.get("WIFI_TIMEOUT_SECONDS"), 20)
log("connecting Wi-Fi:", ssid)
wlan.connect(ssid, password)
deadline = time.ticks_add(time.ticks_ms(), timeout_s * 1000)
while not wlan.isconnected():
if time.ticks_diff(deadline, time.ticks_ms()) <= 0:
raise RuntimeError("Wi-Fi connection timeout")
time.sleep_ms(250)
log("Wi-Fi connected:", wlan.ifconfig()[0])
self._sync_time_best_effort()
return wlan
def _sync_time_best_effort(self):
if not as_bool(self.env.get("NTP_SYNC_ENABLED"), True):
return
try:
import ntptime
ntptime.host = self.env.get("NTP_HOST", "pool.ntp.org")
ntptime.settime()
except Exception as exc:
log("NTP sync skipped/failed:", exc)
def disconnect_if_allowed(self):
if not self.wlan or as_bool(self.env.get("WIFI_KEEP_CONNECTED"), True):
return
try:
self.wlan.disconnect()
self.wlan.active(False)
log("Wi-Fi powered down after network task")
except Exception:
pass
def _ble_adv_payload(name):
payload = bytearray(b"\x02\x01\x06")
name_b = name.encode()
if len(name_b) > 26:
name_b = name_b[:26]
payload.extend(bytes((len(name_b) + 1, 0x09)))
payload.extend(name_b)
return payload
def setup_ble(env):
# ESP32-C3 supports BLE, not Bluetooth Classic.
if not as_bool(env.get("BLUETOOTH_ENABLED"), False):
log("BLE disabled")
return None
try:
import bluetooth
except ImportError:
import ubluetooth as bluetooth
ble = bluetooth.BLE()
ble.active(True)
name = env.get("BLUETOOTH_NAME", "ESP32-C3")
try:
ble.config(gap_name=name)
except Exception:
pass
if as_bool(env.get("BLUETOOTH_ADVERTISE"), True):
interval_ms = as_int(env.get("BLUETOOTH_ADV_INTERVAL_MS"), 1000)
try:
ble.gap_advertise(interval_ms * 1000, _ble_adv_payload(name))
log("BLE advertising as", name)
except Exception as exc:
log("BLE active; advertising failed:", exc)
return ble
def setup_uart(env):
if not as_bool(env.get("UART_ENABLED"), False):
log("UART disabled")
return None
from machine import UART, Pin
tx_raw = env.get("UART_TX_GPIO", "")
rx_raw = env.get("UART_RX_GPIO", "")
if tx_raw == "" or rx_raw == "":
raise RuntimeError("UART_ENABLED=true requires UART_TX_GPIO and UART_RX_GPIO")
parity_text = env.get("UART_PARITY", "none").strip().lower()
parity = None
if parity_text in ("even", "0"):
parity = 0
elif parity_text in ("odd", "1"):
parity = 1
uart_id = as_int(env.get("UART_ID"), 1)
baud = as_int(env.get("UART_BAUD"), 115200)
uart = UART(
uart_id,
baudrate=baud,
bits=as_int(env.get("UART_BITS"), 8),
parity=parity,
stop=as_int(env.get("UART_STOP_BITS"), 1),
tx=Pin(as_int(tx_raw)),
rx=Pin(as_int(rx_raw)),
)
log("UART%d: %d baud TX=GPIO%s RX=GPIO%s" % (uart_id, baud, tx_raw, rx_raw))
return uart
def detect_provider(repo_url, configured="auto"):
provider = (configured or "auto").strip().lower()
if provider != "auto":
return provider
value = (repo_url or "").lower()
if "github.com" in value:
return "github"
if "gitlab.com" in value:
return "gitlab"
if "bitbucket.org" in value:
return "bitbucket"
if "codeberg.org" in value:
return "forgejo"
return "gitea"
def clean_repo_url(url):
value = (url or "").strip().rstrip("/")
if value.endswith(".git"):
value = value[:-4]
return value
def github_repo_parts(repo):
marker = "github.com/"
if marker not in repo:
raise ValueError("Invalid GitHub repository URL")
rest = repo.split(marker, 1)[1].strip("/")
parts = rest.split("/")
if len(parts) < 2:
raise ValueError("Invalid GitHub repository URL")
return parts[0], parts[1]
def raw_url_for(env, path):
explicit_base = env.get("UPDATE_RAW_BASE_URL", "").strip()
if explicit_base:
return explicit_base.rstrip("/") + "/" + path.lstrip("/")
repo = clean_repo_url(env.get("UPDATE_REPO_URL", ""))
if not repo:
raise ValueError("Set UPDATE_REPO_URL or UPDATE_RAW_BASE_URL")
branch = env.get("UPDATE_BRANCH", "main")
provider = detect_provider(repo, env.get("UPDATE_PROVIDER", "auto"))
if provider == "github":
owner, name = github_repo_parts(repo)
return "https://api.github.com/repos/%s/%s/contents/%s?ref=%s" % (
owner, name, path.lstrip("/"), branch
)
if provider == "gitlab":
return "%s/-/raw/%s/%s" % (repo, branch, path.lstrip("/"))
if provider == "bitbucket":
return "%s/raw/%s/%s" % (repo, branch, path.lstrip("/"))
return "%s/raw/branch/%s/%s" % (repo, branch, path.lstrip("/"))
def _basic_auth(username, password):
raw = ("%s:%s" % (username, password)).encode()
return binascii.b2a_base64(raw).decode().strip()
def auth_headers(env):
provider = detect_provider(env.get("UPDATE_REPO_URL", ""), env.get("UPDATE_PROVIDER", "auto"))
mode = env.get("UPDATE_AUTH_MODE", "none").strip().lower()
headers = {}
# GitHub Contents API raw response support.
if provider == "github":
headers["Accept"] = "application/vnd.github.raw+json"
if mode in ("", "none", "public"):
return headers
username = env.get("UPDATE_USERNAME", "")
password = env.get("UPDATE_PASSWORD", "")
token = env.get("UPDATE_TOKEN", "")
if mode == "auto":
if username or password:
headers["Authorization"] = "Basic " + _basic_auth(username, password)
elif token:
if provider in ("gitea", "forgejo"):
headers["Authorization"] = "token " + token
elif provider == "gitlab":
headers["PRIVATE-TOKEN"] = token
else:
headers["Authorization"] = "Bearer " + token
return headers
if mode == "basic":
headers["Authorization"] = "Basic " + _basic_auth(username, password)
elif mode == "bearer":
headers["Authorization"] = "Bearer " + token
elif mode == "token":
header = env.get("UPDATE_AUTH_HEADER", "Authorization")
prefix = env.get("UPDATE_AUTH_PREFIX", "")
if not prefix:
prefix = "token" if provider in ("gitea", "forgejo") else "Bearer"
headers[header] = (prefix + " " + token).strip()
elif mode == "header":
header = env.get("UPDATE_AUTH_HEADER", "Authorization")
prefix = env.get("UPDATE_AUTH_PREFIX", "")
headers[header] = ((prefix + " ") if prefix else "") + token
else:
raise ValueError("Unknown UPDATE_AUTH_MODE=" + mode)
return headers
def http_get(url, headers, timeout_s):
if url.startswith("http://") and not as_bool(_GLOBAL_ENV.get("UPDATE_ALLOW_HTTP"), False):
raise RuntimeError("Plain HTTP OTA blocked; use HTTPS or UPDATE_ALLOW_HTTP=true")
try:
import urequests as requests
except ImportError:
import requests
try:
response = requests.get(url, headers=headers, timeout=timeout_s)
except TypeError:
response = requests.get(url, headers=headers)
code = getattr(response, "status_code", 200)
if code >= 400:
try:
response.close()
except Exception:
pass
raise RuntimeError("HTTP %s: %s" % (code, url))
return response
def fetch_text(url, headers, timeout_s):
response = http_get(url, headers, timeout_s)
try:
return response.text
finally:
try:
response.close()
except Exception:
pass
def fetch_file(url, path, headers, timeout_s, max_bytes):
response = http_get(url, headers, timeout_s)
total = 0
try:
with open(path, "wb") as f:
raw = getattr(response, "raw", None)
if raw is not None and hasattr(raw, "read"):
while True:
chunk = raw.read(1024)
if not chunk:
break
total += len(chunk)
if max_bytes and total > max_bytes:
raise RuntimeError("download exceeds UPDATE_MAX_FILE_BYTES")
f.write(chunk)
else:
data = response.content
total = len(data)
if max_bytes and total > max_bytes:
raise RuntimeError("download exceeds UPDATE_MAX_FILE_BYTES")
f.write(data)
finally:
try:
response.close()
except Exception:
pass
return total
class UpdateManager:
def __init__(self, env, wifi, activity):
self.env = env
self.wifi = wifi
self.activity = activity
self._next_due_ms = None
self._stopped = False
def enabled(self):
return as_bool(self.env.get("UPDATE_ENABLED"), False) and self.wifi.enabled()
def interval_seconds(self):
return max(1, as_int(self.env.get("UPDATE_CHECK_INTERVAL_HOURS"), 24)) * 3600
def idle_seconds(self):
return max(0, as_int(self.env.get("UPDATE_MIN_IDLE_SECONDS"), 300))
def busy_retry_seconds(self):
return max(60, as_int(self.env.get("UPDATE_BUSY_RETRY_SECONDS"), 900))
def boot_delay_seconds(self):
return max(0, as_int(self.env.get("UPDATE_BOOT_DELAY_SECONDS"), 180))
def _schedule_in(self, seconds):
self._next_due_ms = time.ticks_add(time.ticks_ms(), int(seconds * 1000))
def _seconds_until_due(self):
if self._next_due_ms is None:
return 0
return max(0, time.ticks_diff(self._next_due_ms, time.ticks_ms()) // 1000)
def _manifest_url(self):
explicit = self.env.get("UPDATE_MANIFEST_URL", "").strip()
if explicit:
return explicit
return raw_url_for(self.env, self.env.get("UPDATE_MANIFEST_PATH", DEFAULT_MANIFEST_PATH))
def _source_url(self, entry):
return entry.get("url") or raw_url_for(self.env, entry["source"])
def _state(self):
return read_json(STATE_FILE, {}) or {}
def _write_check_state(self, manifest=None):
state = self._state()
if epoch_is_valid():
state["last_check_epoch"] = time.time()
if manifest:
state["last_seen_release"] = manifest.get("release", "")
write_json_atomic(STATE_FILE, state)
def _boot_check_needed(self):
if not as_bool(self.env.get("UPDATE_CHECK_ON_BOOT"), True):
return False
state = self._state()
last = state.get("last_check_epoch", 0)
if epoch_is_valid() and last:
return (time.time() - last) >= self.interval_seconds()
return True
def initialize_schedule(self):
if not self.enabled():
return
if self._boot_check_needed():
self._schedule_in(self.boot_delay_seconds())
else:
self._schedule_in(self.interval_seconds())
def _files_needing_update(self, manifest):
changed = []
for entry in manifest.get("files", []):
if entry.get("enabled", True) is False:
continue
target = entry.get("target")
expected = entry.get("sha256", "").lower()
if not target or not expected:
raise ValueError("manifest file entries require target and sha256")
if not exists(target) or sha256_file(target) != expected:
changed.append(entry)
return changed
def _stage(self, entries, headers):
remove_tree_best_effort(STAGE_DIR)
mkdir_p(STAGE_DIR)
timeout_s = max(5, as_int(self.env.get("UPDATE_HTTP_TIMEOUT_SECONDS"), 20))
max_bytes = max(1024, as_int(self.env.get("UPDATE_MAX_FILE_BYTES"), 262144))
staged = []
for index, entry in enumerate(entries):
path = STAGE_DIR + "/%03d.bin" % index
log("downloading", entry["source"], "->", entry["target"])
fetch_file(self._source_url(entry), path, headers, timeout_s, max_bytes)
actual = sha256_file(path)
expected = entry["sha256"].lower()
if actual != expected:
raise RuntimeError("SHA-256 mismatch for %s" % entry["source"])
staged.append({
"stage": path,
"target": entry["target"],
"backup": entry["target"] + ".bak",
})
return staged
def _commit(self, staged, release):
pending = {"release": release, "commit_complete": False, "files": staged}
write_json_atomic(PENDING_FILE, pending)
for item in staged:
target = item["target"]
backup = item["backup"]
stage = item["stage"]
mkdir_p(dirname(target))
remove(backup)
if exists(target):
os.rename(target, backup)
os.rename(stage, target)
pending["commit_complete"] = True
write_json_atomic(PENDING_FILE, pending)
def check_now(self):
if not self.enabled():
return False
self.wifi.connect()
timeout_s = max(5, as_int(self.env.get("UPDATE_HTTP_TIMEOUT_SECONDS"), 20))
headers = auth_headers(self.env)
url = self._manifest_url()
log("checking manifest:", url)
manifest = json.loads(fetch_text(url, headers, timeout_s))
if manifest.get("schema", 1) != 1:
raise RuntimeError("unsupported manifest schema")
changed = self._files_needing_update(manifest)
self._write_check_state(manifest)
if not changed:
log("firmware current:", manifest.get("release", "unknown"))
self.wifi.disconnect_if_allowed()
return False
log("%d changed file(s); staging update" % len(changed))
staged = self._stage(changed, headers)
self._commit(staged, manifest.get("release", ""))
log("OTA committed; rebooting")
self.wifi.disconnect_if_allowed()
time.sleep_ms(250)
import machine
machine.reset()
return True
async def service(self):
"""
Energy-conscious scheduler.
Default remote traffic is ONE manifest request per 24 hours, not a
five-second poll. If the check becomes due while the device is active,
this sleeps locally for UPDATE_BUSY_RETRY_SECONDS before reconsidering.
"""
if not self.enabled():
return
self.initialize_schedule()
while not self._stopped:
wait_s = self._seconds_until_due()
if wait_s > 0:
await asyncio.sleep(wait_s)
if not self.activity.idle_for(self.idle_seconds()):
# If it is simply not idle *long enough* yet, wake at the
# earliest point the idle threshold could be met. If an
# explicit busy lock is active, use the longer configured
# busy retry. Both are local sleeps; neither hits the repo.
if self.activity.busy():
defer_s = self.busy_retry_seconds()
else:
idle_left_ms = max(0, self.idle_seconds() * 1000 - self.activity.idle_ms())
defer_s = max(60, (idle_left_ms + 999) // 1000)
log("update due but device active/not-idle; deferring", defer_s, "seconds")
self._schedule_in(defer_s)
continue
try:
self.activity.begin()
self.check_now()
except Exception as exc:
log("update check failed:", exc)
self._schedule_in(max(300, as_int(self.env.get("UPDATE_FAILURE_RETRY_SECONDS"), 3600)))
finally:
self.activity.end()
self.wifi.disconnect_if_allowed()
if self._seconds_until_due() == 0:
self._schedule_in(self.interval_seconds())
def stop(self):
self._stopped = True
def clear_pending_after_healthy_start():
pending = read_json(PENDING_FILE, None)
if not pending or not pending.get("commit_complete", False):
return
for item in pending.get("files", []):
backup = item.get("backup")
if backup:
remove(backup)
remove(PENDING_FILE)
remove_tree_best_effort(STAGE_DIR)
log("new firmware accepted; rollback backups removed")
class SystemContext:
def __init__(self, env, activity, updater):
self.env = env
self.activity = activity
self.updater = updater
self._ready = False
def ready(self):
if not self._ready:
clear_pending_after_healthy_start()
self._ready = True
def reboot(self):
import machine
machine.reset()
def load_application():
from app import main as app_main
return app_main
async def run_async():
global _GLOBAL_ENV
env = load_env()
_GLOBAL_ENV = env
if as_bool(env.get("PRINT_CONFIG_ON_BOOT"), False):
safe_config_log(env)
activity = ActivityTracker()
wifi = WiFiManager(env)
if wifi.enabled() and as_bool(env.get("WIFI_CONNECT_ON_BOOT"), True):
try:
wifi.connect()
except Exception as exc:
if as_bool(env.get("WIFI_REQUIRED"), False):
raise
log("Wi-Fi unavailable; continuing:", exc)
ble = setup_ble(env)
uart = setup_uart(env)
updater = UpdateManager(env, wifi, activity)
system = SystemContext(env, activity, updater)
ctx = {
"env": env,
"wifi": wifi,
"ble": ble,
"uart": uart,
"activity": activity,
"updater": updater,
"system": system,
}
app = load_application()
# Optional setup hook. If a newly installed app fails here, boot.py can
# restore the previous launcher/app files.
if hasattr(app, "setup"):
result = app.setup(ctx)
if hasattr(result, "__await__"):
await result
else:
system.ready()
if updater.enabled():
asyncio.create_task(updater.service())
result = app.main(ctx)
if hasattr(result, "__await__"):
await result
else:
log("synchronous main() returned; async main() is recommended")
def run():
try:
asyncio.run(run_async())
finally:
try:
asyncio.new_event_loop()
except Exception:
pass

26
firmware/manifest.json Normal file
View File

@ -0,0 +1,26 @@
{
"schema": 1,
"release": "0.1.1-probe.1",
"files": [
{
"source": "firmware/launcher.py",
"target": "/launcher.py",
"sha256": "f8ef1bba8e298877b39427870a10ea4784db33a943a5e10040bf652b4866cd2f"
},
{
"source": "firmware/app/__init__.py",
"target": "/app/__init__.py",
"sha256": "01fedc22f12a680534e37cb3c0ae5a89deb7af5a30514170ec7b7a193a7a121c"
},
{
"source": "firmware/app/digital_probe.py",
"target": "/app/digital_probe.py",
"sha256": "3d34db776bb39b5ca3a6628df2c73b07079687eaf84dc4d5c97a149064dfe409"
},
{
"source": "firmware/app/main.py",
"target": "/app/main.py",
"sha256": "5b287d984df266007df4014ece8e323339c0da594f4c88bcf4a6da1ff067a16f"
}
]
}

45
tools/build_manifest.py Normal file
View File

@ -0,0 +1,45 @@
#!/usr/bin/env python3
"""Regenerate firmware/manifest.json SHA-256 entries."""
from pathlib import Path
import hashlib
import json
import sys
ROOT = Path(__file__).resolve().parents[1]
OUT = ROOT / "firmware" / "manifest.json"
FILES = [
("firmware/launcher.py", "/launcher.py"),
("firmware/app/__init__.py", "/app/__init__.py"),
("firmware/app/digital_probe.py", "/app/digital_probe.py"),
("firmware/app/main.py", "/app/main.py"),
]
def sha256(path):
h = hashlib.sha256()
with open(path, "rb") as f:
for chunk in iter(lambda: f.read(65536), b""):
h.update(chunk)
return h.hexdigest()
def main():
release = sys.argv[1] if len(sys.argv) > 1 else "dev"
entries = []
for source, target in FILES:
entries.append({
"source": source,
"target": target,
"sha256": sha256(ROOT / source),
})
OUT.write_text(json.dumps({
"schema": 1,
"release": release,
"files": entries,
}, indent=2) + "\n", encoding="utf-8")
print("wrote", OUT)
if __name__ == "__main__":
main()