feat(tower): build routine manifest with cache
Adds asset_key(), build_routine_manifest(), _gen_step_audio(), _routine_cache_load() and ROUTINES_CACHE to lisael_content.py. Cache (STAGING_ROUTINES/.cache.json) skips regeneration when emoji/text unchanged and file exists on disk. SD manifest uses icon/audio keys (firmware schema). Injectable gen_bin/gen_audio for unit-testable TDD without network.
This commit is contained in:
@@ -97,6 +97,79 @@ def routines_load():
|
||||
def routines_save(routines):
|
||||
_atomic_write_json(ROUTINES_JSON, {"routines": routines})
|
||||
|
||||
# ---- routine asset manifest ----
|
||||
ROUTINES_CACHE = lambda: os.path.join(STAGING_ROUTINES, ".cache.json")
|
||||
|
||||
def asset_key(rid, idx):
|
||||
return f"{rid}_{idx:02d}"
|
||||
|
||||
def _routine_cache_load():
|
||||
p = ROUTINES_CACHE()
|
||||
if os.path.exists(p):
|
||||
try:
|
||||
return json.load(open(p, encoding="utf-8"))
|
||||
except Exception:
|
||||
pass
|
||||
return {}
|
||||
|
||||
def build_routine_manifest(gen_bin=None, gen_audio=None):
|
||||
"""Regenerate per-step .bin (emoji) + .mp3 (TTS) into STAGING_ROUTINES,
|
||||
skipping assets whose (emoji|text) is unchanged, then write the SD manifest.
|
||||
gen_bin(emoji, key)->bin_name ; gen_audio(text, key)->mp3_name (injectable)."""
|
||||
gen_bin = gen_bin or emoji_bin
|
||||
gen_audio = gen_audio or _gen_step_audio
|
||||
os.makedirs(STAGING_ROUTINES, exist_ok=True)
|
||||
cache = _routine_cache_load()
|
||||
new_cache = {}
|
||||
generated = skipped = 0
|
||||
manifest = []
|
||||
for r in routines_load():
|
||||
rid = r["id"]
|
||||
# routine icon (emoji -> .bin), key = rid
|
||||
icon_key = rid
|
||||
c = cache.get(icon_key)
|
||||
if c and c.get("emoji") == r.get("emoji") and os.path.exists(
|
||||
os.path.join(STAGING_ROUTINES, icon_key + ".bin")):
|
||||
icon = icon_key + ".bin"; skipped += 1
|
||||
else:
|
||||
icon = gen_bin(r.get("emoji", ""), icon_key) or (icon_key + ".bin"); generated += 1
|
||||
new_cache[icon_key] = {"emoji": r.get("emoji")}
|
||||
steps = []
|
||||
for i, st in enumerate(r.get("steps", []), start=1):
|
||||
k = asset_key(rid, i)
|
||||
prev = cache.get(k) or {}
|
||||
bin_name = k + ".bin"; mp3_name = k + ".mp3"
|
||||
# picto
|
||||
if prev.get("emoji") == st.get("emoji") and os.path.exists(
|
||||
os.path.join(STAGING_ROUTINES, bin_name)):
|
||||
skipped += 1
|
||||
else:
|
||||
gen_bin(st.get("emoji", ""), k); generated += 1
|
||||
# audio
|
||||
if prev.get("text") == st.get("text") and os.path.exists(
|
||||
os.path.join(STAGING_ROUTINES, mp3_name)):
|
||||
skipped += 1
|
||||
else:
|
||||
gen_audio(st.get("text", ""), k); generated += 1
|
||||
new_cache[k] = {"emoji": st.get("emoji"), "text": st.get("text")}
|
||||
steps.append({"text": st.get("text", ""), "icon": bin_name, "audio": mp3_name})
|
||||
manifest.append({"id": rid, "title": r.get("title", rid),
|
||||
"icon": icon, "steps": steps})
|
||||
_atomic_write_json(os.path.join(STAGING_ROUTINES, "routines.json"),
|
||||
{"routines": manifest})
|
||||
_atomic_write_json(ROUTINES_CACHE(), new_cache)
|
||||
return {"generated": generated, "skipped": skipped, "routines": manifest}
|
||||
|
||||
def _gen_step_audio(text, key):
|
||||
"""Default audio generator: TTS -> mono mp3 in STAGING_ROUTINES. '' on failure."""
|
||||
try:
|
||||
au = tts_speak(text)
|
||||
_atomic_write_bytes(os.path.join(STAGING_ROUTINES, key + ".mp3"), au)
|
||||
return key + ".mp3"
|
||||
except Exception as e:
|
||||
print("step audio fail", key, e, flush=True)
|
||||
return ""
|
||||
|
||||
# ---- RSS ----
|
||||
def cover_url(root):
|
||||
el = root.find(".//channel/itunes:image", NS)
|
||||
|
||||
@@ -23,6 +23,42 @@ def test_routines_seed_and_roundtrip():
|
||||
r2 = C.routines_load()
|
||||
assert r2[0]["steps"][-1]["text"] == "Cartable"
|
||||
|
||||
def test_asset_key():
|
||||
assert C.asset_key("matin", 1) == "matin_01"
|
||||
assert C.asset_key("soir", 12) == "soir_12"
|
||||
|
||||
def test_build_routine_manifest_caches():
|
||||
# deterministic content, fake generators that count calls and "write" files
|
||||
C.routines_save([{"id": "matin", "title": "Le matin", "emoji": "☀️",
|
||||
"steps": [{"text": "Habille-toi", "emoji": "\U0001F455"}]}])
|
||||
os.makedirs(C.STAGING_ROUTINES, exist_ok=True)
|
||||
calls = {"bin": 0, "audio": 0}
|
||||
def fake_bin(emoji, key):
|
||||
calls["bin"] += 1
|
||||
open(os.path.join(C.STAGING_ROUTINES, key + ".bin"), "wb").close()
|
||||
return key + ".bin"
|
||||
def fake_audio(text, key):
|
||||
calls["audio"] += 1
|
||||
open(os.path.join(C.STAGING_ROUTINES, key + ".mp3"), "wb").close()
|
||||
return key + ".mp3"
|
||||
r1 = C.build_routine_manifest(gen_bin=fake_bin, gen_audio=fake_audio)
|
||||
# 1 step => 1 step bin + 1 audio + 1 routine icon bin = 2 bin, 1 audio
|
||||
assert calls == {"bin": 2, "audio": 1}
|
||||
man = r1["routines"]
|
||||
assert man[0]["icon"] == "matin.bin"
|
||||
assert man[0]["steps"][0] == {"text": "Habille-toi", "icon": "matin_01.bin", "audio": "matin_01.mp3"}
|
||||
assert os.path.exists(os.path.join(C.STAGING_ROUTINES, "routines.json"))
|
||||
# second build, unchanged content => everything cached, no regeneration
|
||||
calls2 = {"bin": 0, "audio": 0}
|
||||
def fb(e, k): calls2["bin"] += 1; return k + ".bin"
|
||||
def fa(t, k): calls2["audio"] += 1; return k + ".mp3"
|
||||
C.build_routine_manifest(gen_bin=fb, gen_audio=fa)
|
||||
assert calls2 == {"bin": 0, "audio": 0}
|
||||
# restore default state so later tests that seed from default are not affected
|
||||
import os as _os
|
||||
if _os.path.exists(C.ROUTINES_JSON):
|
||||
_os.remove(C.ROUTINES_JSON)
|
||||
|
||||
if __name__ == "__main__":
|
||||
for name, fn in sorted(globals().items()):
|
||||
if name.startswith("test_"):
|
||||
|
||||
Reference in New Issue
Block a user