diff --git a/.gitignore b/.gitignore index af71aa4d..5c65e02a 100644 --- a/.gitignore +++ b/.gitignore @@ -276,3 +276,9 @@ docs/mobile-app-design.md inksight_tech/ lab/ .cursor/ + +# Claude Code local session/worktree state +.claude/ + +# Firmware build output +firmware.bin diff --git a/backend-lite/.env.example b/backend-lite/.env.example new file mode 100644 index 00000000..4465809e --- /dev/null +++ b/backend-lite/.env.example @@ -0,0 +1,11 @@ +# backend-lite 环境变量 + +# LLM key 加密密钥（core/crypto.py 用）。首次启动可用 `python -c "from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())"` 生成。 +LLM_ENCRYPTION_KEY= + +# 服务监听（仅在使用 `python run.py` 启动时生效） +LITE_HOST=0.0.0.0 +LITE_PORT=8090 + +# 上游 backend/ 目录的绝对路径（默认同级 ../backend）。仅在非标准布局下需要覆盖。 +INKSIGHT_BACKEND_DIR= \ No newline at end of file diff --git a/backend-lite/.gitignore b/backend-lite/.gitignore new file mode 100644 index 00000000..e52409a3 --- /dev/null +++ b/backend-lite/.gitignore @@ -0,0 +1,21 @@ +# 运行时数据库 +*.db +*.db-shm +*.db-wal + +# 虚拟环境 +.venv/ +venv/ + +# OTA 上传的固件 +ota_files/ + +# 环境 +.env + +# 字体（由 setup_fonts.py 下载，不入库） +fonts/ + +# Python 缓存 +__pycache__/ +*.pyc \ No newline at end of file diff --git a/backend-lite/README.md b/backend-lite/README.md new file mode 100644 index 00000000..900ce5f3 --- /dev/null +++ b/backend-lite/README.md @@ -0,0 +1,129 @@ +# InkSight backend-lite + +InkSight 的轻量单用户后端。直接复用上游 `backend/core/` 渲染内核（保持上游可同步）， +自己只维护薄路由 + 单用户 store + 单页管理页。100% 兼容现有固件协议。 + +## 特性 + +- 单用户、单设备，无账户/共享/配额/analytics 等多用户平台重量。 +- 兼容固件核心链路：render / token / heartbeat / config / state / runtime / refresh。 +- 支持焦点提醒 alert-bmp、始终活跃 always_active、专注监听 focus_listening。 +- 支持 OTA 固件更新（后端下发 ota_url + 固件 state 解析器补丁）。 +- 管理页：单静态 HTML，配置自己的 LLM API key（加密存 DB）、选模式、预览、远程刷新。 +- 复用上游全部 30 个内置模式；自定义模式直接放 `backend/core/modes/custom/` 即可。 +- 渲染固定 400×300 / 2bpp / 4 色（黑白红黄）。 + +不支持（固件遇非 200 自动降级）：vocab 词汇复习、voice 语音对话、mode marketplace、自定义模式编辑器 UI。 + +## 管理页截图 + +### 管理页总览 + +![backend-lite 管理页总览](imgs/1.png) + +### 设备配置与预览 + +![backend-lite 设备配置与预览](imgs/2.png) + +### 焦点提醒、OTA 与设备控制 + +![backend-lite 焦点提醒、OTA 与设备控制](imgs/3.png) + +## 与上游的关系 + +``` +backend/core/ ← 上游原样，git pull 即同步渲染/模式/LLM 逻辑 +backend-lite/ ← 本目录，只依赖 core/ 的公开 API + api/ 薄路由（固件端点 + 管理端点） + adaptee/ 渲染适配 + 上游 DB 路径重定向 + store/ 单用户 SQLite（lite.db） + static/ 单页管理 UI +``` + +`backend-lite/adaptee/db_redirect.py` 在启动时把上游 `core/` 的 `inksight.db`/`cache.db` +路径重定向到本目录，避免污染上游 `backend/`，并调上游 `init_stats_db()`/`init_db()` +建 `content_history` 等表（LLM 去重提示需要）。 + +## 安装 + +```bash +cd backend-lite +python3 -m venv .venv +. .venv/bin/activate +pip install -r requirements.txt +# 上游 core/ 运行时依赖（zhdate/dashscope/tenacity 等） +pip install -r ../backend/requirements.txt + +# 字体（渲染需要，~70MB，首次必做） +python ../backend/scripts/setup_fonts.py + +# 配置环境 +cp .env.example .env +# 生成加密密钥填入 LLM_ENCRYPTION_KEY： +python -c "from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())" +``` + +## 运行 + +推荐使用本目录自带启动入口，它会读取 `.env` 中的 `LITE_HOST` / `LITE_PORT`： + +```bash +. .venv/bin/activate +python run.py +``` + +默认监听 `0.0.0.0:8090`；如果你修改了 `.env`，例如： + +```env +LITE_HOST=0.0.0.0 +LITE_PORT=21568 +``` + +那么重新运行 `python run.py` 后就会监听在对应地址。 + +> 如果你直接运行 `uvicorn api.index:app --host ... --port ...`，命令行参数会覆盖 `.env`，这也是之前看起来“改了 `.env` 不生效”的原因。 + +管理页：浏览器打开 `http://<本机IP>:/`（局域网免认证）。 + +## 配置流程 + +1. 管理页「LLM API Key」填 provider + key（deepseek/aliyun/moonshot/openai_compat），保存。 +2. 「设备配置」选模式、设刷新策略/间隔/城市，保存。 +3. 固件 captive portal 里把后端地址填成 `http://<本机IP>:8090`。 +4. 设备开机即取图。开「始终活跃」可远程触发刷新 / 切模式。 + +## 固件 OTA（可选） + +本目录附带固件补丁（`firmware/src/network.cpp` + `main.cpp`）： +- `network.cpp` 的 `/state` 解析器新增提取 `ota_url`/`ota_version` 赋给 `g_pending_ota_*`。 +- `main.cpp` 主循环新增 `checkAndPerformOTA()` 调用（上游机器已实现，原本缺触发）。 + +需用 PlatformIO 重新编译刷写固件： + +```bash +cd firmware +pio run --target upload +``` + +之后管理页「OTA」上传 bin → 设备下次轮询 `/state` 自动下载刷写 → `/ota/progress` 上报进度。 + +## 端点速查 + +固件（设备调用）： +- `POST /api/device/{mac}/token` · `POST /api/device/{mac}/heartbeat` +- `GET /api/render` · `GET /api/config/{mac}` · `GET /api/device/{mac}/state` +- `POST /api/config` · `POST /api/device/{mac}/runtime` · `POST /api/device/{mac}/refresh` +- `GET /api/device/{mac}/alert-bmp` · `POST /api/device/{mac}/ota/progress` +- `POST /api/device/{mac}/claim-token` + +管理页： +- `GET/PUT /api/admin/config` · `GET /api/admin/modes` · `GET/PUT /api/admin/llm-key` +- `GET /api/admin/state` · `POST /api/admin/refresh` · `POST /api/admin/set-mode` · `POST /api/admin/runtime` +- `GET/PUT /api/admin/alert` · `POST /api/admin/ota/upload` · `POST /api/admin/ota/set` · `GET /api/admin/ota/file/{name}` +- `GET /api/preview?mode=&as_png=1` + +## 数据库 + +- `lite.db` — backend-lite 自有：config / device_state / llm_key / heartbeats / alert_state。 +- `inksight.db` — 上游 core 运行时表（content_history 去重等），重定向到本目录。 +- `cache.db` — 上游渲染缓存（重定向到本目录）。 diff --git a/backend-lite/adaptee/__init__.py b/backend-lite/adaptee/__init__.py new file mode 100644 index 00000000..3558c137 --- /dev/null +++ b/backend-lite/adaptee/__init__.py @@ -0,0 +1 @@ +"""backend-lite adaptee 包标记。""" \ No newline at end of file diff --git a/backend-lite/adaptee/context.py b/backend-lite/adaptee/context.py new file mode 100644 index 00000000..cc960375 --- /dev/null +++ b/backend-lite/adaptee/context.py @@ -0,0 +1,28 @@ +"""适配 core.context：取 date_ctx + weather。 + +单用户：城市取自 config（或经纬度），直接调 core 的缓存版本。 +""" +from __future__ import annotations + +from typing import Any + +from core.context import get_date_context_cached, get_weather_cached + + +async def build_context(config: dict[str, Any]) -> tuple[dict[str, Any], dict[str, Any]]: + """返回 (date_ctx, weather)。失败降级到空 dict / 默认天气。""" + date_ctx = await get_date_context_cached() + + city = config.get("city") or "" + lat = config.get("latitude") + lon = config.get("longitude") + weather: dict[str, Any] + try: + if lat is not None and lon is not None: + from core.context import get_weather + weather = await get_weather(lat=float(lat), lon=float(lon)) + else: + weather = await get_weather_cached(city=city or None) + except Exception: + weather = {"temp": 0, "weather_code": -1, "weather_str": "--°C"} + return date_ctx, weather \ No newline at end of file diff --git a/backend-lite/adaptee/db_redirect.py b/backend-lite/adaptee/db_redirect.py new file mode 100644 index 00000000..8dabca27 --- /dev/null +++ b/backend-lite/adaptee/db_redirect.py @@ -0,0 +1,48 @@ +"""把上游 core/ 的 SQLite 路径重定向到 backend-lite 目录，避免污染上游 backend/。 + +core 运行时（json_content/pipeline）会 lazy 调用 stats_store / config_store / +cache 的函数，它们读写 backend/inksight.db 和 backend/cache.db。这里把这些 +模块级路径常量改写为 backend-lite 目录下的副本，并调上游建表函数初始化。 + +必须在任何 core DB 调用前调用 redirect_db_paths()。 +""" +from __future__ import annotations + +import os +from pathlib import Path + +_LITE_DIR = Path(__file__).resolve().parent.parent +_INKSIGHT_DB = str(_LITE_DIR / "inksight.db") +_CACHE_DB = str(_LITE_DIR / "cache.db") + + +def redirect_db_paths() -> None: + """改写 core 各模块的 DB 路径常量，指向 backend-lite 目录。""" + import core.db as cdb + import core.stats_store as stats + import core.config_store as cstore + import core.cache as cache + + cdb._MAIN_DB_PATH = _INKSIGHT_DB + cdb._CACHE_DB_PATH = _CACHE_DB + stats.DB_PATH = _INKSIGHT_DB + cstore.DB_PATH = _INKSIGHT_DB + # cache 模块若用 _CACHE_DB_PATH 常量 + if hasattr(cache, "_CACHE_DB_PATH"): + cache._CACHE_DB_PATH = _CACHE_DB + if hasattr(cache, "CACHE_DB_PATH"): + cache.CACHE_DB_PATH = _CACHE_DB + + +async def init_upstream_tables() -> None: + """调用上游建表函数（建在重定向后的 backend-lite 库里）。 + + 只建 core 运行时真正需要的表：stats_store（content_history 去重、 + device_heartbeats 在线判断）、config_store（photo_frame_index 等）。 + 多用户相关表也会被 config_store.init_db 创建，空着不用，无害。 + """ + from core.stats_store import init_stats_db + from core.config_store import init_db as config_init_db + + await init_stats_db() + await config_init_db() \ No newline at end of file diff --git a/backend-lite/adaptee/render.py b/backend-lite/adaptee/render.py new file mode 100644 index 00000000..4ae068e5 --- /dev/null +++ b/backend-lite/adaptee/render.py @@ -0,0 +1,203 @@ +"""渲染适配层：模式选择 + 调 core.pipeline.generate_and_render + BMP 编码。 + +刻意不复用上游 api/shared.build_image（耦合多用户 store）。这里自己实现 +单用户单设备的极薄渲染路径：每次请求直接生成，不做缓存（单设备刷图 +频率低，LLM 调用本就该新鲜；后续如需省调用再加 cache）。 +""" +from __future__ import annotations + +import logging +import random +from datetime import datetime +from typing import Any + +from PIL import Image + +from core.context import calc_battery_pct +from core.mode_registry import get_registry +from core.pipeline import generate_and_render +from core.renderer import image_to_bmp_bytes, image_to_raw_2bpp + +from .context import build_context +from store import config_store, secret_store + +logger = logging.getLogger("backend-lite.render") + +# 与上游 api/shared._SMART_TIME_SLOTS 一致 +_SMART_TIME_SLOTS = [ + (6, 9, ["RECIPE", "DAILY"]), + (9, 12, ["BRIEFING", "STOIC"]), + (12, 14, ["ZEN", "POETRY"]), + (14, 18, ["STOIC", "ROAST"]), + (18, 21, ["FITNESS", "RECIPE"]), + (21, 24, ["ZEN", "POETRY"]), + (0, 6, ["ZEN", "POETRY"]), +] + +DEFAULT_MODES = ["STOIC", "ROAST", "ZEN", "DAILY"] + + +def _choose_persona(config: dict[str, Any], mac: str) -> str: + """单用户模式选择：random / cycle / time_slot / smart。""" + registry = get_registry() + modes = config.get("modes") or DEFAULT_MODES + strategy = config.get("refresh_strategy", "random") + + # pending_mode 优先 + pending = config_store.consume_pending_mode(mac) + if pending and registry.is_supported(pending.upper()): + return pending.upper() + + def _supported(cands: list[str]) -> list[str]: + return [c for c in cands if registry.is_supported(c.upper())] + + if strategy == "cycle" and modes: + idx, persona = config_store.advance_cycle(mac, [m.upper() for m in modes]) + return persona.upper() if registry.is_supported(persona.upper()) else (modes[0].upper() if modes else "STOIC") + + if strategy == "time_slot": + hour = datetime.now().hour + for rule in (config.get("time_slot_rules") or []): + sh = rule.get("startHour", 0) + eh = rule.get("endHour", 24) + rm = rule.get("modes", []) + if sh <= hour < eh and rm: + avail = _supported([m.upper() for m in rm if m in modes]) + if avail: + return random.choice(avail) + return random.choice(_supported([m.upper() for m in modes]) or DEFAULT_MODES) + + if strategy == "smart": + hour = datetime.now().hour + for sh, eh, cands in _SMART_TIME_SLOTS: + if sh <= hour < eh: + avail = _supported([c for c in cands if c in modes]) + if avail: + return random.choice(avail) + return random.choice(_supported([m.upper() for m in modes]) or DEFAULT_MODES) + + # random + avail = _supported([m.upper() for m in modes]) or DEFAULT_MODES + persona = random.choice(avail) + # 记录 last_persona（非 cycle 也记，便于观察） + config_store.upsert_state(mac, last_persona=persona, + last_refresh_at=datetime.now().isoformat(timespec="seconds")) + return persona + + +def _inject_llm_key(config: dict[str, Any]) -> dict[str, Any]: + """把解密后的 key/provider 塞进 cfg，走 core 的 caller-supplied 优先路径。 + + 对 openai_compat：动态把当前模型注册到 core.content.LLM_CONFIGS， + 给它一个更稳妥的默认 max_tokens，避免 llm_json 模式在未知模型名时 + 回落到 120 tokens 而被截断。 + """ + cfg = dict(config) + key = secret_store.get_llm_key() + provider = (key.get("provider") or "").strip() + if provider: + cfg["llm_provider"] = provider + if provider == "openai_compat": + if key.get("base_url"): + cfg["llm_base_url"] = key["base_url"] + model = (cfg.get("llm_model") or "").strip() + if model: + try: + from core.content import LLM_CONFIGS + models = LLM_CONFIGS.setdefault("openai_compat", {}).setdefault("models", {}) + models.setdefault(model, {"name": model, "max_tokens": 1024}) + except Exception: + logger.warning("[render] failed to register openai_compat model %s", model, exc_info=True) + if key.get("api_key"): + cfg["user_api_key"] = key["api_key"] + if key.get("image_api_key"): + cfg["user_image_api_key"] = key["image_api_key"] + return cfg + + +async def render_for_device( + mac: str, + *, + battery_voltage: float, + screen_w: int = 400, + screen_h: int = 300, + colors: int = 4, +) -> dict[str, Any]: + """渲染一张图。返回 {image, persona, fallback, refresh_minutes}。 + + fallback=True 表示走了降级内容（LLM 失败等）。 + """ + config = config_store.get_config() + cfg = _inject_llm_key(config) + date_ctx, weather = await build_context(cfg) + persona = _choose_persona(cfg, mac) + battery_pct = calc_battery_pct(battery_voltage) if battery_voltage else 0 + + refresh_min = int(config.get("refresh_interval") or 60) + if refresh_min < 10: + refresh_min = 10 + if refresh_min > 1440: + refresh_min = 1440 + + try: + img, content = await generate_and_render( + persona, + cfg, + date_ctx, + weather, + battery_pct, + screen_w=screen_w, + screen_h=screen_h, + mac=mac, + colors=colors, + ) + fallback = bool(content is None) + except Exception: + logger.exception("[render] generate_and_render failed for persona=%s", persona) + # 极端兜底：空白图 + img = Image.new("P", (screen_w, screen_h), 1) + fallback = True + content = None + + return { + "image": img, + "persona": persona, + "fallback": fallback, + "refresh_minutes": refresh_min, + } + + +async def render_for_preview( + persona: str, + *, + screen_w: int = 400, + screen_h: int = 300, + colors: int = 4, +) -> Image.Image: + """预览：不碰 device_state，不消费 pending_mode。""" + config = config_store.get_config() + cfg = _inject_llm_key(config) + date_ctx, weather = await build_context(cfg) + registry = get_registry() + p = persona.upper() + if not registry.is_supported(p): + p = (config.get("modes") or DEFAULT_MODES)[0].upper() + img, _ = await generate_and_render( + p, cfg, date_ctx, weather, 0, + screen_w=screen_w, screen_h=screen_h, mac="", colors=colors, + ) + return img + + +def encode_image(img: Image.Image, colors: int) -> bytes: + """colors>=3 → raw 2bpp（4 色 e-ink）；否则标准 BMP。""" + if colors >= 3: + return image_to_raw_2bpp(img) + return image_to_bmp_bytes(img) + + +def encode_png(img: Image.Image) -> bytes: + from io import BytesIO + buf = BytesIO() + img.save(buf, format="PNG") + return buf.getvalue() \ No newline at end of file diff --git a/backend-lite/api/__init__.py b/backend-lite/api/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/backend-lite/api/index.py b/backend-lite/api/index.py new file mode 100644 index 00000000..b477d36e --- /dev/null +++ b/backend-lite/api/index.py @@ -0,0 +1,173 @@ +"""backend-lite FastAPI 入口。 + +- 启动时把上游 backend/ 注入 sys.path，使 `from core.xxx import ...` 可用。 +- 初始化 lite.db。 +- 挂载固件路由 + 管理路由 + 静态管理页。 +""" +from __future__ import annotations + +import logging +import os +import sys +from pathlib import Path + +from dotenv import load_dotenv + +# 先加载 .env，让 core/crypto 能读到 LLM_ENCRYPTION_KEY +load_dotenv() + +# 注入上游 backend/ 到 sys.path +_BACKEND_DIR = os.environ.get("INKSIGHT_BACKEND_DIR") or str( + Path(__file__).resolve().parent.parent.parent / "backend" +) +if _BACKEND_DIR not in sys.path: + sys.path.insert(0, _BACKEND_DIR) + +from fastapi import FastAPI +from fastapi.responses import FileResponse +from fastapi.staticfiles import StaticFiles + +from store.db import close_db, init_db + +from . import routes_admin, routes_device, routes_render + +logging.basicConfig( + level=os.environ.get("LITE_LOG_LEVEL", "INFO"), + format="%(asctime)s %(levelname)s %(name)s: %(message)s", +) +logger = logging.getLogger("backend-lite") + + +class _AccessLogFilter(logging.Filter): + """Hide noisy high-frequency polling endpoints from uvicorn.access logs.""" + + def filter(self, record: logging.LogRecord) -> bool: + try: + msg = record.getMessage() + except Exception: + return True + noisy = ( + " /api/admin/state ", + "/api/device/", + "/state HTTP/1.1", + "/alert-bmp?w=400&h=300", + ) + if "uvicorn.access" in record.name and any(token in msg for token in noisy): + # Keep render/config/heartbeat visible; hide only state/alert-bmp polling noise. + if "/alert-bmp" in msg or "/state HTTP/1.1" in msg: + return False + return True + + +_uvicorn_access_logger = logging.getLogger("uvicorn.access") +if not any(isinstance(f, _AccessLogFilter) for f in _uvicorn_access_logger.filters): + _uvicorn_access_logger.addFilter(_AccessLogFilter()) + + +app = FastAPI(title="InkSight backend-lite", version="0.1.0") + + +def _install_native_dither_fallbacks() -> None: + """在缺少 native dithering 动态库时，回退到 Pillow 方案。 + + 这样 MOYU/ARTWALL 这类依赖图片量化的模式不会直接 500。 + 有 native 库时仍优先走上游实现。 + """ + from PIL import Image + from core.config import EINK_4COLOR_PALETTE + import core.native_dither as native_dither + + orig_bw = native_dither.atkinson_bw + orig_palette = native_dither.atkinson_palette + warned = {"bw": False, "palette": False} + + def _bw_fallback(gray: Image.Image) -> Image.Image: + try: + return orig_bw(gray) + except RuntimeError as exc: + if not warned["bw"]: + logger.warning("[startup] native dithering unavailable, fallback to Pillow B/W dithering: %s", exc) + warned["bw"] = True + return gray.convert("1", dither=Image.Dither.FLOYDSTEINBERG) + + def _palette_fallback(rgb: Image.Image, colors: int) -> Image.Image: + try: + return orig_palette(rgb, colors) + except RuntimeError as exc: + if not warned["palette"]: + logger.warning("[startup] native palette dithering unavailable, fallback to Pillow quantize: %s", exc) + warned["palette"] = True + palette_img = Image.new("P", (1, 1)) + palette_img.putpalette(EINK_4COLOR_PALETTE + [0] * (768 - len(EINK_4COLOR_PALETTE))) + fallback_colors = 3 if colors == 3 else 4 + return rgb.convert("RGB").quantize( + palette=palette_img, + dither=Image.Dither.FLOYDSTEINBERG, + colors=fallback_colors, + ) + + native_dither.atkinson_bw = _bw_fallback + native_dither.atkinson_palette = _palette_fallback + + +@app.on_event("startup") +async def _startup() -> None: + # 1) 重定向上游 core DB 路径到 backend-lite 目录（必须在任何 core DB 调用前） + from adaptee.db_redirect import redirect_db_paths, init_upstream_tables + + redirect_db_paths() + # 2) 初始化 backend-lite 自有库 + init_db() + # 2.5) 安装 native dithering 缺失时的运行时降级，避免图片模式直接 500 + _install_native_dither_fallbacks() + # 3) 初始化上游 stats_store / config_store / cache / static_store 表（core 运行时 lazy 依赖） + try: + await init_upstream_tables() + from core.cache import init_cache_db + from core.static_store import init_static_tables, migrate_device_state_columns + await init_cache_db() + await init_static_tables() + await migrate_device_state_columns() + except Exception: + logger.exception("[startup] upstream table init failed") + # 4) 触发上游模式注册表预加载（扫描 builtin/custom 目录） + try: + from core.mode_registry import get_registry + + get_registry() + except Exception: + logger.exception("[startup] mode registry preload failed") + logger.info("[startup] backend-lite ready") + + +@app.on_event("shutdown") +async def _shutdown() -> None: + """关闭 SQLite 连接/aiosqlite 工作线程，避免 Ctrl+C 后 Python 卡在退出阶段。""" + try: + from core.db import close_all + + await close_all() + except Exception: + logger.exception("[shutdown] upstream DB close failed") + try: + close_db() + except Exception: + logger.exception("[shutdown] lite DB close failed") + logger.info("[shutdown] backend-lite connections closed") + + +# 固件路由（/api 前缀） +app.include_router(routes_render.router) +app.include_router(routes_device.router) +# 管理路由（/api/admin） +app.include_router(routes_admin.router) + + +# 静态管理页 +_STATIC_DIR = Path(__file__).resolve().parent.parent / "static" +app.mount("/static", StaticFiles(directory=str(_STATIC_DIR)), name="static") + + +@app.get("/") +async def root(): + return FileResponse(str(_STATIC_DIR / "index.html")) \ No newline at end of file diff --git a/backend-lite/api/routes_admin.py b/backend-lite/api/routes_admin.py new file mode 100644 index 00000000..208bd73c --- /dev/null +++ b/backend-lite/api/routes_admin.py @@ -0,0 +1,217 @@ +"""管理页 API（/api/admin/*）。阶段 1 只放 config/modes/refresh/set-mode； +LLM key / 预览交互 / OTA 在后续阶段补全。局域网免认证。""" +from __future__ import annotations + +from fastapi import APIRouter, Request, Response +from fastapi.responses import JSONResponse + +from store import config_store + +router = APIRouter(prefix="/api/admin", tags=["admin"]) + + +@router.get("/config") +async def admin_get_config(): + return config_store.get_config() + + +@router.put("/config") +async def admin_put_config(request: Request): + try: + body = await request.json() + except Exception: + body = {} + if isinstance(body, dict): + return config_store.update_config(body) + return JSONResponse({"error": "invalid body"}, status_code=400) + + +@router.patch("/config") +async def admin_patch_config(request: Request): + return await admin_put_config(request) + + +@router.get("/modes") +async def admin_list_modes(): + """列出所有已注册模式（builtin + custom），供管理页多选。""" + from core.mode_registry import get_registry + + registry = get_registry() + items = [] + for info in sorted(registry.list_modes(), key=lambda m: m.mode_id): + items.append({ + "mode_id": info.mode_id, + "display_name": info.display_name or info.mode_id, + "source": info.source, + }) + return {"modes": items} + + +@router.post("/refresh") +async def admin_refresh(request: Request): + """触发设备刷新。body 可选 {mac}，默认用最近一次心跳的 mac。""" + try: + body = await request.json() + except Exception: + body = {} + mac = (body.get("mac") or "").strip().upper() or _last_mac() + if not mac: + return JSONResponse({"error": "no device mac known"}, status_code=400) + config_store.upsert_state(mac, pending_refresh=1) + return {"ok": True, "mac": mac} + + +@router.post("/set-mode") +async def admin_set_mode(request: Request): + """设置下次刷新的模式。body {mode, mac?}。""" + try: + body = await request.json() + except Exception: + body = {} + mode = (body.get("mode") or "").strip().upper() + mac = (body.get("mac") or "").strip().upper() or _last_mac() + if not mac or not mode: + return JSONResponse({"error": "need mode and mac"}, status_code=400) + config_store.upsert_state(mac, pending_mode=mode, pending_refresh=1) + return {"ok": True, "mac": mac, "mode": mode} + + +@router.post("/runtime") +async def admin_runtime(request: Request): + """设置运行模式。body {mode: active|interval, mac?}。""" + try: + body = await request.json() + except Exception: + body = {} + mode = (body.get("mode") or "interval").strip() + mac = (body.get("mac") or "").strip().upper() or _last_mac() + if not mac: + return JSONResponse({"error": "no device mac known"}, status_code=400) + config_store.upsert_state(mac, runtime_mode=mode) + return {"ok": True, "mac": mac, "runtime_mode": mode} + + +@router.get("/state") +async def admin_state(): + """管理页展示设备状态。返回最近心跳 mac 的 state。""" + mac = _last_mac() + if not mac: + return {"online": False, "state": None} + return { + "mac": mac, + "online": config_store.is_online(mac), + "state": config_store.get_state(mac), + } + + +@router.get("/alert") +async def admin_get_alert(): + return config_store.get_alert() + + +@router.put("/alert") +@router.post("/alert") +async def admin_set_alert(request: Request): + try: + body = await request.json() + except Exception: + body = {} + active = bool(body.get("active", False)) + text = (body.get("text") or "")[:64] + config_store.set_alert(active, text) + return config_store.get_alert() + + +def _last_mac() -> str: + """取最近一次心跳的 mac。""" + from store.db import get_conn, _lock + with _lock: + row = get_conn().execute( + "SELECT mac FROM heartbeats ORDER BY ts DESC LIMIT 1" + ).fetchone() + return row["mac"] if row else "" + +# ---------------- LLM key（阶段 2） ---------------- +from store import secret_store + +@router.get("/llm-key") +async def admin_get_llm_key(): + data = secret_store.get_llm_key_masked() + cfg = config_store.get_config() + if cfg.get("llm_provider"): + data["provider"] = cfg.get("llm_provider") + data["model"] = cfg.get("llm_model") or "" + return data + +@router.put("/llm-key") +async def admin_put_llm_key(request: Request): + try: + body = await request.json() + except Exception: + body = {} + if not isinstance(body, dict): + return JSONResponse({"error":"invalid body"}, status_code=400) + + secret_store.set_llm_key(body) + + cfg_patch = {} + if "provider" in body: + cfg_patch["llm_provider"] = body.get("provider") or "" + if "model" in body: + cfg_patch["llm_model"] = body.get("model") or "" + if cfg_patch: + config_store.update_config(cfg_patch) + + data = secret_store.get_llm_key_masked() + cfg = config_store.get_config() + if cfg.get("llm_provider"): + data["provider"] = cfg.get("llm_provider") + data["model"] = cfg.get("llm_model") or "" + return data + + +# ---------------- OTA（阶段 4） ---------------- +import os, uuid, json +_OTA_DIR = os.path.join(os.path.dirname(os.path.dirname(__file__)), "ota_files") +os.makedirs(_OTA_DIR, exist_ok=True) + +@router.post("/ota/upload") +async def ota_upload(request: Request): + """接收 multipart 上传 bin，写 ota_files/，返回设备可达 URL。""" + form = await request.form() + upload = form.get("file") + if upload is None or not hasattr(upload, "filename"): + return JSONResponse({"error":"no file"}, status_code=400) + fname = f"{uuid.uuid4().hex[:8]}_{upload.filename}" + path = os.path.join(_OTA_DIR, fname) + with open(path, "wb") as f: + f.write(await upload.read()) + host = request.headers.get("Host") or request.url.hostname or "localhost" + url = f"http://{host}/api/admin/ota/file/{fname}" + mac = _last_mac() + if mac: + config_store.upsert_state(mac, ota_url=url, ota_version=upload.filename or "") + return {"url": url, "mac": mac} + +@router.post("/ota/set") +async def ota_set(request: Request): + try: + body = await request.json() + except Exception: + body = {} + url = (body.get("ota_url") or "").strip() + ver = (body.get("ota_version") or "").strip() + mac = (body.get("mac") or "").strip().upper() or _last_mac() + if not mac: + return JSONResponse({"error":"no device mac known"}, status_code=400) + config_store.upsert_state(mac, ota_url=url, ota_version=ver) + return {"ok": True, "mac": mac, "ota_url": url} + +@router.get("/ota/file/{name}") +async def ota_file(name: str): + """设备从此拉取 bin。""" + from fastapi.responses import FileResponse + path = os.path.join(_OTA_DIR, name) + if not os.path.isfile(path) or ".." in name or "/" in name: + return Response(status_code=404) + return FileResponse(path, media_type="application/octet-stream", filename=name) diff --git a/backend-lite/api/routes_device.py b/backend-lite/api/routes_device.py new file mode 100644 index 00000000..db2660ed --- /dev/null +++ b/backend-lite/api/routes_device.py @@ -0,0 +1,238 @@ +"""固件设备路由：token / heartbeat / config / state / runtime / refresh / +claim-token / alert-bmp / ota-progress。单用户：MAC 任意，token 固定。""" +from __future__ import annotations + +import json +import logging +import secrets + +from fastapi import APIRouter, Header, Request, Response +from fastapi.responses import JSONResponse, Response as RawResponse + +from store import config_store + +logger = logging.getLogger("backend-lite.device") + +router = APIRouter(prefix="/api", tags=["device"]) + +# 单用户：所有 MAC 共享一个固定 token（首次发放后存 device_state）。 +# 不校验 X-Device-Token 内容——局域网信任模型。 +DEVICE_TOKEN = "inklite-single-user-token" + + +def _mac(request: Request, mac: str | None = None) -> str: + return (mac or request.path_params.get("mac") or "").strip().upper() or "00:00:00:00:00:00:00" + + +@router.post("/device/{mac}/token") +async def issue_token(mac: str): + """发放设备 token。单用户固定值，但回写 device_state 以备追溯。""" + m = mac.strip().upper() + config_store.upsert_state(m, auth_token=DEVICE_TOKEN) + return {"token": DEVICE_TOKEN} + + +@router.post("/device/{mac}/claim-token") +async def claim_token(mac: str, request: Request): + """回显 pair_code（portal 配对兼容）。""" + try: + body = await request.json() + except Exception: + body = {} + return {"pair_code": body.get("pair_code", "")} + + +@router.post("/device/{mac}/heartbeat") +async def heartbeat(mac: str, request: Request): + try: + body = await request.json() + except Exception: + body = {} + bv = float(body.get("battery_voltage") or 0) + rssi = int(body.get("wifi_rssi") or 0) + config_store.add_heartbeat(mac.strip().upper(), bv, rssi) + return Response(status_code=200) + + +@router.get("/config/{mac}") +async def get_device_config(mac: str): + """固件轮询：返回 focus / always_active 开关。""" + cfg = config_store.get_config() + return { + "is_focus_listening": bool(cfg.get("focus_listening")), + "is_always_active": bool(cfg.get("always_active")), + } + + +@router.get("/device/{mac}/state") +async def get_state(mac: str): + """固件轮询运行状态。OTA 字段非空时附上。""" + s = config_store.get_state(mac.strip().upper()) + resp = { + "pending_refresh": int(bool(s.get("pending_refresh"))), + "pending_mode": s.get("pending_mode") or "", + "runtime_mode": s.get("runtime_mode") or "interval", + } + ota_url = s.get("ota_url") or "" + ota_ver = s.get("ota_version") or "" + if ota_url: + resp["ota_url"] = ota_url + resp["ota_version"] = ota_ver + return resp + + +@router.post("/device/{mac}/runtime") +async def set_runtime(mac: str, request: Request): + try: + body = await request.json() + except Exception: + body = {} + mode = (body.get("mode") or "interval").strip() + config_store.upsert_state(mac.strip().upper(), runtime_mode=mode) + return Response(status_code=200) + + +@router.post("/device/{mac}/refresh") +async def trigger_refresh(mac: str): + config_store.upsert_state(mac.strip().upper(), pending_refresh=1) + return Response(status_code=200) + + +@router.post("/config") +async def save_device_config(request: Request): + """固件 POST /api/config：body 是设备的 cfgConfigJson（含 mac）。 + 单用户：把可识别字段写回 config 单行。""" + try: + body = await request.json() + except Exception: + body = {} + if not isinstance(body, dict): + body = {} + patch: dict = {} + for k in ( + "modes", "refresh_strategy", "refresh_interval", "city", + "language", "content_tone", "time_slot_rules", "countdown_events", + "memo_text", "focus_listening", "always_active", + "llm_provider", "llm_model", "image_provider", "image_model", + ): + if k in body: + patch[k] = body[k] + if patch: + config_store.update_config(patch) + return Response(status_code=200) + + +@router.get("/device/{mac}/alert-bmp") +async def alert_bmp(mac: str, w: int = 400, h: int = 300): + """焦点提醒图。无活跃 alert → 204；有 → 渲染更醒目的居中文字 1-bit BMP。""" + alert = config_store.get_alert() + if not alert.get("active"): + return Response(status_code=204) + + from io import BytesIO + from PIL import Image, ImageDraw + from core.patterns.utils import load_font + + width = w or 400 + height = h or 300 + text = (alert.get("text") or "!").strip() or "!" + img = Image.new("1", (width, height), 1) + draw = ImageDraw.Draw(img) + + # 外框，增强“弹窗感” + draw.rounded_rectangle([8, 8, width - 9, height - 9], radius=16, outline=0, width=2) + + max_width = width - 56 + max_height = height - 56 + + def wrap_text(font): + lines = [] + current = "" + for ch in text: + if ch == "\n": + lines.append(current or " ") + current = "" + continue + candidate = current + ch + bbox = draw.textbbox((0, 0), candidate, font=font) + cand_width = bbox[2] - bbox[0] + if current and cand_width > max_width: + lines.append(current) + current = ch + else: + current = candidate + if current: + lines.append(current) + return lines or ["!"] + + chosen_font = None + chosen_lines = [text] + chosen_metrics = [] + for size in (72, 64, 56, 48, 42, 36, 32, 28, 24): + try: + font = load_font("noto_serif_bold", size) + except Exception: + font = None + lines = wrap_text(font) + metrics = [] + widest = 0 + total_h = 0 + spacing = max(6, size // 5) + for line in lines: + bbox = draw.textbbox((0, 0), line, font=font) + left, top, right, bottom = bbox + line_w = max(1, right - left) + line_h = max(1, bottom - top) + widest = max(widest, line_w) + total_h += line_h + metrics.append((line, left, top, line_w, line_h)) + total_h += spacing * max(0, len(metrics) - 1) + if widest <= max_width and total_h <= max_height: + chosen_font = font + chosen_lines = lines + chosen_metrics = metrics + break + chosen_font = font + chosen_lines = lines + chosen_metrics = metrics + + total_height = sum(item[4] for item in chosen_metrics) + total_height += max(6, 24 // 5) * max(0, len(chosen_metrics) - 1) + # 根据最终字号重新计算行距 + if chosen_metrics: + avg_height = max(item[4] for item in chosen_metrics) + line_spacing = max(6, avg_height // 4) + total_height = sum(item[4] for item in chosen_metrics) + line_spacing * max(0, len(chosen_metrics) - 1) + else: + line_spacing = 8 + + y = max(18, (height - total_height) // 2) + for idx, (line, left, top, line_w, line_h) in enumerate(chosen_metrics): + x = max(16, (width - line_w) // 2) + draw.text((x - left, y - top), line, fill=0, font=chosen_font) + y += line_h + (line_spacing if idx < len(chosen_metrics) - 1 else 0) + + buf = BytesIO() + img.save(buf, format="BMP") + return RawResponse(content=buf.getvalue(), media_type="image/bmp") + + +@router.post("/device/{mac}/ota/progress") +async def ota_progress(mac: str, request: Request): + try: + body = await request.json() + except Exception: + body = {} + logger.info("[ota] progress %s: %s", mac, body) + return Response(status_code=200) + + +# vocab / voice 不实现：固件遇非 200 自动降级 +@router.api_route("/device/{mac}/vocab/{rest:path}", methods=["GET", "POST"]) +async def vocab_unsupported(rest: str): + return Response(status_code=404) + + +@router.api_route("/device/{mac}/voice/{rest:path}", methods=["GET", "POST"]) +async def voice_unsupported(rest: str): + return Response(status_code=404) \ No newline at end of file diff --git a/backend-lite/api/routes_render.py b/backend-lite/api/routes_render.py new file mode 100644 index 00000000..f1e90ba1 --- /dev/null +++ b/backend-lite/api/routes_render.py @@ -0,0 +1,70 @@ +"""固件渲染路由：/api/render, /api/preview。""" +from __future__ import annotations + +import logging + +from fastapi import APIRouter, Query, Request, Response +from fastapi.responses import Response as RawResponse + +from adaptee.render import encode_image, encode_png, render_for_device, render_for_preview +from store import config_store + +logger = logging.getLogger("backend-lite.render") + +router = APIRouter(prefix="/api", tags=["render"]) + +# 固件协议默认尺寸 400×300、2bpp、4 色 +SCREEN_W = 400 +SCREEN_H = 300 +COLORS = 4 + + +@router.get("/render") +async def render( + request: Request, + mac: str = Query(default=""), + v: float = Query(default=0.0), + rssi: int = Query(default=0), + refresh_min: int = Query(default=0), + w: int = Query(default=SCREEN_W), + h: int = Query(default=SCREEN_H), + bpp: int = Query(default=2), + colors: int = Query(default=COLORS), + next: int = Query(default=0, alias="next"), +): + """固件核心刷图端点。返回 raw 2bpp（4 色）或 BMP。""" + mac_clean = mac.strip().upper() or "00:00:00:00:00:00" + result = await render_for_device( + mac_clean, + battery_voltage=v, + screen_w=w or SCREEN_W, + screen_h=h or SCREEN_H, + colors=colors if colors >= 2 else COLORS, + ) + # 渲染成功后清除 pending_refresh + config_store.upsert_state(mac_clean, pending_refresh=0) + + body = encode_image(result["image"], result["colors"] if "colors" in result else colors) + headers = { + "X-Mode-Id": result["persona"], + "X-Refresh-Minutes": str(result["refresh_minutes"]), + "Cache-Control": "no-store", + "Connection": "close", + } + if result["fallback"]: + headers["X-Content-Fallback"] = "1" + return RawResponse(content=body, media_type="application/octet-stream", headers=headers) + + +@router.get("/preview") +async def preview( + mode: str = Query(default=""), + colors: int = Query(default=COLORS), + as_png: int = Query(default=0), +): + """管理页预览：固定 400×300/4 色。默认 PNG 便于浏览器内联显示。""" + persona = (mode or "").strip().upper() or "STOIC" + img = await render_for_preview(persona, colors=colors if colors >= 2 else COLORS) + if as_png: + return RawResponse(content=encode_png(img), media_type="image/png") + return RawResponse(content=encode_image(img, colors), media_type="application/octet-stream") \ No newline at end of file diff --git a/backend-lite/imgs/1.png b/backend-lite/imgs/1.png new file mode 100644 index 00000000..85251e18 Binary files /dev/null and b/backend-lite/imgs/1.png differ diff --git a/backend-lite/imgs/2.png b/backend-lite/imgs/2.png new file mode 100644 index 00000000..8945d2a0 Binary files /dev/null and b/backend-lite/imgs/2.png differ diff --git a/backend-lite/imgs/3.png b/backend-lite/imgs/3.png new file mode 100644 index 00000000..14c10d85 Binary files /dev/null and b/backend-lite/imgs/3.png differ diff --git a/backend-lite/requirements.txt b/backend-lite/requirements.txt new file mode 100644 index 00000000..61e485ae --- /dev/null +++ b/backend-lite/requirements.txt @@ -0,0 +1,10 @@ +fastapi>=0.115 +uvicorn[standard]>=0.30 +Pillow>=10.0 +httpx>=0.27 +openai>=1.40 +python-dotenv>=1.0 +python-multipart>=0.0.9 +cryptography>=42.0 +tenacity>=8.2 +zhdate>=0.1 \ No newline at end of file diff --git a/backend-lite/run.py b/backend-lite/run.py new file mode 100644 index 00000000..ec1a93ae --- /dev/null +++ b/backend-lite/run.py @@ -0,0 +1,31 @@ +"""backend-lite 启动入口。 + +直接读取 .env 中的 LITE_HOST / LITE_PORT，再调用 uvicorn.run()。 +这解决了直接使用 `uvicorn ... --host ... --port ...` 时命令行参数覆盖 .env 的问题。 +""" +from __future__ import annotations + +import os + +from dotenv import load_dotenv +import uvicorn + + +load_dotenv() + + +def _env_int(name: str, default: int) -> int: + raw = os.environ.get(name, "").strip() + if not raw: + return default + try: + return int(raw) + except ValueError: + return default + + +if __name__ == "__main__": + host = os.environ.get("LITE_HOST", "0.0.0.0").strip() or "0.0.0.0" + port = _env_int("LITE_PORT", 8090) + reload = os.environ.get("LITE_RELOAD", "").strip().lower() in {"1", "true", "yes", "on"} + uvicorn.run("api.index:app", host=host, port=port, reload=reload) diff --git a/backend-lite/static/app.js b/backend-lite/static/app.js new file mode 100644 index 00000000..13b52247 --- /dev/null +++ b/backend-lite/static/app.js @@ -0,0 +1,247 @@ +// backend-lite 管理页逻辑。原生 JS，无构建。 +const $ = (id) => document.getElementById(id); +const API = "/api/admin"; +let selectedModes = new Set(); + +function flash(el, msg, ok=true){ + if(!el) return; + el.textContent = msg; + el.classList.remove("is-ok", "is-error", "ok", "err"); + el.classList.add(ok ? "is-ok" : "is-error"); + setTimeout(()=>{ el.textContent=""; el.classList.remove("is-ok", "is-error"); }, 2500); +} + +function escapeHtml(value){ + return String(value ?? "").replace(/[&<>\"']/g, (char) => ({ + "&":"&", "<":"<", ">":">", "\"":""", "'":"'" + }[char])); +} + +function stateValue(value, fallback="—"){ + const text = value === undefined || value === null || value === "" ? fallback : value; + return escapeHtml(text); +} + +async function jget(url){ + const r = await fetch(url); + if(!r.ok) throw new Error(`${url} ${r.status}`); + return r.json(); +} +async function jpost(url, body, method="POST"){ + const r = await fetch(url, {method, headers:{"Content-Type":"application/json"}, body: JSON.stringify(body||{})}); + return r; +} + +async function loadStatus(){ + try{ + const s = await jget(`${API}/state`); + const st = s.state || {}; + const status = $("status"); + const deviceState = $("device-state"); + + if(!s.mac){ + status.innerHTML = `
尚无设备连接过，等待设备首次心跳。
`; + if(deviceState){ + deviceState.innerHTML = `
设备尚未上报状态。
`; + } + return; + } + + const online = !!s.online; + const onlineText = online ? "在线" : "离线"; + const statusTitle = online ? "InkSight 设备在线" : "最近连接过的设备"; + const mode = st.last_persona || "未记录"; + const runtime = st.runtime_mode || "interval"; + const pendingMode = st.pending_mode || "无"; + const refreshedAt = st.last_refresh_at || "未记录"; + + status.innerHTML = ` +
+ ${onlineText} + ${statusTitle} + ${stateValue(s.mac)} +
+
+ 运行 ${stateValue(runtime)} + 当前模式 ${stateValue(mode)} + 待刷新 ${stateValue(st.pending_refresh || 0)} + 待切换 ${stateValue(pendingMode)} +
`; + + if(deviceState){ + deviceState.innerHTML = ` +
+
+ 设备 MAC + ${stateValue(s.mac)} +
+
+ 连接状态 + ${onlineText} +
+
+ 运行模式 + ${stateValue(runtime)} +
+
+ 当前内容 + ${stateValue(mode)} +
+
+ 待刷新 / 待切换 + ${stateValue(st.pending_refresh || 0)} / ${stateValue(pendingMode)} +
+
+ 最近刷新 + ${stateValue(refreshedAt)} +
+
`; + } + }catch(e){ + $("status").innerHTML = `
状态读取失败：${escapeHtml(e.message)}
`; + const deviceState = $("device-state"); + if(deviceState){ + deviceState.innerHTML = `
暂时无法读取设备状态。
`; + } + } +} + +async function loadModes(){ + const data = await jget(`${API}/modes`); + const chips = $("modes"); + chips.innerHTML = ""; + for(const m of data.modes){ + const c = document.createElement("div"); + c.className = "chip"; + c.textContent = m.display_name || m.mode_id; + c.dataset.id = m.mode_id; + c.onclick = ()=>{ c.classList.toggle("on"); selectedModes.has(m.mode_id)?selectedModes.delete(m.mode_id):selectedModes.add(m.mode_id); }; + chips.appendChild(c); + } + // 同时填预览下拉 + const sel = $("prev-mode"); + sel.innerHTML = ""; + for(const m of data.modes){ const o=document.createElement("option"); o.value=m.mode_id; o.textContent=m.display_name||m.mode_id; sel.appendChild(o); } +} + +async function loadConfig(){ + const cfg = await jget(`${API}/config`); + selectedModes = new Set(cfg.modes||[]); + document.querySelectorAll("#modes .chip").forEach(c=>{ if(selectedModes.has(c.dataset.id)) c.classList.add("on"); }); + $("cfg-strategy").value = cfg.refresh_strategy||"random"; + $("cfg-interval").value = cfg.refresh_interval||60; + $("cfg-city").value = cfg.city||""; + $("cfg-language").value = cfg.language||"zh"; + $("cfg-tone").value = cfg.content_tone||"neutral"; + $("cfg-memo").value = cfg.memo_text||""; + $("cfg-focus").checked = !!cfg.focus_listening; + $("cfg-active").checked = !!cfg.always_active; + const llmModel = $("llm-model"); + if(llmModel) llmModel.value = cfg.llm_model || ""; +} + +async function saveConfig(){ + const patch = { + modes: [...selectedModes], + refresh_strategy: $("cfg-strategy").value, + refresh_interval: parseInt($("cfg-interval").value||60), + city: $("cfg-city").value, + language: $("cfg-language").value, + content_tone: $("cfg-tone").value, + memo_text: $("cfg-memo").value, + focus_listening: $("cfg-focus").checked?1:0, + always_active: $("cfg-active").checked?1:0, + }; + const r = await jpost(`${API}/config`, patch, "PUT"); + flash($("cfg-status"), r.ok?"已保存":"失败", r.ok); +} + +async function loadKey(){ + const k = await jget(`${API}/llm-key`); + $("llm-provider").value = k.provider||"deepseek"; + $("llm-baseurl").value = k.base_url||""; + const llmModel = $("llm-model"); + if(llmModel) llmModel.value = k.model || ""; + $("llm-key").placeholder = k.has_api_key ? "（已保存，留空保留）" : ""; + $("llm-imgkey").placeholder = k.has_image_api_key ? "（已保存，留空保留）" : ""; +} + +async function saveKey(){ + const provider = $("llm-provider").value; + const baseUrl = $("llm-baseurl").value.trim(); + const model = $("llm-model")?.value.trim() || ""; + + if(provider === "openai_compat"){ + if(!baseUrl){ + flash($("key-status"), "openai_compat 需要填写 Base URL", false); + return; + } + if(!model){ + flash($("key-status"), "openai_compat 需要填写模型名", false); + return; + } + } + + const patch = { provider, base_url: baseUrl, model }; + if($("llm-key").value) patch.api_key = $("llm-key").value; + if($("llm-imgkey").value) patch.image_api_key = $("llm-imgkey").value; + + const r = await jpost(`${API}/llm-key`, patch, "PUT"); + flash($("key-status"), r.ok ? "已保存" : "保存失败", r.ok); + if(r.ok){ + $("llm-key").value = ""; + $("llm-imgkey").value = ""; + loadKey(); + loadConfig(); + } +} + +async function doPreview(){ + $("preview").src = `/api/preview?mode=${encodeURIComponent($("prev-mode").value)}&as_png=1&_=${Date.now()}`; +} + +async function doRefresh(){ + const r = await jpost(`${API}/refresh`, {}); + flash($("op-status"), r.ok?"已触发":"失败", r.ok); + setTimeout(loadStatus, 500); +} +async function exitActive(){ + const r = await jpost(`${API}/runtime`, {mode:"interval"}); + flash($("op-status"), r.ok?"已退出活跃":"失败", r.ok); + setTimeout(loadStatus, 500); +} + +async function loadAlert(){ + try{ + const a = await jget(`${API}/alert`); + $("alert-active").checked = !!a.active; + $("alert-text").value = a.text || ""; + }catch(e){} +} +async function saveAlert(){ + const r = await jpost(`${API}/alert`, {active:$("alert-active").checked, text:$("alert-text").value}, "PUT"); + flash($("alert-status"), r.ok?"已保存":"失败", r.ok); +} + +async function otaSet(){ + const fileInput = $("ota-file"); + if(fileInput.files.length){ + const fd = new FormData(); + fd.append("file", fileInput.files[0]); + const r = await fetch(`${API}/ota/upload`, {method:"POST", body:fd}); + const data = await r.json().catch(()=>({})); + flash($("ota-status"), r.ok?`已下发: ${data.url||''}`:`失败 ${r.status}`, r.ok); + return; + } + const url = $("ota-url").value.trim(); + if(!url){ flash($("ota-status"),"请选择文件或填 URL",false); return; } + const r = await jpost(`${API}/ota/set`, {ota_url:url, ota_version:$("ota-version").value||""}); + flash($("ota-status"), r.ok?"已下发":"失败", r.ok); +} + +(async function init(){ + await loadModes(); + await Promise.all([loadConfig(), loadKey(), loadAlert(), loadStatus()]); + doPreview(); + setInterval(loadStatus, 10000); +})(); \ No newline at end of file diff --git a/backend-lite/static/index.html b/backend-lite/static/index.html new file mode 100644 index 00000000..5c0a4357 --- /dev/null +++ b/backend-lite/static/index.html @@ -0,0 +1,854 @@ + + + + +InkSight · backend-lite + + + + +
+
+
+

InkSight / Local Control Plane

+

backend-lite 单设备管理

+
+
局域网免认证 · 400×300 · 4 色
+
+ +
+
正在读取设备状态…
+ +
+ +
+
+
+
+
+

01 / Provider

+

LLM 设置

+

API Key 会加密保存到本地数据库，仅用于内容生成。

+
+
AI
+
+
+
+
+ + +
+
+ + +
+
+
+
+ + +
+
+ + +
+
+
+
+ + +
+
+
+
+ + +
+
+ +
+
+
+

02 / Device Profile

+

设备配置

+

选择屏幕上轮换的内容，以及设备的更新节奏。

+
+
设
+
+
+
+
+

内容模式

+ 点击启用 / 取消 +
+
+
+ +
+
+

刷新与位置

+ 设备下一次取图时生效 +
+
+
+ + +
+
+ + +
+
+ + +
+
+ + +
+
+
+ +
+
+

内容偏好

+ 影响生成语气与上下文 +
+
+
+ + +
+
+ + +
+
+
+ +
+
+

运行开关

+ 影响设备在线行为 +
+
+ + +
+
+
+
+ + +
+
+ +
+
+
+

03 / Focus

+

焦点提醒

+

专注监听开启后，设备会定期拉取提醒图。

+
+
醒
+
+
+ +
+ + +
+
提醒图会覆盖当前屏幕内容，关闭后设备恢复正常内容。
+
+
+ + +
+
+ +
+
+
+

04 / Firmware

+

OTA 固件更新

+

上传固件或提供设备可访问的下载地址。

+
+
固
+
+
+
+
+ + +
+
+ + +
+
+
+ + +
+
+
+ + +
+
+
+ + +
+
+ + + \ No newline at end of file diff --git a/backend-lite/store/__init__.py b/backend-lite/store/__init__.py new file mode 100644 index 00000000..a2dcfa2a --- /dev/null +++ b/backend-lite/store/__init__.py @@ -0,0 +1 @@ +"""backend-lite store 包标记。""" \ No newline at end of file diff --git a/backend-lite/store/config_store.py b/backend-lite/store/config_store.py new file mode 100644 index 00000000..29c455fd --- /dev/null +++ b/backend-lite/store/config_store.py @@ -0,0 +1,194 @@ +"""单设备配置 + 设备运行状态 CRUD。""" +from __future__ import annotations + +import json +from datetime import datetime +from typing import Any + +from .db import get_conn, _lock + +# ---------------- config（单行） ---------------- + +_CONFIG_INT_FIELDS = {"refresh_interval", "focus_listening", "always_active"} +_CONFIG_JSON_FIELDS = {"time_slot_rules", "countdown_events", "mode_overrides"} +_CONFIG_FLOAT_FIELDS = {"latitude", "longitude"} + + +def get_config() -> dict[str, Any]: + with _lock: + row = get_conn().execute("SELECT * FROM config WHERE id=1").fetchone() + if row is None: + return {} + cfg = dict(row) + for k in _CONFIG_JSON_FIELDS: + try: + cfg[k] = json.loads(cfg.get(k) or "[]") if k != "mode_overrides" else json.loads(cfg.get(k) or "{}") + except (ValueError, TypeError): + cfg[k] = [] if k != "mode_overrides" else {} + # modes 存 CSV + cfg["modes"] = [m.strip() for m in (cfg.get("modes") or "").split(",") if m.strip()] + return cfg + + +def update_config(patch: dict[str, Any]) -> dict[str, Any]: + """部分更新 config 单行。未知字段忽略。""" + allowed = { + "modes", "refresh_strategy", "refresh_interval", "city", + "latitude", "longitude", "language", "content_tone", + "time_slot_rules", "countdown_events", "memo_text", + "focus_listening", "always_active", + "llm_provider", "llm_model", "image_provider", "image_model", + "mode_overrides", + } + sets: list[str] = [] + vals: list[Any] = [] + for k, v in patch.items(): + if k not in allowed: + continue + if k in _CONFIG_INT_FIELDS and not isinstance(v, int): + v = int(bool(v)) + if k == "modes" and isinstance(v, list): + v = ",".join(m.strip() for m in v if str(m).strip()) + if k in _CONFIG_JSON_FIELDS: + v = json.dumps(v, ensure_ascii=False) + if k in _CONFIG_FLOAT_FIELDS and v is not None: + v = float(v) + sets.append(f"{k}=?") + vals.append(v) + if not sets: + return get_config() + with _lock: + conn = get_conn() + conn.execute(f"UPDATE config SET {', '.join(sets)} WHERE id=1", vals) + conn.commit() + return get_config() + + +# ---------------- device_state（按 mac，单用户实际只有一行） ---------------- + +def get_state(mac: str) -> dict[str, Any]: + with _lock: + row = get_conn().execute( + "SELECT * FROM device_state WHERE mac=?", (mac,) + ).fetchone() + if row is None: + return { + "mac": mac, "pending_refresh": 0, "pending_mode": "", + "runtime_mode": "interval", "cycle_index": 0, "last_persona": "", + "last_refresh_at": "", "auth_token": "", "ota_url": "", "ota_version": "", + } + return dict(row) + + +def upsert_state(mac: str, **patch: Any) -> dict[str, Any]: + allowed = { + "pending_refresh", "pending_mode", "runtime_mode", "cycle_index", + "last_persona", "last_refresh_at", "auth_token", "ota_url", "ota_version", + } + cur = get_state(mac) + cur.update({k: v for k, v in patch.items() if k in allowed}) + with _lock: + conn = get_conn() + conn.execute( + """INSERT INTO device_state (mac, pending_refresh, pending_mode, runtime_mode, + cycle_index, last_persona, last_refresh_at, auth_token, ota_url, ota_version) + VALUES (?,?,?,?,?,?,?,?,?,?) + ON CONFLICT(mac) DO UPDATE SET + pending_refresh=excluded.pending_refresh, + pending_mode=excluded.pending_mode, + runtime_mode=excluded.runtime_mode, + cycle_index=excluded.cycle_index, + last_persona=excluded.last_persona, + last_refresh_at=excluded.last_refresh_at, + auth_token=excluded.auth_token, + ota_url=excluded.ota_url, + ota_version=excluded.ota_version""", + (mac, cur["pending_refresh"], cur["pending_mode"], cur["runtime_mode"], + cur["cycle_index"], cur["last_persona"], cur["last_refresh_at"], + cur["auth_token"], cur["ota_url"], cur["ota_version"]), + ) + conn.commit() + return cur + + +def consume_pending_refresh(mac: str) -> bool: + """读取并清除 pending_refresh。""" + s = get_state(mac) + if s.get("pending_refresh"): + upsert_state(mac, pending_refresh=0) + return True + return False + + +def consume_pending_mode(mac: str) -> str | None: + """读取并清除 pending_mode。""" + s = get_state(mac) + pm = s.get("pending_mode") or "" + if pm: + upsert_state(mac, pending_mode="") + return pm + return None + + +def advance_cycle(mac: str, modes: list[str]) -> tuple[int, str]: + """推进 cycle_index，返回 (新 idx, 选中 persona)。""" + if not modes: + return 0, "STOIC" + s = get_state(mac) + idx = int(s.get("cycle_index") or 0) + persona = modes[idx % len(modes)] + upsert_state(mac, cycle_index=idx + 1, last_persona=persona, + last_refresh_at=datetime.now().isoformat(timespec="seconds")) + return idx, persona + + +# ---------------- heartbeats ---------------- + +def add_heartbeat(mac: str, battery_voltage: float, wifi_rssi: int) -> None: + ts = datetime.now().isoformat(timespec="seconds") + with _lock: + conn = get_conn() + conn.execute( + "INSERT OR REPLACE INTO heartbeats (mac, ts, battery_voltage, wifi_rssi) VALUES (?,?,?,?)", + (mac, ts, battery_voltage, wifi_rssi), + ) + # 滚动保留最近 100 条/设备 + conn.execute( + """DELETE FROM heartbeats WHERE mac=? AND ts NOT IN ( + SELECT ts FROM heartbeats WHERE mac=? ORDER BY ts DESC LIMIT 100)""", + (mac, mac), + ) + conn.commit() + + +def is_online(mac: str, within_minutes: int = 15) -> bool: + with _lock: + row = get_conn().execute( + "SELECT MAX(ts) AS latest FROM heartbeats WHERE mac=?", (mac,) + ).fetchone() + if not row or not row["latest"]: + return False + try: + latest = datetime.fromisoformat(row["latest"]) + except ValueError: + return False + return (datetime.now() - latest).total_seconds() < within_minutes * 60 + + +# ---------------- alert_state（单行） ---------------- + +def get_alert() -> dict[str, Any]: + with _lock: + row = get_conn().execute("SELECT * FROM alert_state WHERE id=1").fetchone() + return dict(row) if row else {"active": 0, "text": ""} + + +def set_alert(active: bool, text: str = "") -> dict[str, Any]: + with _lock: + conn = get_conn() + conn.execute( + "UPDATE alert_state SET active=?, text=? WHERE id=1", + (int(bool(active)), text), + ) + conn.commit() + return get_alert() \ No newline at end of file diff --git a/backend-lite/store/db.py b/backend-lite/store/db.py new file mode 100644 index 00000000..42905fc8 --- /dev/null +++ b/backend-lite/store/db.py @@ -0,0 +1,110 @@ +"""SQLite 连接与表初始化（lite.db）。 + +单用户、单设备：所有配置以单行形式存储。 +""" +from __future__ import annotations + +import os +import sqlite3 +import threading +from pathlib import Path + +_DB_PATH = os.environ.get( + "LITE_DB_PATH", + str(Path(__file__).resolve().parent.parent / "lite.db"), +) + +_lock = threading.Lock() +_conn: sqlite3.Connection | None = None + + +def get_conn() -> sqlite3.Connection: + """返回进程级单例连接（check_same_thread=False，由 _lock 串行化写入）。""" + global _conn + if _conn is None: + conn = sqlite3.connect(_DB_PATH, check_same_thread=False) + conn.row_factory = sqlite3.Row + conn.execute("PRAGMA journal_mode=WAL") + conn.execute("PRAGMA foreign_keys=ON") + _conn = conn + return _conn + + +SCHEMA = """ +CREATE TABLE IF NOT EXISTS config ( + id INTEGER PRIMARY KEY CHECK (id = 1), + modes TEXT DEFAULT 'STOIC,ROAST,ZEN,DAILY', + refresh_strategy TEXT DEFAULT 'random', + refresh_interval INTEGER DEFAULT 60, + city TEXT DEFAULT '杭州', + latitude REAL, + longitude REAL, + language TEXT DEFAULT 'zh', + content_tone TEXT DEFAULT 'neutral', + time_slot_rules TEXT DEFAULT '[]', + countdown_events TEXT DEFAULT '[]', + memo_text TEXT DEFAULT '', + focus_listening INTEGER DEFAULT 0, + always_active INTEGER DEFAULT 0, + llm_provider TEXT DEFAULT 'deepseek', + llm_model TEXT DEFAULT '', + image_provider TEXT DEFAULT 'aliyun', + image_model TEXT DEFAULT '', + mode_overrides TEXT DEFAULT '{}' +); + +CREATE TABLE IF NOT EXISTS device_state ( + mac TEXT PRIMARY KEY, + pending_refresh INTEGER DEFAULT 0, + pending_mode TEXT DEFAULT '', + runtime_mode TEXT DEFAULT 'interval', + cycle_index INTEGER DEFAULT 0, + last_persona TEXT DEFAULT '', + last_refresh_at TEXT DEFAULT '', + auth_token TEXT DEFAULT '', + ota_url TEXT DEFAULT '', + ota_version TEXT DEFAULT '' +); + +CREATE TABLE IF NOT EXISTS llm_key ( + id INTEGER PRIMARY KEY CHECK (id = 1), + provider TEXT DEFAULT 'deepseek', + api_key TEXT DEFAULT '', + base_url TEXT DEFAULT '', + image_api_key TEXT DEFAULT '' +); + +CREATE TABLE IF NOT EXISTS heartbeats ( + mac TEXT, + ts TEXT, + battery_voltage REAL, + wifi_rssi INTEGER, + PRIMARY KEY (mac, ts) +); + +CREATE TABLE IF NOT EXISTS alert_state ( + id INTEGER PRIMARY KEY CHECK (id = 1), + active INTEGER DEFAULT 0, + text TEXT DEFAULT '' +); +""" + + +def init_db() -> None: + with _lock: + conn = get_conn() + conn.executescript(SCHEMA) + # 单行种子数据 + conn.execute("INSERT OR IGNORE INTO config (id) VALUES (1)") + conn.execute("INSERT OR IGNORE INTO llm_key (id) VALUES (1)") + conn.execute("INSERT OR IGNORE INTO alert_state (id) VALUES (1)") + conn.commit() + + +def close_db() -> None: + """关闭 backend-lite 自有 SQLite 连接，确保进程可干净退出。""" + global _conn + with _lock: + if _conn is not None: + _conn.close() + _conn = None \ No newline at end of file diff --git a/backend-lite/store/secret_store.py b/backend-lite/store/secret_store.py new file mode 100644 index 00000000..f348d84f --- /dev/null +++ b/backend-lite/store/secret_store.py @@ -0,0 +1,76 @@ +"""LLM key 单行加密存储，复用 core/crypto.py。""" +from __future__ import annotations + +from typing import Any + +from .db import get_conn, _lock + + +def get_llm_key() -> dict[str, Any]: + """返回明文 key（已解密）。供渲染注入用。""" + from core.crypto import decrypt_api_key + + with _lock: + row = get_conn().execute("SELECT * FROM llm_key WHERE id=1").fetchone() + if row is None: + return {"provider": "deepseek", "api_key": "", "base_url": "", "image_api_key": ""} + data = dict(row) + out: dict[str, Any] = { + "provider": data.get("provider") or "deepseek", + "api_key": "", + "base_url": data.get("base_url") or "", + "image_api_key": "", + } + enc = data.get("api_key") or "" + if enc: + try: + out["api_key"] = decrypt_api_key(enc) + except Exception: + out["api_key"] = "" + enc_img = data.get("image_api_key") or "" + if enc_img: + try: + out["image_api_key"] = decrypt_api_key(enc_img) + except Exception: + out["image_api_key"] = "" + return out + + +def get_llm_key_masked() -> dict[str, Any]: + """供管理页展示：key 脱敏。""" + full = get_llm_key() + return { + "provider": full["provider"], + "base_url": full["base_url"], + "has_api_key": bool(full["api_key"]), + "has_image_api_key": bool(full["image_api_key"]), + } + + +def set_llm_key(patch: dict[str, Any]) -> dict[str, Any]: + """部分更新。空字符串字段不覆盖（避免脱敏表单清空已存 key）。""" + from core.crypto import encrypt_api_key + + cur = get_llm_key() + provider = patch.get("provider", cur["provider"]) + base_url = patch.get("base_url", cur["base_url"]) + + # api_key：只有显式传非空字符串才更新密文 + api_key_plain = cur["api_key"] + if "api_key" in patch and patch["api_key"]: + api_key_plain = str(patch["api_key"]) + image_api_key_plain = cur["image_api_key"] + if "image_api_key" in patch and patch["image_api_key"]: + image_api_key_plain = str(patch["image_api_key"]) + + enc_api = encrypt_api_key(api_key_plain) if api_key_plain else "" + enc_img = encrypt_api_key(image_api_key_plain) if image_api_key_plain else "" + + with _lock: + conn = get_conn() + conn.execute( + "UPDATE llm_key SET provider=?, api_key=?, base_url=?, image_api_key=? WHERE id=1", + (provider, enc_api, base_url, enc_img), + ) + conn.commit() + return get_llm_key_masked() \ No newline at end of file diff --git a/backend/api/shared.py b/backend/api/shared.py index f735fed9..3818b772 100644 --- a/backend/api/shared.py +++ b/backend/api/shared.py @@ -20,7 +20,7 @@ from fastapi import HTTPException, Request from fastapi.responses import JSONResponse from PIL import Image, ImageDraw, ImageFont -from core.patterns.utils import load_font +from core.patterns.utils import load_font, refresh_status_bar_dynamic_right try: # pragma: no cover - exercised implicitly at import time from slowapi import Limiter, _rate_limit_exceeded_handler @@ -646,7 +646,14 @@ def apply_user_llm_cfg(user_llm_cfg: dict) -> None: ) if cached_img: cache_hit = True - img = cached_img + img = refresh_status_bar_dynamic_right( + cached_img, + battery_pct, + time_str=datetime.now().strftime("%H:%M:%S"), + screen_w=screen_w, + screen_h=screen_h, + colors=colors, + ) if mac and config and is_mode_cacheable and not skip_cache: if not intent_only: @@ -661,7 +668,14 @@ def apply_user_llm_cfg(user_llm_cfg: dict) -> None: ) if cached_img: cache_hit = True - img = cached_img + img = refresh_status_bar_dynamic_right( + cached_img, + battery_pct, + time_str=datetime.now().strftime("%H:%M:%S"), + screen_w=screen_w, + screen_h=screen_h, + colors=colors, + ) # 即使缓存命中，如果这是需要 LLM 的模式且用户额度为0，也应该检查并返回兜底图 # 避免用户通过缓存绕过额度限制 # Root 用户无需检查额度 diff --git a/backend/core/json_renderer.py b/backend/core/json_renderer.py index 43fe841c..b4160258 100644 --- a/backend/core/json_renderer.py +++ b/backend/core/json_renderer.py @@ -1078,6 +1078,206 @@ def _paint_component_node(ctx: RenderContext, node: ComponentNode, theme: dict, ) +def _adaptive_typography_int(config: dict[str, Any], key: str, minimum: int = 0) -> int | None: + value = config.get(key) + if isinstance(value, bool): + return None + try: + parsed = int(value) + except (TypeError, ValueError): + return None + return parsed if parsed >= minimum else None + + +def _component_apply_adaptive_typography( + node: dict[str, Any], + config: dict[str, Any], + *, + is_root: bool = False, + in_lines_repeat: bool = False, +) -> None: + """Apply a preset's compact typography settings to its definition tree.""" + if not isinstance(node, dict): + return + + kind = node.get("type") + if is_root and kind == "column": + compact_gap = _adaptive_typography_int(config, "compact_gap") + if compact_gap is not None: + node["gap"] = compact_gap + compact_padding_y = _adaptive_typography_int(config, "compact_padding_y") + if compact_padding_y is not None: + node["padding_y"] = compact_padding_y + compact_top_gap = _adaptive_typography_int(config, "compact_top_gap") + if compact_top_gap is not None: + for child in node.get("children", []): + if isinstance(child, dict) and child.get("type") == "spacer": + child["height"] = compact_top_gap + break + + if kind == "text": + field = node.get("field") + if field == config.get("title_field"): + compact_title_size = _adaptive_typography_int(config, "compact_title_font_size", 1) + if compact_title_size is not None: + node["font_size"] = compact_title_size + elif in_lines_repeat and field == "_value": + compact_lines_size = _adaptive_typography_int(config, "compact_lines_font_size", 1) + if compact_lines_size is not None: + node["font_size"] = compact_lines_size + elif field == config.get("note_field"): + compact_note_max_lines = _adaptive_typography_int(config, "compact_note_max_lines") + if compact_note_max_lines is not None: + node["max_lines"] = compact_note_max_lines + + if kind == "repeat" and node.get("field") == config.get("lines_field"): + compact_lines_gap = _adaptive_typography_int(config, "compact_lines_gap") + if compact_lines_gap is not None: + node["gap"] = compact_lines_gap + item = node.get("item") + if isinstance(item, dict): + _component_apply_adaptive_typography( + item, config, in_lines_repeat=True, + ) + + for child in node.get("children", []): + if isinstance(child, dict): + _component_apply_adaptive_typography( + child, config, in_lines_repeat=in_lines_repeat, + ) + + +def _component_maybe_apply_adaptive_typography( + body_tree: dict[str, Any], + root: ComponentNode, + screen_w: int, + screen_h: int | None, + available_height: int, +) -> bool: + """Compact an explicitly opted-in component tree when it nearly overflows.""" + config = body_tree.get("adaptive_typography") + if not isinstance(config, dict) or config.get("enabled") is False: + return False + if screen_h is None: + return False + + target_screen = str(config.get("screen", "") or "").strip() + if target_screen and target_screen != f"{screen_w}x{screen_h}": + return False + + trigger_margin = _adaptive_typography_int(config, "trigger_margin") or 0 + trigger_height = max(0, available_height - trigger_margin) + if root.measured_height < trigger_height: + return False + + _component_apply_adaptive_typography(body_tree, config, is_root=True) + return True + + +def _component_has_auto_pair(node: ComponentNode) -> bool: + """True if any repeat node was condensed two-per-row by auto-pairing.""" + if node.kind == "repeat" and node.props.get("pair_step") == 2 and node.props.get("auto_pair_on_overflow"): + return True + return any(_component_has_auto_pair(child) for child in node.children) + + +def _component_auto_pair_repeat_nodes(node: ComponentNode, footer_top: int, theme: dict, scale: float) -> None: + """Condense overflowing ``repeat`` nodes two-per-row (``pair_step=2``). + + Recurses through the component tree. When a repeat node marked with + ``auto_pair_on_overflow`` extends past ``footer_top``, its ``pair_step`` / + ``pair_separator`` props are set so each line carries two items, then the + node is re-measured in place. + """ + if node.box is not None and node.box.y >= footer_top: + return + for child in node.children: + _component_auto_pair_repeat_nodes(child, footer_top, theme, scale) + if node.kind != "repeat" or not node.props.get("auto_pair_on_overflow"): + return + if node.box is not None and node.measured_height <= max(1, footer_top - node.box.y): + return + pair_sep = str(node.props.get("pair_separator", "，")) + node.props["pair_step"] = 2 + node.props["pair_separator"] = pair_sep + # Rebuild children (pairing happens in _build_component_node) and re-measure. + node.children = _build_component_node(node.props, node.content).children + _measure_component_node(node, node.measured_width, theme, scale) + + +def _fit_component_scale( + body_tree: dict, + content: dict, + screen_w: int, + theme: dict, + base_scale: float, + available_height: int, +) -> float | None: + """Binary-search the largest component scale whose measured height fits. + + Returns None when even the smallest scale still overflows (caller falls back + to the original scale rather than shrinking below a readable floor). + """ + lo = 0.72 + hi = base_scale + if hi <= lo: + return None + best: float | None = None + for _ in range(8): + mid = (lo + hi) / 2.0 + node = _build_component_node(body_tree, content) + _measure_component_node(node, screen_w, theme, mid) + if node.measured_height <= available_height: + best = mid + lo = mid + else: + hi = mid + return best + + +def _auto_fit_component_scale( + body_tree: dict, + content: dict, + screen_w: int, + theme: dict, + base_scale: float, + status_bar_bottom: int, + footer_top: int, + screen_h: int | None = None, +) -> float: + """Return the effective component scale after auto-fitting content. + + Content can exceed the body area (e.g. an 8-line poem on 400×300 where the + last lines would slip under the footer). Strategies, tried in order: + 1. explicit adaptive typography can compact a preset before overflow, + 2. auto_pair_on_overflow: overflowing repeat nodes condense two-per-row, + preserving font size (e.g. 律诗 becomes 每行两句), + 3. scale-down: shrink the whole component scale so everything fits. + + ``auto_pair_on_overflow`` mutates node props in place (shared with the + ``body_tree`` dicts), so pairing persists into the final render. + """ + scale = base_scale + root = _build_component_node(body_tree, content) + available_height = max(0, footer_top - status_bar_bottom) + _measure_component_node(root, screen_w, theme, scale) + if _component_maybe_apply_adaptive_typography( + body_tree, root, screen_w, screen_h, available_height, + ): + root = _build_component_node(body_tree, content) + _measure_component_node(root, screen_w, theme, scale) + _layout_component_node(root, 0, status_bar_bottom, screen_w, available_height, theme, scale) + _component_auto_pair_repeat_nodes(root, footer_top, theme, scale) + if _component_has_auto_pair(root): + # Re-measure the whole tree so parent heights reflect the condensed lines. + _measure_component_node(root, screen_w, theme, scale) + if root.measured_height > available_height: + fit_scale = _fit_component_scale(body_tree, content, screen_w, theme, scale, available_height) + if fit_scale is not None: + scale = fit_scale + return scale + + def _render_component_tree_mode( draw: ImageDraw.ImageDraw, img: Image.Image, @@ -1103,10 +1303,15 @@ def _render_component_tree_mode( footer_top_offset=footer_top_offset, colors=colors, ) - scale = _component_tree_scale(ctx, theme) + scale = _auto_fit_component_scale( + body_tree, content, screen_w, theme, + _component_tree_scale(ctx, theme), + status_bar_bottom, ctx.footer_top, + screen_h=screen_h, + ) root = _build_component_node(body_tree, content) - available_height = max(0, ctx.footer_top - status_bar_bottom) _measure_component_node(root, screen_w, theme, scale) + available_height = max(0, ctx.footer_top - status_bar_bottom) root_height = available_height if root.kind == "column" else min(available_height, root.measured_height) _layout_component_node(root, 0, status_bar_bottom, screen_w, root_height, theme, scale) _paint_component_node(ctx, root, theme, scale) diff --git a/backend/core/layout_presets.py b/backend/core/layout_presets.py index 6b9f3553..f77c9268 100644 --- a/backend/core/layout_presets.py +++ b/backend/core/layout_presets.py @@ -1830,6 +1830,7 @@ def _poetry_card(props: dict[str, Any]) -> dict[str, Any]: "field": props.get("lines_field", "lines"), "limit": props.get("lines_limit", 8), "gap": props.get("lines_gap", 8), + "auto_pair_on_overflow": props.get("auto_pair_on_overflow", True), "item": _text_node( field="_value", font=props.get("lines_font"), @@ -1870,7 +1871,7 @@ def _poetry_card(props: dict[str, Any]) -> dict[str, Any]: }, ) ) - return _compact( + root = _compact( { "type": "column", "padding_x": props.get("padding_x", 18), @@ -1881,6 +1882,14 @@ def _poetry_card(props: dict[str, Any]) -> dict[str, Any]: "children": children, } ) + adaptive_typography = props.get("adaptive_typography") + if isinstance(adaptive_typography, dict): + adaptive_meta = deepcopy(adaptive_typography) + adaptive_meta.setdefault("title_field", props.get("title_field", "title")) + adaptive_meta.setdefault("lines_field", props.get("lines_field", "lines")) + adaptive_meta.setdefault("note_field", props.get("note_field", "note")) + root["adaptive_typography"] = adaptive_meta + return root def _progress_metric(metric: dict[str, Any]) -> dict[str, Any]: @@ -2354,7 +2363,7 @@ def _fitness_card(props: dict[str, Any]) -> dict[str, Any]: ), "poetry_card": PresetSpec( builder=_poetry_card, - props=("padding_x", "padding_y", "justify", "gap", "content_bias_px", "top_gap", "title_field", "title_font", "title_font_name", "title_font_size", "title_max_lines", "author_field", "author_font", "author_font_name", "author_font_size", "author_max_lines", "separator_width", "lines_inset_x", "lines_field", "lines_limit", "lines_gap", "lines_font", "lines_font_name", "lines_font_size", "line_max_lines", "lines_pair_step", "lines_pair_separator", "note_field", "note_font", "note_font_name", "note_font_size", "note_max_lines"), + props=("padding_x", "padding_y", "justify", "gap", "content_bias_px", "top_gap", "title_field", "title_font", "title_font_name", "title_font_size", "title_max_lines", "author_field", "author_font", "author_font_name", "author_font_size", "author_max_lines", "separator_width", "lines_inset_x", "lines_field", "lines_limit", "lines_gap", "lines_font", "lines_font_name", "lines_font_size", "line_max_lines", "lines_pair_step", "lines_pair_separator", "note_field", "note_font", "note_font_name", "note_font_size", "note_max_lines", "adaptive_typography"), defaults={"title_field": "title", "title_font": "noto_serif_bold", "title_font_size": 20, "title_max_lines": 1, "author_field": "author", "author_font": "noto_serif_light", "author_font_size": 14, "author_max_lines": 1, "separator_width": 50, "lines_field": "lines", "lines_font": "noto_serif_light", "lines_font_size": 16, "lines_limit": 8, "lines_gap": 4, "line_max_lines": 1, "lines_inset_x": 24, "note_field": "note", "note_font": "noto_serif_light", "note_font_size": 12, "note_max_lines": 2, "padding_x": 18, "padding_y": 10, "justify": "center", "gap": 5, "top_gap": 6}, ), "lifebar_card": PresetSpec( diff --git a/backend/core/modes/builtin/poetry.json b/backend/core/modes/builtin/poetry.json index 6b2b997a..ae628da8 100644 --- a/backend/core/modes/builtin/poetry.json +++ b/backend/core/modes/builtin/poetry.json @@ -114,7 +114,18 @@ "note_field": "note", "note_font": "noto_serif_light", "note_font_size": 10, - "note_max_lines": 2 + "note_max_lines": 2, + "adaptive_typography": { + "screen": "400x300", + "trigger_margin": 8, + "compact_title_font_size": 16, + "compact_lines_font_size": 14, + "compact_lines_gap": 4, + "compact_gap": 4, + "compact_top_gap": 8, + "compact_padding_y": 6, + "compact_note_max_lines": 1 + } }, "footer": { "label": "POETRY", diff --git a/backend/core/patterns/utils.py b/backend/core/patterns/utils.py index bee02e11..1eec192e 100644 --- a/backend/core/patterns/utils.py +++ b/backend/core/patterns/utils.py @@ -327,6 +327,141 @@ def draw_dashed_line( x += dash_len + gap_len +def _resolve_status_bar_time(time_str: str = "") -> tuple[str, int]: + """Return a zero-padded HH:MM string and its hour for period labels.""" + match = re.match(r"^\s*(\d{1,2}):(\d{2})(?::(\d{2}))?\s*$", str(time_str or "")) + if match: + try: + hour = int(match.group(1)) + minute = int(match.group(2)) + second = int(match.group(3) or 0) + if 0 <= hour <= 23 and 0 <= minute <= 59 and 0 <= second <= 59: + return f"{hour:02d}:{minute:02d}", hour + except ValueError: + pass + + now = datetime.now() + return f"{now.hour:02d}:{now.minute:02d}", now.hour + + +def _status_bar_dynamic_group_layout( + draw: ImageDraw.ImageDraw, + time_text: str, + battery_pct: int, + font_en: ImageFont.ImageFont, + screen_w: int, + screen_h: int, +) -> dict[str, int]: + """Calculate the right-aligned time and battery group using real text bounds.""" + scale = screen_w / 400.0 + pad_pct = 0.02 if screen_h < 200 else 0.03 + pad_x = int(screen_w * pad_pct) + battery_text = f"{int(battery_pct)}%" + battery_bbox = draw.textbbox((0, 0), battery_text, font=font_en) + time_bbox = draw.textbbox((0, 0), time_text, font=font_en) + battery_text_x = screen_w - pad_x - battery_bbox[2] + battery_text_left = battery_text_x + battery_bbox[0] + battery_box_w = int(22 * scale) + battery_box_h = int(11 * scale) + battery_gap = int(6 * scale) + battery_x = battery_text_left - battery_gap - battery_box_w + time_gap = int(8 * scale) + time_right = battery_x - time_gap + time_x = time_right - time_bbox[2] + return { + "time_x": time_x, + "time_y": int(screen_h * (0.02 if screen_h < 200 else 0.03)), + "time_left": time_x + time_bbox[0], + "time_right": time_x + time_bbox[2], + "battery_x": battery_x, + "battery_y": int(screen_h * (0.02 if screen_h < 200 else 0.03)) + 1, + "battery_box_w": battery_box_w, + "battery_box_h": battery_box_h, + "battery_text_x": battery_text_x, + "battery_text_y": int(screen_h * (0.02 if screen_h < 200 else 0.03)), + "battery_text_right": battery_text_x + battery_bbox[2], + "group_left": min(time_x + time_bbox[0], battery_x), + "group_right": screen_w - pad_x, + } + + +def _draw_status_bar_dynamic_group( + draw: ImageDraw.ImageDraw, + time_text: str, + battery_pct: int, + font_en: ImageFont.ImageFont, + screen_w: int, + screen_h: int, + colors: int, +) -> dict[str, int]: + """Draw the right-side time and battery group and return its geometry.""" + layout = _status_bar_dynamic_group_layout( + draw, time_text, battery_pct, font_en, screen_w, screen_h, + ) + battery_text = f"{int(battery_pct)}%" + battery_fill = EINK_FG + available = EINK_COLOR_AVAILABILITY.get(colors, frozenset()) + if battery_pct < 20 and "red" in available: + battery_fill = EINK_COLOR_NAME_MAP["red"] + elif battery_pct < 50 and "yellow" in available: + battery_fill = EINK_COLOR_NAME_MAP["yellow"] + + draw.text((layout["time_x"], layout["time_y"]), time_text, fill=EINK_FG, font=font_en) + bx = layout["battery_x"] + by = layout["battery_y"] + batt_box_w = layout["battery_box_w"] + batt_box_h = layout["battery_box_h"] + scale = screen_w / 400.0 + draw.rectangle([bx, by, bx + batt_box_w, by + batt_box_h], outline=battery_fill, width=1) + draw.rectangle( + [bx + batt_box_w, by + int(3 * scale), bx + batt_box_w + int(2 * scale), by + int(8 * scale)], + fill=battery_fill, + ) + fill_w = int((batt_box_w - 4) * battery_pct / 100) + if fill_w > 0: + draw.rectangle([bx + 2, by + 2, bx + 2 + fill_w, by + batt_box_h - 2], fill=battery_fill) + draw.text( + (layout["battery_text_x"], layout["battery_text_y"]), + battery_text, + fill=battery_fill, + font=font_en, + ) + return layout + + +def refresh_status_bar_dynamic_right( + img: Image.Image, + battery_pct: int, + *, + time_str: str = "", + screen_w: int | None = None, + screen_h: int | None = None, + colors: int = 2, +) -> Image.Image: + """Refresh only the right-side time/battery area on a cached image copy.""" + updated = img.copy() + width = screen_w or updated.width + height = screen_h or updated.height + draw = ImageDraw.Draw(updated) + apply_text_fontmode(draw) + scale = width / 400.0 + time_text, _ = _resolve_status_bar_time(time_str) + font_en = load_font("inter_medium", int(FONT_SIZES["status_bar"]["en"] * scale)) + + max_layout = _status_bar_dynamic_group_layout( + draw, time_text, 100, font_en, width, height, + ) + line_y = int(height * 0.11) + clear_bottom = max(0, min(height, line_y)) + clear_left = max(0, max_layout["group_left"] - max(1, int(2 * scale))) + if clear_bottom > 0: + draw.rectangle([clear_left, 0, width - 1, clear_bottom - 1], fill=EINK_BG) + _draw_status_bar_dynamic_group( + draw, time_text, int(battery_pct), font_en, width, height, colors, + ) + return updated + + def draw_status_bar( draw: ImageDraw.ImageDraw, img: Image.Image, @@ -357,15 +492,7 @@ def draw_status_bar( period_font = load_font("noto_serif_regular", period_font_size) font_en = load_font("inter_medium", int(FONT_SIZES["status_bar"]["en"] * scale)) - match = re.match(r"^\s*(\d{1,2})\s*:", time_str or "") - hour = datetime.now().hour - if match: - try: - parsed_hour = int(match.group(1)) - if 0 <= parsed_hour <= 23: - hour = parsed_hour - except ValueError: - pass + time_text, hour = _resolve_status_bar_time(time_str) if is_en: if hour >= 23 or hour < 5: @@ -413,28 +540,9 @@ def draw_status_bar( else: draw.text((wx, y), weather_str, fill=EINK_FG, font=font_date) - batt_text = f"{battery_pct}%" - bbox = draw.textbbox((0, 0), batt_text, font=font_en) - batt_text_w = bbox[2] - bbox[0] - - batt_fill = EINK_FG - available = EINK_COLOR_AVAILABILITY.get(colors, frozenset()) - if battery_pct < 20 and "red" in available: - batt_fill = EINK_COLOR_NAME_MAP["red"] - elif battery_pct < 50 and "yellow" in available: - batt_fill = EINK_COLOR_NAME_MAP["yellow"] - - batt_box_w = int(22 * scale) - batt_box_h = int(11 * scale) - bx = screen_w - pad_x - batt_text_w - int(6 * scale) - batt_box_w - by = y + 1 - draw.rectangle([bx, by, bx + batt_box_w, by + batt_box_h], outline=batt_fill, width=1) - draw.rectangle([bx + batt_box_w, by + int(3 * scale), bx + batt_box_w + int(2 * scale), by + int(8 * scale)], fill=batt_fill) - fill_w = int((batt_box_w - 4) * battery_pct / 100) - if fill_w > 0: - draw.rectangle([bx + 2, by + 2, bx + 2 + fill_w, by + batt_box_h - 2], fill=batt_fill) - - draw.text((bx + batt_box_w + int(6 * scale), y), batt_text, fill=batt_fill, font=font_en) + _draw_status_bar_dynamic_group( + draw, time_text, battery_pct, font_en, screen_w, screen_h, colors, + ) if separator_y is not None: line_y = max(0, min(int(separator_y), max(0, screen_h - 1))) diff --git a/backend/core/pipeline.py b/backend/core/pipeline.py index deb8584a..c97948fd 100644 --- a/backend/core/pipeline.py +++ b/backend/core/pipeline.py @@ -101,7 +101,6 @@ async def generate_and_render( Returns: Tuple of (rendered image, content dict). """ - time_str = date_ctx.get("time_str", "") weather_str = weather["weather_str"] weather_code = weather.get("weather_code", -1) cfg = get_effective_mode_config(config, persona) @@ -119,6 +118,7 @@ async def generate_and_render( eff_cfg = get_effective_mode_config(config, persona) _eff_lang = eff_cfg.get("mode_language", "") or DEFAULT_LANGUAGE date_str = _format_date_str(date_ctx, _eff_lang) + render_time_str = datetime.now().strftime("%H:%M:%S") img = await _render_for_persona( persona, @@ -127,7 +127,7 @@ async def generate_and_render( weather_str=weather_str, battery_pct=battery_pct, weather_code=weather_code, - time_str=time_str, + time_str=render_time_str, date_ctx=date_ctx, screen_w=screen_w, screen_h=screen_h, diff --git a/backend/scripts/setup_fonts.py b/backend/scripts/setup_fonts.py index 72f9f3cb..e6a910a7 100644 --- a/backend/scripts/setup_fonts.py +++ b/backend/scripts/setup_fonts.py @@ -32,6 +32,9 @@ "Inter": [ "Inter_24pt-Medium.ttf", ], + "Gentium+Plus": [ + "GentiumPlus-Regular.ttf", + ], } BITMAP_FONT_URLS: dict[str, str] = { diff --git a/backend/tests/test_json_renderer.py b/backend/tests/test_json_renderer.py index 68e013a8..6c783206 100644 --- a/backend/tests/test_json_renderer.py +++ b/backend/tests/test_json_renderer.py @@ -2,15 +2,31 @@ 测试 JSON 渲染引擎 验证各种布局原语能正确渲染到 1-bit e-ink 图像 """ +import json import os import sys from io import BytesIO +from pathlib import Path from unittest.mock import patch sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) from PIL import Image -from core.json_renderer import render_json_mode, RenderContext, _localized_footer_label, _localized_footer_attribution, _component_aligned_y +from copy import deepcopy + +from core.json_renderer import ( + render_json_mode, + RenderContext, + _localized_footer_label, + _localized_footer_attribution, + _component_aligned_y, + _auto_fit_component_scale, + _merge_layout_dict, + _build_component_node, + _component_has_auto_pair, + _measure_component_node, +) +from core.layout_presets import expand_layout_presets from core.config import SCREEN_WIDTH as SCREEN_W, SCREEN_HEIGHT as SCREEN_H @@ -51,6 +67,28 @@ def _make_component_tree_mode_def(body_tree, footer=None): } +def _load_builtin_poetry_mode(): + path = Path(__file__).resolve().parents[1] / "core" / "modes" / "builtin" / "poetry.json" + return json.loads(path.read_text()) + + +def _find_component_def(node, predicate): + if not isinstance(node, dict): + return None + if predicate(node): + return node + for child in node.get("children", []): + found = _find_component_def(child, predicate) + if found is not None: + return found + item = node.get("item") + if isinstance(item, dict): + found = _find_component_def(item, predicate) + if found is not None: + return found + return None + + def test_render_list_wraps_to_multiple_lines(): mode_def = _make_mode_def([ {"type": "spacer", "height": 14}, @@ -425,6 +463,297 @@ def test_render_component_tree_poetry_preset_mode(): assert img.size == (SCREEN_W, SCREEN_H) +def test_poetry_auto_pair_on_overflow(): + """8-line poems auto-pair two-per-row; 4-line poems stay single-column.""" + mode_def = { + "mode_id": "POETRY_PRESET", + "display_name": "Poetry Preset", + "content": {"type": "static"}, + "layout": { + "layout_engine": "component_tree", + "status_bar": {"line_width": 1, "dashed": True}, + "body_preset": "poetry_card", + "preset_props": { + "title_field": "title", + "author_field": "author", + "lines_field": "lines", + "note_field": "note", + }, + "footer": {"label": "POETRY", "attribution_template": "— {author}"}, + }, + } + status_bar_bottom = int(SCREEN_H * 0.12) + footer_top = SCREEN_H - 30 + available = footer_top - status_bar_bottom + + eight_line = { + "title": "春江花月夜", + "author": "唐·张若虚", + "lines": [ + "春江潮水连海平", "海上明月共潮生", + "滟滟随波千万里", "何处春江无月明", + "江流宛转绕芳甸", "月照花林皆似霰", + "空里流霜不觉飞", "汀上白沙看不见", + ], + "note": "孤篇压全唐", + } + four_line = { + "title": "静夜思", + "author": "唐·李白", + "lines": ["床前明月光", "疑是地上霜", "举头望明月", "低头思故乡"], + "note": "千古思乡名篇", + } + + layout = expand_layout_presets(mode_def["layout"]) + body = layout["body"] + + # 8-line overflows without pairing, then auto-pairs and fits. + base_root = _build_component_node(deepcopy(body), eight_line) + _measure_component_node(base_root, SCREEN_W, {}, 1.0) + assert base_root.measured_height > available + + body_8 = deepcopy(body) + scale_8 = _auto_fit_component_scale( + body_8, eight_line, SCREEN_W, {}, 1.0, status_bar_bottom, footer_top, + ) + root_8 = _build_component_node(body_8, eight_line) + _measure_component_node(root_8, SCREEN_W, {}, scale_8) + assert _component_has_auto_pair(root_8) + assert root_8.measured_height <= available + assert scale_8 == 1.0 + + # 4-line already fits — no pairing, no scale-down. + body_4 = deepcopy(body) + scale_4 = _auto_fit_component_scale( + body_4, four_line, SCREEN_W, {}, 1.0, status_bar_bottom, footer_top, + ) + root_4 = _build_component_node(body_4, four_line) + _measure_component_node(root_4, SCREEN_W, {}, scale_4) + assert not _component_has_auto_pair(root_4) + assert root_4.measured_height <= available + assert scale_4 == 1.0 + + # End-to-end render still succeeds for both. + for content in (eight_line, four_line): + img = render_json_mode( + mode_def, content, + date_str="2月18日", weather_str="晴", battery_pct=80, + ) + assert img.size == (SCREEN_W, SCREEN_H) + + +def test_poetry_adaptive_typography_compacts_long_ci_only(): + """Long ci uses a larger title and smaller verse text on 400x300.""" + mode_def = _load_builtin_poetry_mode() + layout = expand_layout_presets(mode_def["layout"]) + body = layout["body"] + long_ci = { + "title": "水调歌头·明月几时有", + "author": "宋·苏轼", + "lines": [ + "明月几时有？把酒问青天。", + "不知天上宫阙，今夕是何年。", + "我欲乘风归去，又恐琼楼玉宇，高处不胜寒。", + "起舞弄清影，何似在人间。", + "转朱阁，低绮户，照无眠。", + "不应有恨，何事长向别时圆？", + ], + "note": "此事古难全·宋苏轼", + } + status_bar_bottom = 36 + footer_top = 270 + available = footer_top - status_bar_bottom + + normal_root = _build_component_node(deepcopy(body), long_ci) + _measure_component_node(normal_root, SCREEN_W, {}, 1.0) + assert normal_root.measured_height >= available - 8 + + compact_body = deepcopy(body) + scale = _auto_fit_component_scale( + compact_body, + long_ci, + SCREEN_W, + {}, + 1.0, + status_bar_bottom, + footer_top, + screen_h=SCREEN_H, + ) + compact_root = _build_component_node(compact_body, long_ci) + _measure_component_node(compact_root, SCREEN_W, {}, scale) + + title = _find_component_def( + compact_body, + lambda node: node.get("type") == "text" and node.get("field") == "title", + ) + verse = _find_component_def( + compact_body, + lambda node: node.get("type") == "text" and node.get("field") == "_value", + ) + repeat = _find_component_def( + compact_body, + lambda node: node.get("type") == "repeat" and node.get("field") == "lines", + ) + note = _find_component_def( + compact_body, + lambda node: node.get("type") == "text" and node.get("field") == "note", + ) + assert title is not None and verse is not None and repeat is not None and note is not None + assert title["font_size"] == 16 + assert verse["font_size"] == 14 + assert title["font_size"] > verse["font_size"] + assert repeat["gap"] == 4 + assert compact_body["gap"] == 4 + assert compact_body["padding_y"] == 6 + assert compact_body["children"][0]["height"] == 8 + assert note["max_lines"] == 1 + assert not _component_has_auto_pair(compact_root) + assert scale == 1.0 + assert compact_root.measured_height <= available + + image = render_json_mode( + mode_def, + long_ci, + date_str="8月10日", + weather_str="晴", + battery_pct=80, + screen_w=SCREEN_W, + screen_h=SCREEN_H, + ) + assert image.size == (SCREEN_W, SCREEN_H) + + +def test_poetry_short_ci_keeps_normal_typography(): + mode_def = _load_builtin_poetry_mode() + layout = expand_layout_presets(mode_def["layout"]) + body = deepcopy(layout["body"]) + short_ci = { + "title": "静夜思", + "author": "唐·李白", + "lines": ["床前明月光", "疑是地上霜", "举头望明月", "低头思故乡"], + "note": "千古思乡名篇", + } + scale = _auto_fit_component_scale( + body, short_ci, SCREEN_W, {}, 1.0, 36, 270, screen_h=SCREEN_H, + ) + title = _find_component_def( + body, + lambda node: node.get("type") == "text" and node.get("field") == "title", + ) + verse = _find_component_def( + body, + lambda node: node.get("type") == "text" and node.get("field") == "_value", + ) + root = _build_component_node(body, short_ci) + _measure_component_node(root, SCREEN_W, {}, scale) + assert title is not None and verse is not None + assert title["font_size"] == 14 + assert verse["font_size"] == 16 + assert not _component_has_auto_pair(root) + assert scale == 1.0 + + +def test_poetry_adaptive_typography_is_scoped_to_400x300(): + mode_def = _load_builtin_poetry_mode() + content = { + "title": "水调歌头·明月几时有", + "author": "宋·苏轼", + "lines": ["明月几时有？把酒问青天。"] * 8, + "note": "此事古难全", + } + expected = { + "296x128": (11, 10), + "648x480": (18, 18), + "800x480": (22, 24), + } + for size_key, (title_size, verse_size) in expected.items(): + width, height = (int(value) for value in size_key.split("x")) + layout = _merge_layout_dict(mode_def["layout"], mode_def["layout_overrides"][size_key]) + body = expand_layout_presets(layout)["body"] + title = _find_component_def( + body, + lambda node: node.get("type") == "text" and node.get("field") == "title", + ) + verse = _find_component_def( + body, + lambda node: node.get("type") == "text" and node.get("field") == "_value", + ) + assert title is not None and verse is not None + assert title["font_size"] == title_size + assert verse["font_size"] == verse_size + + _auto_fit_component_scale( + body, + content, + width, + {}, + 0.92 if width < 400 else 1.35, + int(height * (0.11 if height <= 128 else 0.12)) + (2 if height <= 128 else 0), + height - (24 if height <= 128 else int(30 * min(width / 400, height / 300))), + screen_h=height, + ) + assert title["font_size"] == title_size + assert verse["font_size"] == verse_size + + +def test_component_tree_scale_down_when_pairing_unavailable(): + """Without auto_pair, overflowing content shrinks via binary-search scale.""" + body_tree = { + "type": "column", + "padding_x": 12, + "gap": 6, + "children": [ + { + "type": "repeat", + "field": "lines", + "gap": 10, + "item": { + "type": "text", + "field": "_value", + "font": "noto_serif_regular", + "font_size": 20, + "align": "center", + "max_lines": 1, + }, + } + ], + } + content = { + "lines": [ + "第一行内容足够长一些", + "第二行内容足够长一些", + "第三行内容足够长一些", + "第四行内容足够长一些", + "第五行内容足够长一些", + "第六行内容足够长一些", + "第七行内容足够长一些", + "第八行内容足够长一些", + "第九行内容足够长一些", + "第十行内容足够长一些", + ] + } + status_bar_bottom = 36 + footer_top = 270 + available = footer_top - status_bar_bottom + + base_root = _build_component_node(deepcopy(body_tree), content) + _measure_component_node(base_root, SCREEN_W, {}, 1.0) + assert base_root.measured_height > available + + body = deepcopy(body_tree) + fit_scale = _auto_fit_component_scale( + body, content, SCREEN_W, {}, 1.0, status_bar_bottom, footer_top, + ) + assert fit_scale is not None + assert fit_scale < 1.0 + assert fit_scale >= 0.72 + assert not _component_has_auto_pair(_build_component_node(body, content)) + + fitted = _build_component_node(body, content) + _measure_component_node(fitted, SCREEN_W, {}, fit_scale) + assert fitted.measured_height <= available + + def test_render_component_tree_lifebar_preset_mode(): mode_def = { "mode_id": "LIFEBAR_PRESET", diff --git a/backend/tests/test_status_bar.py b/backend/tests/test_status_bar.py new file mode 100644 index 00000000..9766d3c5 --- /dev/null +++ b/backend/tests/test_status_bar.py @@ -0,0 +1,111 @@ +from datetime import datetime as real_datetime + +import pytest +from PIL import Image, ImageDraw + +import core.patterns.utils as utils + + +@pytest.mark.parametrize( + ("raw", "expected", "hour"), + [ + ("09:30:45", "09:30", 9), + ("9:03", "09:03", 9), + ("23:59:59", "23:59", 23), + ("00:00", "00:00", 0), + ], +) +def test_resolve_status_bar_time_formats_hh_mm(raw, expected, hour): + assert utils._resolve_status_bar_time(raw) == (expected, hour) + + +def test_resolve_status_bar_time_falls_back_to_current_minute(monkeypatch): + class FrozenDatetime: + @classmethod + def now(cls): + return real_datetime(2026, 8, 11, 14, 37, 59) + + monkeypatch.setattr(utils, "datetime", FrozenDatetime) + assert utils._resolve_status_bar_time("") == ("14:37", 14) + assert utils._resolve_status_bar_time("not-a-time") == ("14:37", 14) + assert utils._resolve_status_bar_time("24:80:00") == ("14:37", 14) + + +def test_draw_status_bar_uses_normalized_time_without_parentheses(monkeypatch): + captured = {} + + def capture_group(draw, time_text, battery_pct, font_en, screen_w, screen_h, colors): + captured["time_text"] = time_text + captured["battery_pct"] = battery_pct + return {} + + monkeypatch.setattr(utils, "_draw_status_bar_dynamic_group", capture_group) + image = Image.new("1", (400, 300), 1) + utils.draw_status_bar( + ImageDraw.Draw(image), + image, + date_str="8月11日 周二", + weather_str="晴", + battery_pct=85, + time_str="9:03:45", + screen_w=400, + screen_h=300, + ) + assert captured == {"time_text": "09:03", "battery_pct": 85} + assert "(" not in captured["time_text"] + assert ")" not in captured["time_text"] + + +@pytest.mark.parametrize("screen_w, screen_h", [(296, 128), (400, 300), (800, 480)]) +@pytest.mark.parametrize("battery_pct", [0, 85, 100]) +def test_dynamic_group_fits_screen(screen_w, screen_h, battery_pct): + image = Image.new("1", (screen_w, screen_h), 1) + draw = ImageDraw.Draw(image) + scale = screen_w / 400.0 + font = utils.load_font( + "inter_medium", + int(utils.FONT_SIZES["status_bar"]["en"] * scale), + ) + layout = utils._status_bar_dynamic_group_layout( + draw, + "23:59", + battery_pct, + font, + screen_w, + screen_h, + ) + assert layout["group_left"] >= 0 + assert layout["time_left"] >= 0 + assert layout["time_right"] <= layout["battery_x"] + assert layout["battery_x"] + layout["battery_box_w"] + int(2 * scale) <= screen_w + assert layout["battery_text_right"] <= layout["group_right"] + + +def test_refresh_dynamic_group_does_not_mutate_cached_image(): + original = Image.new("1", (400, 300), 1) + original_bytes = original.tobytes() + updated = utils.refresh_status_bar_dynamic_right( + original, + 85, + time_str="14:37:59", + screen_w=400, + screen_h=300, + ) + assert updated is not original + assert original.tobytes() == original_bytes + assert updated.tobytes() != original_bytes + + +def test_refresh_dynamic_group_supports_four_color_image(): + image = Image.new("P", (400, 300), utils.EINK_BG) + image.putpalette(utils.EINK_COLOR_NAME_MAP.get("white", 1).to_bytes(1, "big") * 3 + bytes(765)) + updated = utils.refresh_status_bar_dynamic_right( + image, + 20, + time_str="02:00:00", + screen_w=400, + screen_h=300, + colors=4, + ) + assert updated.mode == "P" + assert updated.size == (400, 300) diff --git a/backend/tests/test_unit_pipeline.py b/backend/tests/test_unit_pipeline.py index 0454a848..e1fd0036 100644 --- a/backend/tests/test_unit_pipeline.py +++ b/backend/tests/test_unit_pipeline.py @@ -1,6 +1,8 @@ """ Unit tests for the unified generate_and_render pipeline. """ +from datetime import datetime as real_datetime + import pytest from unittest.mock import AsyncMock, patch, MagicMock from PIL import Image @@ -159,7 +161,9 @@ async def test_full_pipeline(self, sample_config, sample_date_ctx, sample_weathe patch("core.mode_registry.get_registry", return_value=mock_reg), patch("core.json_content.generate_json_mode_content", new_callable=AsyncMock) as mock_gc, patch("core.json_renderer.render_json_mode", return_value=mock_img) as mock_rm, + patch("core.pipeline.datetime") as mock_datetime, ): + mock_datetime.now.return_value = real_datetime(2026, 8, 11, 14, 37, 59) mock_gc.return_value = {"quote": "Test", "author": "Author"} result_img, result_content = await generate_and_render( @@ -169,6 +173,7 @@ async def test_full_pipeline(self, sample_config, sample_date_ctx, sample_weathe assert result_content == {"quote": "Test", "author": "Author"} mock_gc.assert_called_once() mock_rm.assert_called_once() + assert mock_rm.call_args.kwargs["time_str"] == "14:37:59" @pytest.mark.asyncio async def test_none_config_treated_as_empty(self, sample_date_ctx, sample_weather):