adebaumann/Selsyn_inspired_gauge
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Compare commits

base: adebaumann:Discovery
Branches Tags
adebaumann:main adebaumann:dev adebaumann:Discovery adebaumann:wrover
..
compare: adebaumann:main
Branches Tags
adebaumann:dev adebaumann:Discovery adebaumann:wrover adebaumann:main

1 Commits
Discovery ... main

Author SHA1 Message Date
Adrian A. Baumann
0fa9493928 Merge pull request 'dev' (#1) from dev into main 
Reviewed-on: #1
 2026-04-11 17:47:37 +00:00
6 changed files with 188 additions and 341 deletions
Expand all files Collapse all files

BIN
3d_model/Box for one.stl
View File

Binary file not shown.

BIN
3d_model/case_test.stl
View File

Binary file not shown.

11
config.multi.example.json
View File

@@ -20,9 +20,7 @@
"unit": "W",
"leds": {
"red_pin": 33,
"green_pin": 32,
"ws2812_red": [255, 0, 0],
"ws2812_green": [0, 255, 0]
"green_pin": 32
}
},
{
@@ -36,17 +34,14 @@
"unit": "C",
"leds": {
"red_pin": 21,
"green_pin": 20,
"ws2812_red": [255, 0, 0],
"ws2812_green": [0, 255, 0]
"green_pin": 20
}
}
],
"backlight": {
"pin": 23,
"num_leds_per_gauge": 3,
"num_status_leds_per_gauge": 2
"num_leds_per_gauge": 3
},
"device": {

2
gauge_vid6008.py
View File

@@ -108,7 +108,7 @@ class Gauge:
else:
for _ in range(abs(delta)):
self._step(1 if delta > 0 else -1)
utime.sleep_us(self._step_us)
utime.sleep_ms(self._step_us)
self._current_step = target_step

394
gaugemqttcontinuous.py
View File

@@ -23,7 +23,6 @@ Serial log format: [HH:MM:SS.mmm] LEVEL message
import network
import utime
import ujson
import urandom
from umqtt.robust import MQTTClient
from machine import Pin
from neopixel import NeoPixel
@@ -99,7 +98,6 @@ REZERO_INTERVAL_MS = int(_cfg.get("rezero_interval_ms", 3600000))
backlight_cfg = _cfg.get("backlight", {})
BACKLIGHT_PIN = int(backlight_cfg.get("pin", _cfg.get("led_bl_pin", 23)))
BACKLIGHT_LEDS_PER_GAUGE = int(backlight_cfg.get("num_leds_per_gauge", 3))
STATUS_LEDS_PER_GAUGE = int(backlight_cfg.get("num_status_leds_per_gauge", 2))
device_cfg = _cfg.get("device", {})
DEVICE_NAME = device_cfg.get("name", _cfg.get("device_name", "Selsyn Multi"))
@@ -128,8 +126,6 @@ if "gauges" in _cfg:
"unit": g.get("unit", ""),
"red_pin": int(led_cfg.get("red_pin", 33)),
"green_pin": int(led_cfg.get("green_pin", 32)),
"ws2812_red": tuple(led_cfg.get("ws2812_red", [255, 0, 0])),
"ws2812_green": tuple(led_cfg.get("ws2812_green", [0, 255, 0])),
}
)
else:
@@ -146,8 +142,6 @@ else:
"unit": _cfg.get("gauge_unit", "W"),
"red_pin": int(_cfg.get("led_red_pin", 33)),
"green_pin": int(_cfg.get("led_green_pin", 32)),
"ws2812_red": tuple(_cfg.get("ws2812_red", [255, 0, 0])),
"ws2812_green": tuple(_cfg.get("ws2812_green", [0, 255, 0])),
}
)
BL_UNIT = _cfg.get("backlight_unit", "%")
@@ -195,17 +189,18 @@ def make_gauge_topics(prefix, gauge_id):
"led_red_disc": f"homeassistant/switch/{MQTT_CLIENT_ID}_g{gauge_id}_red/config",
"led_green_disc": f"homeassistant/switch/{MQTT_CLIENT_ID}_g{gauge_id}_green/config",
"led_bl_disc": f"homeassistant/light/{MQTT_CLIENT_ID}_g{gauge_id}_bl/config",
"status_red": f"{prefix}/gauge{gauge_id}/status_led/red/set",
"status_green": f"{prefix}/gauge{gauge_id}/status_led/green/set",
"status_red_state": f"{prefix}/gauge{gauge_id}/status_led/red/state",
"status_green_state": f"{prefix}/gauge{gauge_id}/status_led/green/state",
"status_red_disc": f"homeassistant/switch/{MQTT_CLIENT_ID}_g{gauge_id}_status_red/config",
"status_green_disc": f"homeassistant/switch/{MQTT_CLIENT_ID}_g{gauge_id}_status_green/config",
}
gauge_topics = [make_gauge_topics(MQTT_PREFIX, g["id"]) for g in gauges]
T_SET = f"{MQTT_PREFIX}/set"
T_STATE = f"{MQTT_PREFIX}/state"
T_STATUS = f"{MQTT_PREFIX}/status"
T_ZERO = f"{MQTT_PREFIX}/zero"
gauge_topics = [make_gauge_topics(MQTT_PREFIX, g["id"]) for g in gauges]
T_SET = f"{MQTT_PREFIX}/set"
T_STATE = f"{MQTT_PREFIX}/state"
@@ -218,6 +213,25 @@ T_DISC_GREEN = f"homeassistant/switch/{MQTT_CLIENT_ID}_green/config"
T_DISC_BL = f"homeassistant/light/{MQTT_CLIENT_ID}_bl/config"
def make_gauge_topics(prefix, gauge_id):
return {
"set": f"{prefix}/gauge{gauge_id}/set",
"state": f"{prefix}/gauge{gauge_id}/state",
"status": f"{prefix}/gauge{gauge_id}/status",
"zero": f"{prefix}/gauge{gauge_id}/zero",
"disc": f"homeassistant/number/{MQTT_CLIENT_ID}_g{gauge_id}/config",
"led_red": f"{prefix}/gauge{gauge_id}/led/red/set",
"led_green": f"{prefix}/gauge{gauge_id}/led/green/set",
"led_bl": f"{prefix}/gauge{gauge_id}/led/backlight/set",
"led_red_state": f"{prefix}/gauge{gauge_id}/led/red/state",
"led_green_state": f"{prefix}/gauge{gauge_id}/led/green/state",
"led_bl_state": f"{prefix}/gauge{gauge_id}/led/backlight/state",
"led_red_disc": f"homeassistant/switch/{MQTT_CLIENT_ID}_g{gauge_id}_red/config",
"led_green_disc": f"homeassistant/switch/{MQTT_CLIENT_ID}_g{gauge_id}_green/config",
"led_bl_disc": f"homeassistant/light/{MQTT_CLIENT_ID}_g{gauge_id}_bl/config",
}
_DEVICE = {
"identifiers": [MQTT_CLIENT_ID],
"name": DEVICE_NAME,
@@ -235,6 +249,7 @@ _wifi_check_interval_ms = 30000
_last_wifi_check = 0
_wifi_sta = None
def connect_wifi(ssid, password, timeout_s=15):
global _wifi_sta
_wifi_sta = network.WLAN(network.STA_IF)
@@ -290,6 +305,10 @@ def check_wifi():
log_err(f"WiFi reconnect failed: {e}")
# ---------------------------------------------------------------------------
# LEDs (per gauge)
# ---------------------------------------------------------------------------
num_gauges = len(gauges)
leds_red = []
leds_green = []
@@ -298,18 +317,40 @@ for g in gauges:
leds_red.append(Pin(g["red_pin"], Pin.OUT, value=0))
leds_green.append(Pin(g["green_pin"], Pin.OUT, value=0))
total_backlight_leds = num_gauges * (BACKLIGHT_LEDS_PER_GAUGE + STATUS_LEDS_PER_GAUGE)
total_backlight_leds = num_gauges * BACKLIGHT_LEDS_PER_GAUGE
leds_bl = NeoPixel(Pin(BACKLIGHT_PIN), total_backlight_leds)
_backlight_color = (0, 0, 0)
_backlight_brightness = 100
_backlight_on = False
_bl_dirty_since = None
_BL_SAVE_DELAY_MS = 5000
backlight_color = [(0, 0, 0) for _ in range(num_gauges)]
backlight_brightness = [100 for _ in range(num_gauges)]
backlight_on = [False for _ in range(num_gauges)]
status_led_red = [False for _ in range(num_gauges)]
status_led_green = [False for _ in range(num_gauges)]
_bl_dirty_since = None
_BL_SAVE_DELAY_MS = 5000
def _flush_backlight(client):
for i in range(num_gauges):
gt = gauge_topics[i]
payload = {
"state": "ON" if backlight_on[i] else "OFF",
"color": {
"r": backlight_color[i][0],
"g": backlight_color[i][1],
"b": backlight_color[i][2],
},
"brightness": int(backlight_brightness[i] * 2.55),
}
client.publish(
f"{gt['set'].replace('/set', '/backlight/state')}",
ujson.dumps(payload),
retain=True,
)
info(
f"Gauge {i} backlight: {payload['state']} {backlight_color[i]} @ {backlight_brightness[i]}%"
)
def _backlight_changed(gauge_idx, new_color, new_on, new_brightness):
@@ -338,11 +379,9 @@ def set_backlight_color(gauge_idx, r, g, b, brightness=None):
backlight_on[gauge_idx] = new_on
scale = brightness / 100
leds_per_gauge = BACKLIGHT_LEDS_PER_GAUGE + STATUS_LEDS_PER_GAUGE
base_idx = gauge_idx * leds_per_gauge
base_idx = gauge_idx * BACKLIGHT_LEDS_PER_GAUGE
for j in range(BACKLIGHT_LEDS_PER_GAUGE):
leds_bl[base_idx + j] = (int(g * scale), int(r * scale), int(b * scale))
_update_status_leds(gauge_idx)
leds_bl.write()
_mark_bl_dirty()
@@ -358,73 +397,21 @@ def set_backlight_brightness(gauge_idx, brightness):
backlight_on[gauge_idx] = new_on
r, g, b = backlight_color[gauge_idx]
scale = clamped / 100
leds_per_gauge = BACKLIGHT_LEDS_PER_GAUGE + STATUS_LEDS_PER_GAUGE
base_idx = gauge_idx * leds_per_gauge
base_idx = gauge_idx * BACKLIGHT_LEDS_PER_GAUGE
for j in range(BACKLIGHT_LEDS_PER_GAUGE):
leds_bl[base_idx + j] = (int(g * scale), int(r * scale), int(b * scale))
_update_status_leds(gauge_idx)
leds_bl.write()
_mark_bl_dirty()
def _update_status_leds(gauge_idx):
leds_per_gauge = BACKLIGHT_LEDS_PER_GAUGE + STATUS_LEDS_PER_GAUGE
base_idx = gauge_idx * leds_per_gauge + BACKLIGHT_LEDS_PER_GAUGE
# ---------------------------------------------------------------------------
# State
# ---------------------------------------------------------------------------
g_cfg = gauges[gauge_idx]
red_color = g_cfg["ws2812_red"]
green_color = g_cfg["ws2812_green"]
if status_led_red[gauge_idx]:
leds_bl[base_idx] = (red_color[1], red_color[0], red_color[2])
else:
leds_bl[base_idx] = (0, 0, 0)
if status_led_green[gauge_idx]:
leds_bl[base_idx + 1] = (green_color[1], green_color[0], green_color[2])
else:
leds_bl[base_idx + 1] = (0, 0, 0)
def set_status_led(gauge_idx, led_type, state):
global status_led_red, status_led_green
if led_type == "red":
status_led_red[gauge_idx] = state
elif led_type == "green":
status_led_green[gauge_idx] = state
_update_status_leds(gauge_idx)
leds_bl.write()
def publish_backlight_states(client):
"""Publish current backlight state for all gauges as retained MQTT messages."""
for i in range(num_gauges):
gt = gauge_topics[i]
r, g, b = backlight_color[i]
brightness = backlight_brightness[i]
state = {
"state": "ON" if backlight_on[i] else "OFF",
"color_mode": "rgb",
"brightness": int(brightness * 2.55),
"color": {"r": r, "g": g, "b": b},
}
try:
client.publish(gt["led_bl_state"], ujson.dumps(state), retain=True)
except Exception as e:
log_err(f"Backlight state publish failed for gauge {i}: {e}")
def _flush_backlight_state():
global _bl_dirty_since
if _bl_dirty_since is None:
return
if utime.ticks_diff(utime.ticks_ms(), _bl_dirty_since) < _BL_SAVE_DELAY_MS:
return
if client_ref is None:
return
publish_backlight_states(client_ref)
_bl_dirty_since = None
info("Backlight state flushed to MQTT")
_last_rezero_ms = None # set to ticks_ms() in main()
client_ref = None
_mqtt_connected = False
_last_mqtt_check = 0
def _publish(topic, payload, retain=False):
@@ -445,7 +432,6 @@ def _publish(topic, payload, retain=False):
def on_message(topic, payload):
global backlight_brightness, backlight_color
if client_ref is None:
return
topic = topic.decode()
@@ -469,6 +455,15 @@ def on_message(topic, payload):
warn(f"Invalid set value for gauge {i}: '{payload}'")
return
if topic == T_ZERO:
for i, g in enumerate(gauge_objects):
info(f"Zeroing all gauges")
g.zero()
gauge_last_rezero[i] = utime.ticks_ms()
info("All gauges zeroed")
return
for i, gt in enumerate(gauge_topics):
if topic == gt["led_red"]:
state = payload.upper() == "ON"
leds_red[i].value(1 if state else 0)
@@ -518,73 +513,12 @@ def on_message(topic, payload):
info(f"Gauge {i} backlight → #{r:02x}{g:02x}{b:02x} @ {brightness}%")
return
if topic == gt["status_red"]:
state = payload.upper() == "ON"
set_status_led(i, "red", state)
_publish(gt["status_red_state"], "ON" if state else "OFF", retain=True)
info(f"Gauge {i} status red → {'ON' if state else 'OFF'}")
return
if topic == gt["status_green"]:
state = payload.upper() == "ON"
set_status_led(i, "green", state)
_publish(gt["status_green_state"], "ON" if state else "OFF", retain=True)
info(f"Gauge {i} status green → {'ON' if state else 'OFF'}")
return
if topic == T_ZERO:
for i, g in enumerate(gauge_objects):
g.zero()
gauge_last_rezero[i] = utime.ticks_ms()
info("All gauges zeroed")
return
if topic == T_SET:
try:
data = ujson.loads(payload)
if isinstance(data, dict):
for i, val in enumerate(data.values()):
if i < len(gauges):
g = gauges[i]
gauge_targets[i] = max(g["min"], min(g["max"], float(val)))
info(f"Gauge {i} target → {gauge_targets[i]:.1f}")
else:
val = float(payload)
for i in range(len(gauges)):
gauge_targets[i] = max(gauges[i]["min"], min(gauges[i]["max"], val))
info(f"All gauges target → {val:.1f}")
except Exception:
try:
val = float(payload)
for i in range(len(gauges)):
gauge_targets[i] = max(gauges[i]["min"], min(gauges[i]["max"], val))
info(f"All gauges target → {val:.1f}")
except:
warn(f"Invalid set value: '{payload}'")
return
# ---------------------------------------------------------------------------
# MQTT connect + discovery
# ---------------------------------------------------------------------------
def _subscribe_all(c):
c.subscribe(f"{MQTT_PREFIX}/set")
c.subscribe(f"{MQTT_PREFIX}/zero")
for i in range(num_gauges):
prefix = f"{MQTT_PREFIX}/gauge{i}"
c.subscribe(f"{prefix}/set")
c.subscribe(f"{prefix}/zero")
c.subscribe(f"{prefix}/led/red/set")
c.subscribe(f"{prefix}/led/green/set")
c.subscribe(f"{prefix}/led/backlight/set")
c.subscribe(f"{prefix}/status_led/red/set")
c.subscribe(f"{prefix}/status_led/green/set")
def connect_mqtt():
global client_ref, _mqtt_connected
info(f"Connecting to MQTT broker {MQTT_BROKER}:{MQTT_PORT} ...")
@@ -594,22 +528,27 @@ def connect_mqtt():
port=MQTT_PORT,
user=MQTT_USER,
password=MQTT_PASSWORD,
keepalive=30,
keepalive=60,
)
# Don't set last will - it might be causing issues
# client.set_last_will(T_STATUS, b"offline", retain=True, qos=0)
client.set_last_will(T_STATUS, b"offline", retain=True, qos=0)
client.set_callback(on_message)
client.connect()
client_ref = client
client.subscribe(T_SET)
client.subscribe(T_ZERO)
for gt in gauge_topics:
client.subscribe(gt["set"])
client.subscribe(gt["zero"])
client.subscribe(gt["led_red"])
client.subscribe(gt["led_green"])
client.subscribe(gt["led_bl"])
_mqtt_connected = True
info(f"MQTT connected client_id={MQTT_CLIENT_ID}")
return client
_mqtt_check_interval_ms = 30000
_last_mqtt_check = 0
client_ref = None
_mqtt_connected = False
def check_mqtt():
@@ -645,7 +584,14 @@ def check_mqtt():
client_ref.set_last_will(T_STATUS, b"offline", retain=True, qos=0)
client_ref.set_callback(on_message)
client_ref.connect()
_subscribe_all(client_ref)
client_ref.subscribe(T_SET)
client_ref.subscribe(T_ZERO)
for gt in gauge_topics:
client_ref.subscribe(gt["set"])
client_ref.subscribe(gt["zero"])
client_ref.subscribe(gt["led_red"])
client_ref.subscribe(gt["led_green"])
client_ref.subscribe(gt["led_bl"])
_mqtt_connected = True
info("MQTT reconnected!")
publish_discovery(client_ref)
@@ -659,10 +605,15 @@ def check_mqtt():
return False
def publish_discovery(client):
"""Publish all HA MQTT discovery payloads for gauges and LEDs."""
_dev_ref = _DEVICE
_dev_ref = {
"identifiers": [MQTT_CLIENT_ID],
"name": DEVICE_NAME,
"model": DEVICE_MODEL,
"manufacturer": DEVICE_MFR,
"suggested_area": DEVICE_AREA,
}
for i, g in enumerate(gauges):
gt = gauge_topics[i]
@@ -688,11 +639,6 @@ def publish_discovery(client):
)
info(f"Discovery: gauge {i} ({g['name']})")
# Process MQTT messages between gauges
for _ in range(5):
client.check_msg()
utime.sleep_ms(10)
client.publish(
gt["led_red_disc"],
ujson.dumps(
@@ -731,11 +677,6 @@ def publish_discovery(client):
)
info(f"Discovery: gauge {i} green LED")
# Process MQTT messages
for _ in range(5):
client.check_msg()
utime.sleep_ms(10)
client.publish(
gt["led_bl_disc"],
ujson.dumps(
@@ -755,55 +696,14 @@ def publish_discovery(client):
)
info(f"Discovery: gauge {i} backlight")
client.publish(
gt["status_red_disc"],
ujson.dumps(
{
"name": f"{g['name']} Status Red",
"uniq_id": f"{MQTT_CLIENT_ID}_g{i}_status_red",
"cmd_t": gt["status_red"],
"stat_t": gt["status_red_state"],
"pl_on": "ON",
"pl_off": "OFF",
"icon": "mdi:led-on",
"dev": _dev_ref,
"ret": True,
}
),
retain=True,
)
info(f"Discovery: gauge {i} status red")
client.publish(
gt["status_green_disc"],
ujson.dumps(
{
"name": f"{g['name']} Status Green",
"uniq_id": f"{MQTT_CLIENT_ID}_g{i}_status_green",
"cmd_t": gt["status_green"],
"stat_t": gt["status_green_state"],
"pl_on": "ON",
"pl_off": "OFF",
"icon": "mdi:led-on",
"dev": _dev_ref,
"ret": True,
}
),
retain=True,
)
info(f"Discovery: gauge {i} status green")
# Process between gauges to avoid MQTT blocking
for _ in range(5):
client.check_msg()
utime.sleep_ms(10)
def publish_state(client):
for i, g in enumerate(gauge_objects):
gt = gauge_topics[i]
val = g.get()
client.publish(gt["state"], str(val))
client.publish(gt["state"], str(round(val, 1)), retain=True)
client.publish(gt["status"], "online", retain=True)
info(f"Gauge {i} state: {val:.1f} step={g._current_step}")
# ---------------------------------------------------------------------------
@@ -818,52 +718,32 @@ def main():
connect_wifi(WIFI_SSID, WIFI_PASSWORD)
# Connect MQTT (no subscriptions yet — keeps broker silent during discovery)
connect_mqtt()
# Publish discovery — broker has nothing to send back yet
info("Publishing discovery...")
publish_discovery(client_ref)
# Subscribe now — retained messages will start arriving from here
_subscribe_all(client_ref)
info("Draining initial retained messages...")
for _ in range(50):
client_ref.check_msg()
utime.sleep_ms(20)
# Now initialize gauges
info("Zeroing gauges on startup ...")
for i, g in enumerate(gauge_objects):
g.zero()
info(f"Zeroed gauge {i}")
info("Zero complete")
info("Publishing state...")
connect_mqtt()
publish_discovery(client_ref)
publish_state(client_ref)
utime.sleep_ms(50)
for _ in range(5):
client_ref.check_msg()
utime.sleep_ms(20)
info("Entering main loop")
info("-" * 48)
try:
import ota
ota.mark_ok()
except:
info("OTA OK flag set")
except ImportError:
pass
# Initialize variables for main loop
global _bl_dirty_since
last_heartbeat = utime.ticks_ms()
now = 0
was_moving = False
while True:
try:
now = utime.ticks_ms()
check_wifi()
if not check_mqtt():
@@ -872,35 +752,49 @@ def main():
client_ref.check_msg()
pending = []
for i, g in enumerate(gauge_objects):
delta = g._val_to_step(gauge_targets[i]) - g._current_step
steps = max(-5, min(5, delta))
pending.append(steps)
now = utime.ticks_ms()
moved_any = any(s != 0 for s in pending)
if moved_any:
was_moving = True
delay_us = 1_000_000 // MICROSTEPS_PER_SECOND
for tick in range(max(abs(s) for s in pending)):
moved_any = False
for i, g in enumerate(gauge_objects):
if tick < abs(pending[i]):
g.step(1 if pending[i] > 0 else -1)
current_target = g._val_to_step(gauge_targets[i])
if current_target != g._current_step:
direction = 1 if current_target > g._current_step else -1
steps_to_move = current_target - g._current_step
steps_to_move = max(-5, min(5, steps_to_move))
for _ in range(abs(steps_to_move)):
g.step(direction)
moved_any = True
if moved_any:
delay_us = 1_000_000 // MICROSTEPS_PER_SECOND
utime.sleep_us(delay_us)
else:
if was_moving:
if (
REZERO_INTERVAL_MS > 0
and utime.ticks_diff(now, gauge_last_rezero[0]) >= REZERO_INTERVAL_MS
):
for i, g in enumerate(gauge_objects):
info(f"Auto-rezero gauge {i}")
saved = gauge_targets[i]
g.zero()
if saved > gauges[i]["min"]:
g.set(saved)
gauge_last_rezero[i] = now
publish_state(client_ref)
was_moving = False
utime.sleep_ms(10)
info("Auto-rezero complete")
if (
_bl_dirty_since is not None
and utime.ticks_diff(now, _bl_dirty_since) >= _BL_SAVE_DELAY_MS
):
_flush_backlight(client_ref)
_bl_dirty_since = None
if utime.ticks_diff(now, last_heartbeat) >= HEARTBEAT_MS:
info(f"Heartbeat: {gauge_targets}")
publish_state(client_ref)
last_heartbeat = now
except Exception as e:
import sys
sys.print_exception(e)
log_err(f"Main loop error: {e} — continuing")
utime.sleep_ms(100)

72
ota.py
View File

@@ -82,55 +82,31 @@ OTA_MANIFEST = "ota_manifest.txt"
# Logging
# ---------------------------------------------------------------------------
def _ts():
ms = utime.ticks_ms()
return f"{(ms//3600000)%24:02d}:{(ms//60000)%60:02d}:{(ms//1000)%60:02d}.{ms%1000:03d}"
def _log(level, msg):
print(f"[{_ts()}] {level:5s} [OTA] {msg}")
def info(msg):
_log("INFO", msg)
def warn(msg):
_log("WARN", msg)
def log_err(msg):
_log("ERROR", msg)
def _log(level, msg): print(f"[{_ts()}] {level:5s} [OTA] {msg}")
def info(msg): _log("INFO", msg)
def warn(msg): _log("WARN", msg)
def log_err(msg): _log("ERROR", msg)
# ---------------------------------------------------------------------------
# HTTP helpers
# ---------------------------------------------------------------------------
def _headers():
h = {"Accept": "application/json"}
if API_TOKEN:
h["Authorization"] = f"token {API_TOKEN}"
return h
# ---------------------------------------------------------------------------
# Config loader
# ---------------------------------------------------------------------------
def load_config():
global \
GITEA_BASE, \
REPO_OWNER, \
REPO_NAME, \
REPO_FOLDER, \
REPO_BRANCH, \
API_TOKEN, \
WIFI_SSID, \
WIFI_PASSWORD
global GITEA_BASE, REPO_OWNER, REPO_NAME, REPO_FOLDER, REPO_BRANCH, API_TOKEN, WIFI_SSID, WIFI_PASSWORD
try:
with open(SETTINGS_FILE) as f:
cfg = ujson.load(f)
@@ -148,12 +124,10 @@ def load_config():
except Exception as e:
warn(f"Config parse error: {e} — using defaults")
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _match_pattern(name, pattern):
if "*" not in pattern:
return name == pattern
@@ -176,9 +150,11 @@ def _match_pattern(name, pattern):
i += 1
return i == n and j == m
def _fetch_commit_sha():
url = f"{GITEA_BASE}/api/v1/repos/{REPO_OWNER}/{REPO_NAME}/branches/{REPO_BRANCH}"
url = (
f"{GITEA_BASE}/api/v1/repos/{REPO_OWNER}/{REPO_NAME}"
f"/branches/{REPO_BRANCH}"
)
try:
r = urequests.get(url, headers=_headers())
if r.status_code == 200:
@@ -190,7 +166,6 @@ def _fetch_commit_sha():
log_err(f"Failed to fetch commit: {e}")
return None
def _fetch_manifest():
url = (
f"{GITEA_BASE}/api/v1/repos/{REPO_OWNER}/{REPO_NAME}"
@@ -203,7 +178,6 @@ def _fetch_manifest():
r.close()
if data.get("content"):
import ubinascii
content = ubinascii.a2b_base64(data["content"]).decode()
patterns = [line.strip() for line in content.splitlines()]
return [p for p in patterns if p and not p.startswith("#")]
@@ -214,7 +188,6 @@ def _fetch_manifest():
log_err(f"Failed to fetch manifest: {e}")
return None
def _fetch_dir(path):
url = (
f"{GITEA_BASE}/api/v1/repos/{REPO_OWNER}/{REPO_NAME}"
@@ -222,7 +195,6 @@ def _fetch_dir(path):
)
return _api_get(url)
def _api_get(url):
"""GET a URL and return parsed JSON, or None on failure."""
try:
@@ -237,7 +209,6 @@ def _api_get(url):
log_err(f"GET {url} failed: {e}")
return None
def _download(url, dest_path):
"""Download url to dest_path. Returns True on success."""
tmp = dest_path + ".tmp"
@@ -265,7 +236,6 @@ def _download(url, dest_path):
pass
return False
def _load_manifest():
try:
with open(MANIFEST_FILE) as f:
@@ -273,7 +243,6 @@ def _load_manifest():
except Exception:
return {}
def _save_manifest(manifest, commit_sha=None):
try:
with open(MANIFEST_FILE, "w") as f:
@@ -283,7 +252,6 @@ def _save_manifest(manifest, commit_sha=None):
except Exception as e:
warn(f"Could not save manifest: {e}")
def _wipe_manifest():
try:
os.remove(MANIFEST_FILE)
@@ -291,7 +259,6 @@ def _wipe_manifest():
except OSError:
pass
def _ok_flag_exists():
try:
os.stat(OK_FLAG_FILE)
@@ -299,14 +266,12 @@ def _ok_flag_exists():
except OSError:
return False
def _clear_ok_flag():
try:
os.remove(OK_FLAG_FILE)
except OSError:
pass
def mark_ok():
"""
Call this from main.py after successful startup.
@@ -318,12 +283,10 @@ def mark_ok():
except Exception as e:
warn(f"Could not write OK flag: {e}")
# ---------------------------------------------------------------------------
# Core update logic
# ---------------------------------------------------------------------------
def _fetch_file_list():
"""
Returns list of {name, sha, download_url} dicts based on the
@@ -356,13 +319,11 @@ def _fetch_file_list():
if _match_pattern(name, p) or _match_pattern(entry["path"], p):
if entry["path"] not in visited:
visited.add(entry["path"])
files.append(
{
files.append({
"name": entry["path"],
"sha": entry["sha"],
"download_url": entry["download_url"],
}
)
})
break
root = _fetch_dir(REPO_FOLDER)
@@ -372,15 +333,12 @@ def _fetch_file_list():
fetch_matching(root, manifest_patterns)
return files
def _do_update(commit_sha=None):
"""
Fetch file list, download changed files, update manifest.
Returns True if all succeeded (or nothing needed updating).
"""
info(
f"Checking {GITEA_BASE}/{REPO_OWNER}/{REPO_NAME}/{REPO_FOLDER} @ {REPO_BRANCH}"
)
info(f"Checking {GITEA_BASE}/{REPO_OWNER}/{REPO_NAME}/{REPO_FOLDER} @ {REPO_BRANCH}")
file_list = _fetch_file_list()
if file_list is None:
log_err("Could not fetch file list — skipping update")
@@ -424,12 +382,10 @@ def _do_update(commit_sha=None):
return True
# ---------------------------------------------------------------------------
# Public entry point
# ---------------------------------------------------------------------------
def update():
"""
Main entry point. Call from boot.py before importing application code.
@@ -452,7 +408,9 @@ def update():
if not ok_flag:
warn("OK flag missing — last boot may have failed")
warn("Re-checking all files, will only download changed ones")
warn("Wiping manifest to force full re-fetch")
_wipe_manifest()
manifest = {}
else:
info("OK flag present — last boot was good")