852 lines
25 KiB
Python
852 lines
25 KiB
Python
"""
|
|
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
|
|
|
|
|
|
def is_awaitable(value):
|
|
"""Support CPython and MicroPython coroutine objects.
|
|
|
|
On the ESP32 MicroPython port, native coroutine objects expose ``send``
|
|
but not ``__await__``. Checking both lets application hooks be either
|
|
synchronous functions or ``async def`` coroutines.
|
|
"""
|
|
return hasattr(value, "__await__") or hasattr(value, "send")
|
|
|
|
|
|
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 is_awaitable(result):
|
|
await result
|
|
else:
|
|
system.ready()
|
|
|
|
if updater.enabled():
|
|
asyncio.create_task(updater.service())
|
|
|
|
result = app.main(ctx)
|
|
if is_awaitable(result):
|
|
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
|