commit f26c4908615e1a4aca59378b84785ea2d574fe2b Author: YANG JIANKUAN Date: Mon Jun 29 09:43:20 2026 +0800 init: init proj diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..131334c --- /dev/null +++ b/.gitignore @@ -0,0 +1,6 @@ +output/ +__pycache__/ +*.pyc +.weather_cache.json +.trend_cache.json +run.sh diff --git a/CLAUDE.md b/CLAUDE.md new file mode 100644 index 0000000..eeeb8e4 --- /dev/null +++ b/CLAUDE.md @@ -0,0 +1,47 @@ +# CLAUDE.md + +This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository. + +## 项目定位 + +`eink-push` 是 JM monorepo 中一个**独立的 Python 工具**，与其余 Java/Vue 子项目技术栈无关，不受父级 CLAUDE.md 的 jm-cloud/uniapp/admin 开发约束。 + +作用：把橘喵今日经营、近30天趋势、日期与天气合成为一张 **400×300 纯黑白（1bit）** PNG，推送到 Zectrix 极趣云墨水屏设备（`zectrix-s3-epaper-4.2`）。纯 Python + Pillow，**无浏览器依赖**，适合 cron 定时运行。 + +## 命令 + +```bash +pip install -r requirements.txt # 仅 Pillow>=10.0；urllib 用标准库 +python3 main.py # 取数据 → 渲染 → 推送（设备取列表第一个） +python3 main.py --render-only # 只渲染到 output/dashboard.png，不推送（本地调试渲染时用这个） + +# 环境变量覆盖配置（见 config.py） +ZECTRIX_API_KEY=zt_xxx ZECTRIX_DEVICE_ID=AA:BB:CC:DD:EE:FF python3 main.py +``` + +无测试、无 lint。调试渲染效果就跑 `--render-only` 看 `output/dashboard.png`。 + +## 架构（数据 → 渲染 → 推送 三层） + +`main.py` 串联三层，各层职责单一、低耦合： + +- **`data.py`** — 唯一数据入口 `get_dashboard_data()`，返回固定结构的 dict。日期/时间用真实系统时间；橘喵今日经营+趋势由 `_mao()` 经 `jm_api`、天气由 `_weather()` 经 `weather_api`（和风，定位南京·秦淮）拉真实数据，**任何接口异常都回退占位 `"--"`，不回退 mock**。改数据保持返回结构不变即可（renderer 依赖其 key，尤其 `mao.trend.series` 每条须带 `style`；环比/同比 `(direction, text)`，direction 为 `None` 时不画涨跌三角）。 +- **`jm_api.py`** — jm-devops 后端统计接口客户端，**免鉴权 GET**（`/system/statistics/realtime`、`/recent-days?days=`）。注：早期曾实现登录+RSA/AES 加密+验证码 OCR，后改为后端直接放开这两个接口，故已全部删除——若后端再次收紧鉴权，参考 jm-devops 的 `src/utils/{crypto,jsencrypt,request}.ts` 复刻。 +- **`weather_api.py`** — 和风天气（QWeather）客户端。用**用户专属 API Host**（`QWEATHER_HOST`）+ `X-QW-Api-Key` 头鉴权；响应 **gzip 压缩**（按 magic number 手动解压）、返回 `code` 为**字符串**。`get_now()` 取实时（`now.text` 中文天气 / `now.temp`），`get_today()` 取 `/3d` 的 `daily[0]` 今日温区。`QWEATHER_HOST`/`QWEATHER_KEY` 未配置时直接抛错→天气区显示 `"--"`。 +- **`renderer.py`** — `DashboardRenderer.render(data)` 返回 PIL `'L'` 模式纯 0/255 图；`render_to_file()` 落盘。布局硬编码为三区：大时钟+日期 / 天气 / 橘喵三列 / 近30天趋势折线图。趋势线用 `_styled_polyline()` 按 solid/dashed/dotted 区分（设备无颜色），各指标按自身极值独立归一化（故只看趋势形状、不可横向比绝对值）。 +- **`pusher.py`** — `push_image()` 走 Zectrix API（`POST /devices/{id}/display/image`，手写 multipart）。`resolve_device_id()` 在 `config.DEVICE_ID` 为空时自动取设备列表第一个。 +- **`config.py`** — API、设备、画布尺寸、字体路径、推送参数；敏感项可被环境变量覆盖。 + +## 墨水屏渲染约束（改 renderer.py 必读） + +设备是 **1bit、无真灰阶**——灰阶只能抖动成网点（难看），所以全程纯黑白，靠构图/字号/留白建立层级，**不要引入灰色填充或抖动**： + +- 字体固定用 **Fusion Pixel 12px 点阵字体**（`fonts/`，OFL 协议），只按整数倍尺寸 **12/24/48** 渲染（`F12/F24/F48`），并设 `d.fontmode = "1"` 关闭抗锯齿，否则像素会糊。 +- 加粗用「伪粗体」：`_draw()` 按 1~bold px 水平偏移叠绘笔画（点阵字体只有单一字重）。 +- 推送时 `DITHER = False`（硬阈值），纯黑白图最锐利，勿改成 `true`。 +- 坐标、字号均为像素网格上的硬编码常量；调布局时注意各区分隔线 `hdash` 的 y 值与下方组件位置联动。 + +## 注意 + +- `config.py` 内置了一个默认 `API_KEY`（硬编码 fallback）。修改/分享代码时留意，优先用 `ZECTRIX_API_KEY` 环境变量。 +- `output/` 与 `__pycache__/` 已在 `.gitignore` 中忽略。 diff --git a/README.md b/README.md new file mode 100644 index 0000000..6da3256 --- /dev/null +++ b/README.md @@ -0,0 +1,72 @@ +# 墨水屏仪表盘推送（eink-push） + +将橘喵今日经营、近30天趋势、日期与天气合成为一张 **400×300 纯黑白**图片， +推送到 Zectrix 极趣云墨水屏设备（`zectrix-s3-epaper-4.2`，1bit）。 + +纯 Python + Pillow + Fusion Pixel 点阵字体，**无浏览器依赖**，适合后台/定时运行。 + +## 效果 + +- 顶部：大号时钟 + 日期 ｜ 天气（图标 + 温度 + 区间） +- 中部：橘喵今日经营（单量 / 流水 / 毛利，含环比、同比涨跌） +- 底部：近30天三指标趋势折线图（同一图表，实线 / 虚线 / 点线区分，X 轴仅月/日） + +## 目录结构 + +``` +eink-push/ +├── main.py # 入口：取数据 → 渲染 → 推送 +├── config.py # 配置：API Key、设备、字体、尺寸、推送参数 +├── data.py # 数据层：拉真实数据并映射，失败回退 mock +├── jm_api.py # jm-devops 后端统计接口客户端（免鉴权 GET） +├── renderer.py # 渲染层：DashboardRenderer 生成 400×300 图片 +├── pusher.py # 推送层：设备列表 / 推送图片（标准库 urllib） +├── fonts/ # Fusion Pixel 点阵字体（OFL 协议） +├── output/ # 生成的图片（git 忽略） +├── requirements.txt +└── README.md +``` + +## 安装 + +```bash +pip install -r requirements.txt +``` + +## 使用 + +```bash +python3 main.py # 渲染并推送（默认取设备列表第一个） +python3 main.py --render-only # 只渲染到 output/dashboard.png，不推送 +``` + +环境变量可覆盖配置： + +```bash +ZECTRIX_API_KEY=zt_xxx ZECTRIX_DEVICE_ID=AA:BB:CC:DD:EE:FF python3 main.py +``` + +## 定时刷新（cron 示例，每 15 分钟） + +```cron +*/15 * * * * cd /path/to/eink-push && /usr/bin/python3 main.py >> /tmp/eink.log 2>&1 +``` + +## 数据来源 + +- **橘喵今日经营 + 近30天趋势**：已接入 jm-devops 后端统计接口（`jm_api.py`）， + 免鉴权直接 GET： + - `GET /system/statistics/realtime` —— 今日单量/流水/毛利 + 环比/同比 + - `GET /system/statistics/recent-days?days=30` —— 近30天每日数据 + `data.py` 负责把响应映射成渲染结构；**拉取失败会自动回退 mock**， + 设 `JM_USE_MOCK=1` 可强制用 mock 离线调试。后端地址等见 `config.py`。 +- **天气**：`data.py` 的 `_weather()` 仍为 mock 占位，接入真实天气 API 时替换它即可。 +- **日期 / 时间 / 上次更新**：真实系统时间。 + +## 设计说明 + +- 设备为 **1bit 无真灰阶**，灰阶只能抖动成网点（难看），故全程纯黑白，靠构图、字号、留白建立层级。 +- 字体用 **Fusion Pixel 12px 点阵字体**，按整数倍尺寸(12/24/48)渲染、关闭抗锯齿 → 像素锐利； + 重点数据用「伪粗体」(偏移叠绘)加粗。 +- 推送用 `dither=false`（硬阈值），纯黑白图最锐利。 +``` diff --git a/config.py b/config.py new file mode 100644 index 0000000..eec054f --- /dev/null +++ b/config.py @@ -0,0 +1,43 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +"""全局配置。敏感项支持用环境变量覆盖。""" +import os + +BASE_DIR = os.path.dirname(os.path.abspath(__file__)) + +# ---- Zectrix 极趣云平台 API ---- +API_BASE = "https://cloud.zectrix.com/open/v1" +API_KEY = os.environ.get("ZECTRIX_API_KEY", "zt_3266e9b843768257893ff95e4bf4d740") + +# 目标设备 ID（MAC）。留空则自动取设备列表的第一个。 +DEVICE_ID = os.environ.get("ZECTRIX_DEVICE_ID", "") + +# 推送参数 +PAGE_ID = "1" # 持久化页面编号 1-5 +DITHER = False # 纯黑白点阵图用硬阈值最锐利；设备无真灰阶，勿开抖动 + +# ---- 画布 ---- +WIDTH, HEIGHT = 400, 300 + +# ---- 资源路径 ---- +FONT_PATH = os.path.join(BASE_DIR, "fonts", "fusion-pixel-12px-proportional-zh_hans.otf") +OUTPUT_DIR = os.path.join(BASE_DIR, "output") +OUTPUT_PATH = os.path.join(OUTPUT_DIR, "dashboard.png") + +# ---- 橘喵 jm-devops 后端（今日经营 / 近30天趋势 数据源，免鉴权 GET 调用）---- +BASE_API = os.environ.get("JM_BASE_API", "https://admin.jumiaotandian.com/prod-api") +CLIENT_ID = os.environ.get("JM_CLIENT_ID", "e5cd7e4891bf95d1d19206ce24a7b32e") + +# ---- 和风天气（QWeather）数据源 ---- +# 专属 API Host 与 Key 在 https://console.qweather.com/setting 查看，须用环境变量配置； +# 未配置时天气区显示占位 "--"，不影响其余渲染与推送。 +QWEATHER_HOST = os.environ.get("QWEATHER_HOST", "n57fc3yv6r.re.qweatherapi.com") # 用户专属 API Host +QWEATHER_KEY = os.environ.get("QWEATHER_KEY", "b27f545219a34d42adb54bbf2c5b81dd") +# 定位固定为南京·秦淮区。和风经纬度格式为「经度,纬度」 +WEATHER_LOCATION = os.environ.get("WEATHER_LOCATION", "118.788,32.014") +WEATHER_LOC_NAME = "南京·秦淮" +# 天气缓存：业务数据每分钟刷新，但天气 15 分钟才重新拉一次（cron 每次独立进程，缓存须落盘）。 +WEATHER_TTL = int(os.environ.get("WEATHER_TTL", "900")) # 秒，默认 900=15 分钟 +WEATHER_CACHE_PATH = os.path.join(BASE_DIR, ".weather_cache.json") +# 近30天趋势缓存：recent-days 每天只更新一次，按日期为 key 缓存，当天命中即不再请求。 +TREND_CACHE_PATH = os.path.join(BASE_DIR, ".trend_cache.json") diff --git a/data.py b/data.py new file mode 100644 index 0000000..6bd7ec1 --- /dev/null +++ b/data.py @@ -0,0 +1,180 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +数据层 —— 汇总仪表盘所需的全部数据。 + +橘喵今日经营 + 近30天趋势来自 jm_api（jm-devops 后端统计接口，免鉴权 GET）。 +天气来自 weather_api（和风天气，定位南京·秦淮）。日期/时间用真实系统时间。 +任何接口失败时相关数据以占位符 "--" 呈现，不回退 mock。 +""" +import json +from datetime import datetime + +import config +import jm_api +import weather_api + +WEEK_CN = ["周一", "周二", "周三", "周四", "周五", "周六", "周日"] + +TREND_DAYS = 30 # 近 N 天趋势 + + +def _read_weather_cache(): + try: + with open(config.WEATHER_CACHE_PATH, encoding="utf-8") as f: + return json.load(f) # {"ts": float, "data": {...}} + except Exception: + return None + + +def _write_weather_cache(ts, data): + try: + with open(config.WEATHER_CACHE_PATH, "w", encoding="utf-8") as f: + json.dump({"ts": ts, "data": data}, f, ensure_ascii=False) + except Exception as e: + print(f"[警告] 天气缓存写入失败：{e}") + + +def _read_trend_cache(): + try: + with open(config.TREND_CACHE_PATH, encoding="utf-8") as f: + return json.load(f) # {"date": "YYYY-MM-DD", "data": {...}} + except Exception: + return None + + +def _write_trend_cache(date_key, data): + try: + with open(config.TREND_CACHE_PATH, "w", encoding="utf-8") as f: + json.dump({"date": date_key, "data": data}, f, ensure_ascii=False) + except Exception as e: + print(f"[警告] 趋势缓存写入失败：{e}") + + +def _trend(now): + """近30天趋势，按日期缓存：recent-days 每天只更新一次。 + + 当天已有缓存（且有数据）→ 直接复用，不发请求；当天还没有 → 拉取并写缓存。 + 拉取失败时沿用任意旧缓存（哪怕非当天），仍无缓存才回退空。 + API 降序 → 升序：左旧右新。 + """ + date_key = now.strftime("%Y-%m-%d") + cached = _read_trend_cache() + if cached and cached.get("date") == date_key and cached.get("data", {}).get("series"): + return cached["data"] + try: + asc = list(reversed(jm_api.get_recent_days(TREND_DAYS))) + trend = { + "dates": [d["date"][5:].replace("-", "/") for d in asc], # yyyy-MM-dd → MM/DD + "series": [ + {"k": "单量", "style": "solid", "data": [int(d["orderCount"]) for d in asc]}, + {"k": "流水", "style": "dashed", "data": [float(d["payAmount"]) for d in asc]}, + {"k": "毛利", "style": "dotted", "data": [float(d["grossProfit"]) for d in asc]}, + ], + } + _write_trend_cache(date_key, trend) + return trend + except Exception as e: + print(f"[警告] 近30天趋势拉取失败：{e}") + if cached and cached.get("data", {}).get("series"): # 失败沿用旧缓存，避免趋势图整片 "--" + print("[警告] 沿用上次趋势缓存") + return cached["data"] + return {"dates": [], "series": []} + + +def _weather(now): + """和风天气（南京·秦淮），带 WEATHER_TTL 文件缓存。 + + 缓存未过期 → 直接复用，不发请求（每分钟刷新时省掉 14/15 次天气调用）。 + 过期 → 重新拉取并写缓存；全部接口失败时沿用上次缓存，仍无缓存才回退占位 "--"。 + """ + ts = now.timestamp() + cached = _read_weather_cache() + if cached and ts - cached.get("ts", 0) < config.WEATHER_TTL: + return cached["data"] + out = {"loc": config.WEATHER_LOC_NAME, "cond": "--", "temp": "--", "range": "--", "icon": ""} + ok = False + try: + w = weather_api.get_now() + out["cond"] = w.get("text") or "--" + out["temp"] = f"{w['temp']}°" + out["icon"] = w.get("icon") or "" # 和风图标代码，供 renderer 映射图标 + ok = True + except Exception as e: + print(f"[警告] 实时天气拉取失败：{e}") + try: + today = weather_api.get_today() + out["range"] = f"{today['tempMin']}°~{today['tempMax']}°" + ok = True + except Exception as e: + print(f"[警告] 今日温区拉取失败：{e}") + if ok: + _write_weather_cache(ts, out) + return out + if cached: # 全部失败：沿用上次缓存，避免天气区整片 "--" + print("[警告] 天气全部拉取失败，沿用上次缓存") + return cached["data"] + return out + + +# ---------- 格式化 ---------- +def _fmt_count(n): + return f"{int(n):,}" # 千分位，如 1,284 + + +def _fmt_money(yuan): + """元金额 → 紧凑展示：≥1万显示「X.XX万」(两位小数)，否则取整元。""" + v = float(yuan) + return f"{v / 10000:.2f}万" if abs(v) >= 10000 else f"{v:.0f}" + + +def _fmt_rate(rate): + """增长率字符串 → (方向, 文本)。null/空 → (None, '--')，方向 None 表示不画三角。 + 数值统一两位小数；整数部分不足两位用前导 0 补齐(6.5→06.50)，便于环比/同比上下对齐； + 三/四位数(≥100%)原样完整输出不截断。""" + if rate is None or rate == "": + return (None, "--") + v = float(rate) + return ("down" if v < 0 else "up", f"{abs(v):05.2f}%") + + +def _col(k, v, wow, yoy): + return {"k": k, "v": v, "hb": _fmt_rate(wow), "tb": _fmt_rate(yoy)} + + +def _placeholder_col(k): + return {"k": k, "v": "--", "hb": (None, "--"), "tb": (None, "--")} + + +# ---------- 橘喵业务数据 ---------- +def _mao(now): + upd = now.strftime("%H:%M") + # 今日经营（realtime） + try: + rt = jm_api.get_realtime() + upd = (rt.get("dataTime") or "")[11:16] or upd + cols = [ + _col("单量", _fmt_count(rt["todayOrderCount"]), rt["orderCountWowRate"], rt["orderCountYoyRate"]), + _col("流水", _fmt_money(rt["todayPayAmount"]), rt["payAmountWowRate"], rt["payAmountYoyRate"]), + _col("毛利", _fmt_money(rt["todayGrossProfit"]), rt["grossProfitWowRate"], rt["grossProfitYoyRate"]), + ] + except Exception as e: + print(f"[警告] 今日经营拉取失败：{e}") + cols = [_placeholder_col(k) for k in ("单量", "流水", "毛利")] + # 近30天趋势（按日期缓存，每天只拉一次） + trend = _trend(now) + return {"upd": upd, "cols": cols, "trend": trend} + + +def get_dashboard_data(now=None): + """返回渲染所需的完整数据字典。""" + now = now or datetime.now() + return { + "date": { + "greg": now.strftime("%Y/%m/%d"), + "week": WEEK_CN[now.weekday()], + "time": now.strftime("%H:%M"), + }, + "weather": _weather(now), + "mao": _mao(now), + } diff --git a/fonts/fusion-pixel-12px-proportional-zh_hans.otf b/fonts/fusion-pixel-12px-proportional-zh_hans.otf new file mode 100644 index 0000000..b4b45cf Binary files /dev/null and b/fonts/fusion-pixel-12px-proportional-zh_hans.otf differ diff --git a/jm_api.py b/jm_api.py new file mode 100644 index 0000000..71d4652 --- /dev/null +++ b/jm_api.py @@ -0,0 +1,43 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +橘喵 jm-devops 后端客户端 —— 直接调用免鉴权的统计接口（无需登录/token）。 + +接口（参考 jm-devops 首页 src/api/dashboard.ts）： + - GET /system/statistics/realtime 今日经营实时数据 + - GET /system/statistics/recent-days?days 近 N 天每日数据（按日期降序、不含今天） +""" +import json +import urllib.request +import urllib.error + +import config + + +def _get(path): + url = config.BASE_API + path + headers = {"Content-Type": "application/json;charset=utf-8", "clientid": config.CLIENT_ID} + req = urllib.request.Request(url, headers=headers, method="GET") + try: + with urllib.request.urlopen(req, timeout=30) as resp: + data = json.loads(resp.read().decode("utf-8")) + except urllib.error.HTTPError as e: + raise RuntimeError(f"HTTP {e.code}: {e.read().decode('utf-8', 'ignore')[:200]}") from e + if data.get("code") != 200: + raise RuntimeError(f"{path} 返回 code={data.get('code')}: {data.get('msg')}") + return data["data"] + + +def get_realtime() -> dict: + """今日经营实时数据（单量/流水/毛利 + 环比/同比）。""" + return _get("/system/statistics/realtime") + + +def get_recent_days(days=30) -> list: + """近 N 天每日经营数据（API 按日期降序、不含今天）。""" + return _get(f"/system/statistics/recent-days?days={days}")["days"] + + +if __name__ == "__main__": + print("realtime:", json.dumps(get_realtime(), ensure_ascii=False)[:300]) + print("recent[0]:", json.dumps(get_recent_days(30)[0], ensure_ascii=False)) diff --git a/main.py b/main.py new file mode 100644 index 0000000..de6f7c0 --- /dev/null +++ b/main.py @@ -0,0 +1,37 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +入口 —— 取数据 → 渲染 → 推送到设备。 + +用法： + python3 main.py # 渲染并推送（默认取设备列表第一个） + python3 main.py --render-only # 只渲染生成图片，不推送（output/dashboard.png） + +可用环境变量覆盖配置：ZECTRIX_API_KEY、ZECTRIX_DEVICE_ID +""" +import argparse + +import config +from data import get_dashboard_data +from renderer import DashboardRenderer + + +def main(): + parser = argparse.ArgumentParser(description="墨水屏仪表盘：渲染并推送") + parser.add_argument("--render-only", action="store_true", help="只渲染，不推送") + args = parser.parse_args() + + data = get_dashboard_data() + path = DashboardRenderer().render_to_file(data, config.OUTPUT_PATH) + print(f"[渲染] 已生成 {path}") + + if args.render_only: + return + + from pusher import push_image + device_id, resp = push_image(path) + print(f"[推送] 设备 {device_id} 成功：{resp.get('data')}") + + +if __name__ == "__main__": + main() diff --git a/pusher.py b/pusher.py new file mode 100644 index 0000000..c8c6ff5 --- /dev/null +++ b/pusher.py @@ -0,0 +1,76 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +推送层 —— 调用 Zectrix 极趣云平台 API（仅用标准库 urllib，无第三方依赖）。 +文档：显示推送 / 设备管理。 +""" +import json +import uuid +import urllib.request +import urllib.error + +import config + + +def _request(method, path, headers=None, body=None): + url = f"{config.API_BASE}{path}" + h = {"X-API-Key": config.API_KEY} + if headers: + h.update(headers) + req = urllib.request.Request(url, data=body, headers=h, method=method) + try: + with urllib.request.urlopen(req, timeout=30) as resp: + return json.loads(resp.read().decode("utf-8")) + except urllib.error.HTTPError as e: + raise RuntimeError(f"HTTP {e.code}: {e.read().decode('utf-8', 'ignore')}") from e + + +def get_devices(): + """获取设备列表。""" + data = _request("GET", "/devices") + if data.get("code") != 0: + raise RuntimeError(f"获取设备失败: {data}") + return data.get("data", []) + + +def resolve_device_id(): + """返回目标设备 ID：优先用 config.DEVICE_ID，否则取列表第一个。""" + if config.DEVICE_ID: + return config.DEVICE_ID + devices = get_devices() + if not devices: + raise RuntimeError("账号下没有任何设备") + return devices[0]["deviceId"] + + +def _multipart(fields, files): + """构造 multipart/form-data 请求体。fields: dict[str,str]; files: list[(name, filename, bytes, mime)]""" + boundary = uuid.uuid4().hex + crlf = b"\r\n" + buf = bytearray() + for k, v in fields.items(): + buf += b"--" + boundary.encode() + crlf + buf += f'Content-Disposition: form-data; name="{k}"'.encode() + crlf + crlf + buf += str(v).encode() + crlf + for name, filename, content, mime in files: + buf += b"--" + boundary.encode() + crlf + buf += f'Content-Disposition: form-data; name="{name}"; filename="{filename}"'.encode() + crlf + buf += f"Content-Type: {mime}".encode() + crlf + crlf + buf += content + crlf + buf += b"--" + boundary.encode() + b"--" + crlf + return bytes(buf), f"multipart/form-data; boundary={boundary}" + + +def push_image(image_path, device_id=None, page_id=config.PAGE_ID, dither=config.DITHER): + """推送图片到设备显示。""" + device_id = device_id or resolve_device_id() + with open(image_path, "rb") as fp: + content = fp.read() + fields = {"dither": "true" if dither else "false", "pageId": str(page_id)} + files = [("images", "dashboard.png", content, "image/png")] + body, content_type = _multipart(fields, files) + data = _request("POST", f"/devices/{device_id}/display/image", + headers={"Content-Type": content_type}, body=body) + if data.get("code") != 0: + raise RuntimeError(f"推送失败: {data}") + return device_id, data diff --git a/renderer.py b/renderer.py new file mode 100644 index 0000000..47bd7e7 --- /dev/null +++ b/renderer.py @@ -0,0 +1,330 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +渲染层 —— 将数据绘制成 400x300 纯黑白墨水屏仪表盘。 + +设计要点： + - 纯黑白（设备无真灰阶，灰阶只能抖动成网点，弃用）；用构图 + 字号 + 留白建立层级。 + - Fusion Pixel 点阵字体，按整数倍尺寸(12/24/48)渲染，fontmode='1' 关闭抗锯齿 → 像素锐利。 + - 重点数据用「伪粗体」(偏移叠绘)加粗，仍对齐像素网格。 +布局：上=大时钟+日期 / 天气；中=橘喵今日经营三列；下=近30天三指标趋势折线图。 +""" +import math +from PIL import Image, ImageDraw, ImageFont + +import config + +BLACK, WHITE = 0, 255 + + +class DashboardRenderer: + def __init__(self, font_path=config.FONT_PATH, size=(config.WIDTH, config.HEIGHT)): + self.W, self.H = size + self.F12 = ImageFont.truetype(font_path, 12) + self.F24 = ImageFont.truetype(font_path, 24) + self.F48 = ImageFont.truetype(font_path, 48) + + # ---------- 基础绘制工具 ---------- + def _tw(self, s, ft): + return self.d.textlength(s, font=ft) + + def _draw(self, x, y, s, ft, fill, bold): + # 伪粗体：点阵字体单一字重，按 1~bold px 偏移叠绘加粗笔画 + self.d.text((x, y), s, font=ft, fill=fill) + for dx in range(1, bold + 1): + self.d.text((x + dx, y), s, font=ft, fill=fill) + + def text(self, x, y, s, ft, fill=BLACK, bold=0): + self._draw(x, y, s, ft, fill, bold) + + def rtext(self, xr, y, s, ft, fill=BLACK, bold=0): + self._draw(xr - self._tw(s, ft) - bold, y, s, ft, fill, bold) + + def ctext(self, cx, y, s, ft, fill=BLACK, bold=0): + self._draw(cx - (self._tw(s, ft) + bold) / 2, y, s, ft, fill, bold) + + def hdash(self, x0, x1, y, dash=4, gap=4, fill=BLACK): + x = x0 + while x < x1: + self.d.line([x, y, min(x + dash, x1), y], fill=fill, width=1) + x += dash + gap + + def vdash(self, x, y0, y1, dash=4, gap=4, fill=BLACK): + y = y0 + while y < y1: + self.d.line([x, y, x, min(y + dash, y1)], fill=fill, width=1) + y += dash + gap + + def tri(self, cx, cy, sz, up=True, fill=BLACK): + if up: + pts = [(cx, cy - sz / 2), (cx - sz / 2, cy + sz / 2), (cx + sz / 2, cy + sz / 2)] + else: + pts = [(cx, cy + sz / 2), (cx - sz / 2, cy - sz / 2), (cx + sz / 2, cy - sz / 2)] + self.d.polygon(pts, fill=fill) + + # ---------- 组件 ---------- + def _cmp_line(self, cx, y, label, pair, fg=BLACK): + """环比/同比一行：标签 + 涨跌三角 + 数值，整体居中。direction 为 None 时不画三角。""" + direction, val = pair + lw = self._tw(label, self.F12); gap = 5; nw = self._tw(val, self.F12) + trw, tgap = (8, 4) if direction else (0, 0) + x = cx - (lw + gap + trw + tgap + nw) / 2 + self.text(x, y, label, self.F12, fill=fg); x += lw + gap + if direction: + # 三角中心对齐 F12 数字墨迹竖直中心(y+8.5)，否则箭头偏高 + self.tri(x + trw / 2, y + 8.5, 7, up=(direction == "up"), fill=fg); x += trw + tgap + self.text(x, y, val, self.F12, fill=fg) + + def _title_bar(self, y, title, upd): + """标题栏：■ 标题（左） + 上次更新 HH:mm（右）。""" + sq, sq_y = 6, y + 4 + self.d.rectangle([10, sq_y, 10 + sq, sq_y + sq], fill=BLACK) + self.text(10 + sq + 6, y, title, self.F12) + self.rtext(self.W - 10, y, f"上次更新 {upd}", self.F12) + + def _loc_pin(self, x, y, w=8, fg=BLACK, bg=WHITE): + """定位图标：实心水滴(圆头+尖尾) + 白色挖孔，1-bit 下轮廓清晰。 + (x,y) 为头部圆左上角，w 为头径，总高约 w*1.5。""" + cx, cy = x + w / 2, y + w / 2 # 头部圆心 + self.d.ellipse([x, y, x + w, y + w], fill=fg) # 头部实心圆 + self.d.polygon([(x, cy), (x + w, cy), (cx, y + w + w * 0.5)], fill=fg) # 尾部尖 + hr = w * 0.2 + self.d.ellipse([cx - hr, cy - hr, cx + hr, cy + hr], fill=bg) # 头部白孔 + + @staticmethod + def _weather_category(icon): + """和风天气 icon 代码 → 可枚举类别。无法识别时回退 'cloudy'。 + 参考 https://dev.qweather.com/docs/resource/icons/ ： + 100=晴日 150=晴夜 / 101~103,151~153=多云 / 104,154=阴 / + 302~304=雷阵雨 / 300~399=雨 / 400~499=雪 / 500~599=雾霾沙尘。""" + try: + c = int(icon) + except (TypeError, ValueError): + return "cloudy" + if c == 100: + return "sunny" + if c == 150: + return "clear_night" + if c in (101, 102, 103, 151, 152, 153): + return "cloudy" + if c in (104, 154): + return "overcast" + if c in (302, 303, 304): + return "thunder" + if 300 <= c <= 399: + return "rain" + if 400 <= c <= 499: + return "snow" + if 500 <= c <= 599: + return "fog" + return "cloudy" + + def _sun(self, cx, cy, r, s, fg, rays=True): + """太阳：圆 + 8 道光芒（光芒可关）。""" + S = lambda v: v * s + self.d.ellipse([cx - r, cy - r, cx + r, cy + r], outline=fg, width=2) + if rays: + for a in range(0, 360, 45): + rad = math.radians(a) + self.d.line([cx + math.cos(rad) * (r + S(3)), cy + math.sin(rad) * (r + S(3)), + cx + math.cos(rad) * (r + S(7)), cy + math.sin(rad) * (r + S(7))], fill=fg, width=2) + + def _cloud(self, x, cy, s, fg, bg): + """一朵云，cy 为云体中线高度；bg 填充内部以盖住其后的太阳光芒。""" + S = lambda v: v * s + self.d.ellipse([x + S(4), cy - S(2), x + S(20), cy + S(12)], fill=bg) + self.d.ellipse([x + S(14), cy - S(8), x + S(34), cy + S(12)], fill=bg) + self.d.rectangle([x + S(6), cy + S(4), x + S(34), cy + S(12)], fill=bg) + self.d.arc([x + S(2), cy - S(4), x + S(22), cy + S(14)], 90, 270, fill=fg, width=2) + self.d.arc([x + S(12), cy - S(10), x + S(36), cy + S(14)], 270, 90, fill=fg, width=2) + self.d.line([x + S(9), cy + S(13), x + S(29), cy + S(13)], fill=fg, width=2) + + def _weather_icon(self, x, y, category, s=1.0, fg=BLACK, bg=WHITE): + """按天气类别画图标，s 为缩放系数，原生约 36x38（左上角 x,y）。""" + S = lambda v: v * s + + if category == "sunny": + self._sun(x + S(18), y + S(19), S(11), s, fg) + return + + if category == "clear_night": # 月牙：实心圆挖去偏上右的背景圆 + mx, my, r = x + S(18), y + S(19), S(13) + self.d.ellipse([mx - r, my - r, mx + r, my + r], fill=fg) + ox, oy, r2 = mx + S(6), my - S(5), S(12) + self.d.ellipse([ox - r2, oy - r2, ox + r2, oy + r2], fill=bg) + return + + if category == "overcast": # 阴：单独一朵云，居中偏下 + self._cloud(x, y + S(20), s, fg, bg) + return + + if category == "fog": # 雾/霾/沙尘：横向雾线，长短交错 + for i, dy in enumerate((9, 17, 25, 33)): + inset = S(5) if i % 2 else 0 + self.d.line([x + inset, y + S(dy), x + S(34) - inset, y + S(dy)], fill=fg, width=2) + return + + if category != "rain" and category != "snow" and category != "thunder": # 多云及未识别：太阳 + 云 + self._sun(x + S(10), y + S(9), S(6), s, fg) + self._cloud(x, y + S(24), s, fg, bg) + return + + # 以下三类：云在上，降水/天气符号在云下方 + cb = y + S(15) + S(13) # 云底 y + self._cloud(x, y + S(15), s, fg, bg) + if category == "rain": + for dx in (9, 18, 27): + self.d.line([x + S(dx) + S(3), cb + S(3), x + S(dx) - S(1), cb + S(10)], fill=fg, width=2) + elif category == "snow": + for dx in (10, 19, 28): + fx, fy, rr = x + S(dx), cb + S(7), S(4) + for a in (0, 60, 120): + rad = math.radians(a) + self.d.line([fx - math.cos(rad) * rr, fy - math.sin(rad) * rr, + fx + math.cos(rad) * rr, fy + math.sin(rad) * rr], fill=fg, width=1) + elif category == "thunder": + bx = x + S(18) + self.d.polygon([(bx - S(1), cb + S(2)), (bx + S(5), cb + S(2)), (bx + S(1), cb + S(8)), + (bx + S(6), cb + S(8)), (bx - S(4), cb + S(18)), (bx, cb + S(9)), + (bx - S(4), cb + S(9))], fill=fg) + + # 线型样式：(实线段长, 间隙长)；实线用 dash=0 表示连续 + _STYLES = {"solid": (0, 0), "dashed": (7, 4), "dotted": (1, 2)} + + def _styled_polyline(self, pts, style, fill=BLACK): + """按线型(solid/dashed/dotted)沿折线绘制。点线/虚线靠等距步进采样。""" + dash, gap = self._STYLES.get(style, (0, 0)) + if dash == 0: # 实线：直接连线 + for (x0, y0), (x1, y1) in zip(pts, pts[1:]): + self.d.line([x0, y0, x1, y1], fill=fill, width=1) + return + period, drawn = dash + gap, 0.0 # drawn: 当前所在周期内已走过的距离 + for (x0, y0), (x1, y1) in zip(pts, pts[1:]): + seg = math.hypot(x1 - x0, y1 - y0) + if seg == 0: + continue + ux, uy = (x1 - x0) / seg, (y1 - y0) / seg + t = 0.0 + while t < seg: + phase = drawn % period + if phase < dash: # 处于“实”阶段 → 画一小段 + step = min(dash - phase, seg - t) + sx, sy = x0 + ux * t, y0 + uy * t + ex, ey = x0 + ux * (t + step), y0 + uy * (t + step) + if dash <= 1: # 点线：画方点更醒目 + self.d.point((round(sx), round(sy)), fill=fill) + else: + self.d.line([sx, sy, ex, ey], fill=fill, width=1) + t += step; drawn += step + else: # 处于“虚”阶段 → 跳过 + step = min(period - phase, seg - t) + t += step; drawn += step + + def _style_sample(self, x, y, style, w=22, fill=BLACK): + """图例用的线型样例（一小段水平线）。""" + self._styled_polyline([(x, y), (x + w, y)], style, fill=fill) + + def _trend_chart(self, y0, trend): + """近30天三指标折线图：标题(左) + 紧凑图例(右) + 折线 + X轴(仅月/日)。 + 各指标按自身极值独立归一化，故只表达趋势形状，不可横向比绝对值。""" + dates, series = trend["dates"], trend["series"] + leg_y = y0 + # ---- 标题（左）：■ 近30天趋势 ---- + sq, sq_y = 6, leg_y + 4 + self.d.rectangle([10, sq_y, 10 + sq, sq_y + sq], fill=BLACK) + self.text(10 + sq + 6, leg_y, "橘喵·30天趋势", self.F12) + # 无数据：画占位符 "--"，跳过图例/折线/X轴 + if not series or not dates or not any(s.get("data") for s in series): + self.ctext(self.W / 2, leg_y + 36, "--", self.F24) + return + # ---- 图例（右）：紧凑、整体右对齐，从右往左逐项排布 ---- + SAMP, G1, GAP = 16, 8, 10 # 样例线长 / 线与名间隙 / 图例项间隙 + x = self.W - 10 + for s in reversed(series): + self.rtext(x, leg_y, s["k"], self.F12) # 指标名右端对齐 x + x -= self._tw(s["k"], self.F12) + G1 + self._style_sample(x - SAMP, leg_y + 7, s["style"]) + x -= SAMP + GAP + # ---- 绘图区 ---- + px0, px1 = 14, self.W - 10 + py0, py1 = leg_y + 22, leg_y + 22 + 56 # 顶/底 + self.hdash(px0, px1, py1, dash=2, gap=3) # 基线 + n = len(dates) + xstep = (px1 - px0) / (n - 1) if n > 1 else 0 + for s in series: + data = s["data"] + lo, hi = min(data), max(data) + span = (hi - lo) or 1 + pad = 4 # 上下留白，避免贴边 + pts = [] + for i, v in enumerate(data): + x = px0 + xstep * i + yv = py1 - pad - (v - lo) / span * (py1 - py0 - 2 * pad) + pts.append((x, yv)) + self._styled_polyline(pts, s["style"]) + # ---- X 轴标签：仅首 / 中 / 末三个日期 ---- + xl_y = py1 + 3 + self.text(px0, xl_y, dates[0], self.F12) + self.ctext((px0 + px1) / 2, xl_y, dates[n // 2], self.F12) + self.rtext(px1, xl_y, dates[-1], self.F12) + + # ---------- 入口 ---------- + def render(self, data): + """根据 data 渲染并返回 PIL.Image（'L' 模式，纯 0/255）。""" + self.img = Image.new("L", (self.W, self.H), WHITE) + self.d = ImageDraw.Draw(self.img) + self.d.fontmode = "1" # 关闭抗锯齿，点阵锐利 + + dt, wt, mao = data["date"], data["weather"], data["mao"] + + # A. Hero：大时钟 + 日期（时钟上移使「顶部留白 ≈ 与日期间距」对称） + self.text(10, 0, dt["time"], self.F48, bold=2) + self.text(12, 64, f'{dt["week"]} {dt["greg"]}', self.F12) + + # B. 天气：右对齐文字 + 居中的 28° + 放大右靠的图标 + DIV_Y = 90 # 天气区底部分割线 + loc, temp, cr = wt["loc"], wt["temp"], f'{wt["cond"]} {wt["range"]}' + self.rtext(390, 8, loc, self.F12) + self.rtext(390, 30, temp, self.F24, bold=1) + self.rtext(390, 64, cr, self.F12) + # 定位图标：置于 loc 文字左侧，留间距并与文字竖直居中 + PIN_W, PIN_GAP = 8, 6 + loc_pin_x = 390 - self._tw(loc, self.F12) - PIN_GAP - PIN_W + self._loc_pin(loc_pin_x, 10, PIN_W) + s = 1.5 + iw, ih = 36 * s, 38 * s # 图标原生约 36x38 + text_left = min(loc_pin_x, + 390 - self._tw(temp, self.F24) - 1, + 390 - self._tw(cr, self.F12)) + icon_y = (DIV_Y - ih) / 2 # 顶部(0)与分割线正中：上下留白相等 + self._weather_icon(text_left - 12 - iw, icon_y, self._weather_category(wt.get("icon")), s=s) + + self.hdash(8, 392, DIV_Y) + + # C. 橘喵今日经营：三列（数值 + 环比/同比） + self._title_bar(96, "橘喵·今日实时", mao["upd"]) + cols = mao["cols"]; n = len(cols) + ax0, ax1 = 8, 392 + cw = (ax1 - ax0) / n + for i in range(1, n): + self.vdash(ax0 + cw * i, 128, 192) + for i, c in enumerate(cols): + cx = ax0 + cw * (i + 0.5) + self.ctext(cx, 114, c["k"], self.F12) + self.ctext(cx, 130, c["v"], self.F24, bold=1) + self._cmp_line(cx, 160, "环比", c["hb"]) + self._cmp_line(cx, 178, "同比", c["tb"]) + + self.hdash(8, 392, 200) + + # D. 近30天三指标趋势折线图 + self._trend_chart(204, mao["trend"]) + + return self.img + + def render_to_file(self, data, path): + img = self.render(data) + import os + os.makedirs(os.path.dirname(path), exist_ok=True) + img.save(path) + return path diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..750ca02 --- /dev/null +++ b/requirements.txt @@ -0,0 +1 @@ +Pillow>=10.0 diff --git a/weather_api.py b/weather_api.py new file mode 100644 index 0000000..62f6227 --- /dev/null +++ b/weather_api.py @@ -0,0 +1,55 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +和风天气（QWeather）客户端 —— 实时天气 + 今日温区。 + +和风要求使用「用户专属 API Host」（控制台-设置查看 https://console.qweather.com/setting）， +用 API Key 鉴权（请求头 X-QW-Api-Key）；响应体 Gzip 压缩需手动解压，且返回 code 为字符串。 +免费订阅额度足够墨水屏每日推送。HOST/KEY 通过环境变量配置，未配置时调用方回退占位。 + +接口（参考 https://dev.qweather.com/docs/api/）： + - GET /v7/weather/now?location=经度,纬度 实时天气（now.text 中文天气 / now.temp 温度） + - GET /v7/weather/3d?location=经度,纬度 逐日预报，取 daily[0] 的今日 tempMin/tempMax +""" +import gzip +import json +import urllib.request +import urllib.error + +import config + + +def _get(path): + if not config.QWEATHER_HOST or not config.QWEATHER_KEY: + raise RuntimeError("未配置 QWEATHER_HOST / QWEATHER_KEY") + host = config.QWEATHER_HOST.rstrip("/") + if not host.startswith("http"): + host = "https://" + host + headers = {"X-QW-Api-Key": config.QWEATHER_KEY, "Accept-Encoding": "gzip"} + req = urllib.request.Request(host + path, headers=headers, method="GET") + try: + with urllib.request.urlopen(req, timeout=30) as resp: + raw = resp.read() + except urllib.error.HTTPError as e: + raise RuntimeError(f"HTTP {e.code}: {e.read().decode('utf-8', 'ignore')[:200]}") from e + if raw[:2] == b"\x1f\x8b": # gzip magic number，和风默认压缩 + raw = gzip.decompress(raw) + data = json.loads(raw.decode("utf-8")) + if str(data.get("code")) != "200": # 和风 code 为字符串 + raise RuntimeError(f"{path} 返回 code={data.get('code')}") + return data + + +def get_now() -> dict: + """实时天气：now.text（中文天气）/ now.temp（摄氏温度）。""" + return _get(f"/v7/weather/now?location={config.WEATHER_LOCATION}")["now"] + + +def get_today() -> dict: + """今日逐日预报：daily[0] 含 tempMin / tempMax。""" + return _get(f"/v7/weather/3d?location={config.WEATHER_LOCATION}")["daily"][0] + + +if __name__ == "__main__": + print("now: ", json.dumps(get_now(), ensure_ascii=False)[:300]) + print("today:", json.dumps(get_today(), ensure_ascii=False)[:300])