commit 935395797a15470c36b4da8ac9c8e1182c0103c5 Author: Franz Rolfsvaag Date: Fri Aug 28 10:50:16 2026 +0200 Add ESP32-C3 launcher and digital probe diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..077790c --- /dev/null +++ b/.env.example @@ -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= diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..f524e9c --- /dev/null +++ b/.gitignore @@ -0,0 +1,8 @@ +.env +*.env +!.env.example +__pycache__/ +*.pyc +.DS_Store +# Local Codex project log; intentionally never version-controlled. +AGENTS.md diff --git a/README.md b/README.md new file mode 100644 index 0000000..e1072fc --- /dev/null +++ b/README.md @@ -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. diff --git a/docs/DIGITAL_PROBE.md b/docs/DIGITAL_PROBE.md new file mode 100644 index 0000000..a108512 --- /dev/null +++ b/docs/DIGITAL_PROBE.md @@ -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. diff --git a/docs/OTA_PROTOCOL.md b/docs/OTA_PROTOCOL.md new file mode 100644 index 0000000..d5d325f --- /dev/null +++ b/docs/OTA_PROTOCOL.md @@ -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. diff --git a/docs/SETUP.md b/docs/SETUP.md new file mode 100644 index 0000000..1338b30 --- /dev/null +++ b/docs/SETUP.md @@ -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. diff --git a/firmware/app/__init__.py b/firmware/app/__init__.py new file mode 100644 index 0000000..18b665e --- /dev/null +++ b/firmware/app/__init__.py @@ -0,0 +1 @@ +"""Application package.""" diff --git a/firmware/app/digital_probe.py b/firmware/app/digital_probe.py new file mode 100644 index 0000000..8f7d418 --- /dev/null +++ b/firmware/app/digital_probe.py @@ -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 diff --git a/firmware/app/main.py b/firmware/app/main.py new file mode 100644 index 0000000..d2d3317 --- /dev/null +++ b/firmware/app/main.py @@ -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) diff --git a/firmware/boot.py b/firmware/boot.py new file mode 100644 index 0000000..bacf8d8 --- /dev/null +++ b/firmware/boot.py @@ -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 diff --git a/firmware/launcher.py b/firmware/launcher.py new file mode 100644 index 0000000..912c137 --- /dev/null +++ b/firmware/launcher.py @@ -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 diff --git a/firmware/manifest.json b/firmware/manifest.json new file mode 100644 index 0000000..409e357 --- /dev/null +++ b/firmware/manifest.json @@ -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" + } + ] +} diff --git a/tools/build_manifest.py b/tools/build_manifest.py new file mode 100644 index 0000000..d881ce5 --- /dev/null +++ b/tools/build_manifest.py @@ -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()