commit e5331caa41a155c986a3c373850a8fcf3d4441ce Author: 渡口浪人 Date: Sat Oct 17 22:51:01 2026 +0800 Initial Frame Flow read-only Console adapter diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..975ea0a --- /dev/null +++ b/.gitignore @@ -0,0 +1,7 @@ +.venv/ +__pycache__/ +.pytest_cache/ +.ruff_cache/ +dist/ +*.egg-info/ +.DS_Store diff --git a/LICENSE b/LICENSE new file mode 100644 index 0000000..61d52b2 --- /dev/null +++ b/LICENSE @@ -0,0 +1,21 @@ +MIT License + +Copyright (c) 2026 Pigway + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/README.md b/README.md new file mode 100644 index 0000000..53b346b --- /dev/null +++ b/README.md @@ -0,0 +1,32 @@ +# PIGWay Console Adapter — Frame Flow + +独立 ZIP 插件 `pigway-frame-flow` v1.0.0，采用现有 Console adapter protocol v1，仅依赖 Python 标准库。源码在独立仓库 `pigway-console-adapter-frame-flow` 维护，无需修改 Console 核心或部署额外服务。 + +## 安装 + +1. 在 Console「机器管理」上传 `dist/pigway-frame-flow-1.0.0.zip`，启用插件。 +2. 添加设备，选择 PIGWay Frame Flow，填写 Console 能访问的 Frame Flow HTTP(S) 根地址（不含 `/api/v1`）和 **Frame Flow API 令牌**，不是 KoKonna Key 或网页登录密码。 +3. 若使用受限 API 客户端，授予 `devices:read`、`jobs:read`；默认 `FRAME_FLOW_TOKEN` 也可使用。令牌保存于 Console 敏感字段。跨不可信网络使用 HTTPS。 +4. 查看通用资源看板、告警与历史；默认每 60 秒采集一次。无需自定义前端。 + +采集 `/health`、`/ready`、`/api/v1/devices`、`/api/v1/jobs`，不发送写请求、不刷新画框、不执行任务。设备接口可能按 Frame Flow 现有缓存规则查询厂商只读状态。 + +服务：就绪状态、设备数、任务数。画框：在线状态、可用时的电量。任务：启用状态和最近终态结果。`FAILED` 与 `UNKNOWN` 分别告警；未运行或执行中不当作成功。停用任务仍显示其最近失败，成功后恢复。设备离线、服务未就绪分别告警。 + +`/ready` 的 HTTP 503 作为未就绪状态采集；认证、超时、格式错误或其他采集失败交给 Console 保留旧快照并标记未更新，不伪造离线和零值。完整成功快照才确认告警恢复。历史从安装后的采样开始；设备指标是读取时观察到的 API 状态，不证明物理屏幕已刷新。Mock 服务返回的也是 Mock 设备状态。 + +## 本地验证与打包 + +在本仓库根目录执行（运行插件无需安装第三方依赖）： + +```sh +python3 -m venv .venv +.venv/bin/python -m pip install pytest ruff +.venv/bin/python -m pytest -q +mkdir -p dist +.venv/bin/python package.py plugin dist/pigway-frame-flow-1.0.0.zip +``` + +开发时可将 `PIGWAY_ADAPTER_PATH` 指向本目录 `plugin` 的绝对路径。ZIP 安装默认停用；生产安装与设备登记需在 Console 中执行。本次验证为本地模拟接口、RPC 和隔离安装，不代表已接入生产。 + +Frame Flow API 契约测试为可选集成验证，需要另行安装 Frame Flow 及其依赖。例如相邻项目保留现有虚拟环境时，执行 `PYTHONPATH=../pigway-frame-flow ../pigway-frame-flow/.venv/bin/python -m pytest -q`。独立环境未安装 Frame Flow 时跳过该项，其余测试正常运行。 diff --git a/package.py b/package.py new file mode 100644 index 0000000..7cab172 --- /dev/null +++ b/package.py @@ -0,0 +1,29 @@ +#!/usr/bin/env python3 +"""Package an independently maintained adapter without Console internals.""" + +import argparse +import json +from pathlib import Path +from zipfile import ZIP_DEFLATED, ZipFile + +parser = argparse.ArgumentParser() +parser.add_argument("directory", type=Path) +parser.add_argument("output", type=Path) +args = parser.parse_args() +root = args.directory.resolve() +manifest = json.loads((root / "manifest.json").read_text()) +entry = (root / manifest["entrypoint"]).resolve() +if not entry.is_relative_to(root) or not entry.is_file(): + parser.error("invalid adapter entrypoint") +with ZipFile(args.output, "w", ZIP_DEFLATED) as archive: + for path in sorted(root.rglob("*")): + if path.is_symlink(): + parser.error("symlinks are not supported") + if ( + path.is_file() + and path.resolve() != args.output.resolve() + and not any(part in (".git", "__pycache__") for part in path.parts) + and path.suffix != ".pyc" + ): + archive.write(path, path.relative_to(root)) +print("Adapter package created:", args.output) diff --git a/plugin/manifest.json b/plugin/manifest.json new file mode 100644 index 0000000..8ebee97 --- /dev/null +++ b/plugin/manifest.json @@ -0,0 +1,34 @@ +{ + "id": "pigway-frame-flow", + "name": "PIGWay Frame Flow", + "version": "1.0.0", + "protocol_version": 1, + "runtime": "python3", + "entrypoint": "worker.py", + "capabilities": [ + "status" + ], + "collection": { + "interval_seconds": 60 + }, + "connection_schema": { + "url": { + "type": "string", + "required": true, + "format": "http-origin", + "label": { + "zh": "Frame Flow 地址", + "en": "Frame Flow URL" + } + }, + "token": { + "type": "string", + "required": true, + "secret": true, + "label": { + "zh": "Frame Flow API 令牌", + "en": "Frame Flow API token" + } + } + } +} diff --git a/plugin/worker.py b/plugin/worker.py new file mode 100644 index 0000000..e9806e4 --- /dev/null +++ b/plugin/worker.py @@ -0,0 +1,207 @@ +#!/usr/bin/env python3 +"""Read-only Frame Flow monitoring via Console adapter protocol v1.""" + +import concurrent.futures +import json +import sys +import threading +import time +from pathlib import Path +from urllib.error import HTTPError +from urllib.parse import urlparse +from urllib.request import HTTPRedirectHandler, Request, build_opener + +MANIFEST = json.loads(Path(__file__).with_name("manifest.json").read_text()) +OUTPUT_LOCK = threading.Lock() +DEVICE_LOCKS = {} +LOCK = threading.Lock() +MAX_MESSAGE = 16 * 1024 * 1024 + + +class AdapterError(Exception): + def __init__(self, code): + self.code = code + + +class NoRedirect(HTTPRedirectHandler): + def redirect_request(self, *args, **kwargs): + return None + + +def request(device, endpoint): + if endpoint not in {"/health", "/ready", "/api/v1/devices", "/api/v1/jobs"}: + raise AdapterError("UNSUPPORTED") + origin = device.get("connection", {}).get("url", "").rstrip("/") + parsed = urlparse(origin) + if ( + parsed.scheme not in ("http", "https") + or not parsed.hostname + or parsed.username + or parsed.password + or parsed.path + or parsed.query + or parsed.fragment + ): + raise AdapterError("INVALID_CONFIG") + token = device.get("credentials", {}).get("token") + if not isinstance(token, str) or not token or any(c in token for c in "\r\n"): + raise AdapterError("INVALID_CONFIG") + headers = {"Accept": "application/json"} + if endpoint.startswith("/api/"): + headers["Authorization"] = "Bearer " + token + try: + req = Request(origin + endpoint, headers=headers, method="GET") + try: + response = build_opener(NoRedirect).open(req, timeout=3) + except HTTPError as exc: + if endpoint == "/ready" and exc.code == 503: + response = exc + else: + raise AdapterError( + "AUTH_FAILED" if exc.code in (401, 403) else "DEVICE_API_ERROR" + ) from None + with response: + raw = response.read(2 * 1024 * 1024 + 1) + if len(raw) > 2 * 1024 * 1024: + raise AdapterError("INVALID_RESPONSE") + try: + return json.loads(raw) + except ValueError: + raise AdapterError("INVALID_RESPONSE") from None + except (OSError, ValueError): + raise AdapterError("DEVICE_UNAVAILABLE") from None + + +def collect(device): + health = request(device, "/health") + ready = request(device, "/ready") + devices = request(device, "/api/v1/devices") + jobs = request(device, "/api/v1/jobs") + if ( + not isinstance(health, dict) + or health.get("status") != "ok" + or not isinstance(ready, dict) + or ready.get("status") not in ("ready", "not_ready") + ): + raise AdapterError("INVALID_RESPONSE") + resources = [{"id": "service", "type": "service", "name": "Frame Flow"}] + metrics = [] + alerts = [] + + def metric(rid, key, value, unit="boolean"): + metrics.append({"resource_id": rid, "key": key, "value": value, "unit": unit}) + + def alert(identifier): + alerts.append({"id": identifier, "severity": 1}) + + metric("service", "service_ready", int(ready["status"] == "ready")) + if ready["status"] != "ready": + alert("SERVICE_NOT_READY") + for items, kind in ((devices, "device"), (jobs, "job")): + if not isinstance(items, list) or len(items) > 2000: + raise AdapterError("INVALID_RESPONSE") + seen = set() + for item in items: + if ( + not isinstance(item, dict) + or not isinstance(item.get("id"), str) + or not item["id"] + or item["id"] in seen + ): + raise AdapterError("INVALID_RESPONSE") + seen.add(item["id"]) + rid = kind + ":" + item["id"] + resources.append( + { + "id": rid, + "parent_id": "service", + "type": kind, + "name": item.get("name") or item["id"], + } + ) + if kind == "device": + if item.get("status") not in ("online", "offline"): + raise AdapterError("INVALID_RESPONSE") + metric(rid, "device_online", int(item["status"] == "online")) + if item["status"] == "offline": + alert("DEVICE_OFFLINE:" + item["id"]) + power = item.get("power") or {} + battery = power.get("battery_percent") + if type(battery) is int and 0 <= battery <= 100: + metric(rid, "battery_percent", battery, "percent") + else: + if type(item.get("enabled")) is not bool: + raise AdapterError("INVALID_RESPONSE") + metric(rid, "job_enabled", int(item["enabled"])) + result = item.get("last_result") + if result in ("COMPLETED", "FAILED", "UNKNOWN"): + metric(rid, "last_run_success", int(result == "COMPLETED")) + if result in ("FAILED", "UNKNOWN"): + alert("JOB_" + result + ":" + item["id"]) + metric("service", kind + "_count", len(items), "count") + return { + "schema_version": 1, + "observed_at": time.time(), + "identity": {"reported_name": "Frame Flow", "category": "host"}, + "collection": MANIFEST["collection"], + "capabilities": ["status"], + "resources": resources, + "metrics": metrics, + "alerts": alerts, + "alerts_complete": True, + } + + +def dispatch(method, params): + if method in ("adapter.initialize", "adapter.describe", "adapter.health"): + return MANIFEST + if method not in ("device.validate", "device.collect"): + raise AdapterError("UNSUPPORTED") + device = params["device"] + with LOCK: + device_lock = DEVICE_LOCKS.setdefault(device["id"], threading.Lock()) + with device_lock: + snapshot = collect(device) + return {"valid": True} if method == "device.validate" else snapshot + + +def handle(message): + try: + result = dispatch(message["method"], message.get("params", {})) + reply = {"jsonrpc": "2.0", "id": message["id"], "result": result} + except AdapterError as exc: + reply = { + "jsonrpc": "2.0", + "id": message["id"], + "error": {"code": -32000, "data": exc.code}, + } + except Exception: + # Do not copy vendor exception messages: they may contain credentials or URLs. + reply = { + "jsonrpc": "2.0", + "id": message["id"], + "error": {"code": -32000, "data": "DEVICE_UNAVAILABLE"}, + } + raw = json.dumps(reply, allow_nan=False, separators=(",", ":")) + with OUTPUT_LOCK: + print(raw, flush=True) + + +def main(): + # Bound both running and queued calls even if the parent violates its limits. + slots = threading.BoundedSemaphore(8) + with concurrent.futures.ThreadPoolExecutor(max_workers=8) as pool: + while True: + line = sys.stdin.buffer.readline(MAX_MESSAGE + 1) + if not line: + break + if len(line) > MAX_MESSAGE or not line.endswith(b"\n"): + break + message = json.loads(line) + slots.acquire() + task = pool.submit(handle, message) + task.add_done_callback(lambda _: slots.release()) + + +if __name__ == "__main__": + main() diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..1415174 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,14 @@ +[project] +name = "pigway-console-adapter-frame-flow" +version = "1.0.0" +requires-python = ">=3.12" +dependencies = [] + +[project.optional-dependencies] +test = ["pytest>=8", "ruff>=0.9"] + +[tool.pytest.ini_options] +testpaths = ["tests"] + +[tool.ruff.lint] +select = ["E4", "E7", "E9", "F", "I", "B"] diff --git a/tests/test_adapter.py b/tests/test_adapter.py new file mode 100644 index 0000000..1e07454 --- /dev/null +++ b/tests/test_adapter.py @@ -0,0 +1,170 @@ +import importlib.util +import json +import subprocess +import sys +import threading +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path +from unittest.mock import patch + +import pytest + +ROOT = Path(__file__).resolve().parents[1] +spec = importlib.util.spec_from_file_location( + "frame_adapter", ROOT / "plugin/worker.py" +) +worker = importlib.util.module_from_spec(spec) +spec.loader.exec_module(worker) + + +@pytest.fixture +def endpoint(): + calls = [] + data = { + "/health": {"status": "ok"}, + "/ready": {"status": "ready", "checks": {"sqlite": True}}, + "/api/v1/devices": [ + { + "id": "frame", + "name": "Frame", + "status": "offline", + "power": {"battery_percent": 45}, + } + ], + "/api/v1/jobs": [{"id": "morning", "enabled": True, "last_result": "FAILED"}], + } + + class Handler(BaseHTTPRequestHandler): + def do_GET(self): + calls.append((self.path, self.headers.get("Authorization"))) + value = data[self.path] + self.send_response( + 503 if self.path == "/ready" and value["status"] == "not_ready" else 200 + ) + self.end_headers() + self.wfile.write(json.dumps(value).encode()) + + def log_message(self, *args): + pass + + server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) + thread = threading.Thread(target=server.serve_forever) + thread.start() + device = { + "id": "test", + "connection": {"url": f"http://127.0.0.1:{server.server_port}"}, + "credentials": {"token": "private-test-token"}, + } + yield device, data, calls + server.shutdown() + server.server_close() + thread.join() + + +def test_collect_and_not_ready(endpoint): + device, data, calls = endpoint + snapshot = worker.collect(device) + assert {a["id"] for a in snapshot["alerts"]} == { + "DEVICE_OFFLINE:frame", + "JOB_FAILED:morning", + } + assert snapshot["alerts_complete"] is True + assert len(snapshot["resources"]) == 3 + assert all( + auth == "Bearer private-test-token" + for path, auth in calls + if path.startswith("/api/") + ) + data["/ready"]["status"] = "not_ready" + assert "SERVICE_NOT_READY" in {a["id"] for a in worker.collect(device)["alerts"]} + + +def test_no_result_is_not_success(endpoint): + device, data, _ = endpoint + data["/api/v1/jobs"][0]["last_result"] = None + assert not any( + m["key"] == "last_run_success" for m in worker.collect(device)["metrics"] + ) + data["/api/v1/devices"] = {} + with pytest.raises(worker.AdapterError, match="INVALID_RESPONSE"): + worker.collect(device) + + +def test_rpc_and_write_rejected(endpoint): + device, _, _ = endpoint + messages = [ + { + "jsonrpc": "2.0", + "id": 1, + "method": "adapter.initialize", + "params": {"protocol_version": 1}, + }, + { + "jsonrpc": "2.0", + "id": 2, + "method": "device.collect", + "params": {"device": device}, + }, + { + "jsonrpc": "2.0", + "id": 3, + "method": "device.write", + "params": {"device": device}, + }, + ] + result = subprocess.run( + [sys.executable, str(ROOT / "plugin/worker.py")], + input="".join(json.dumps(m) + "\n" for m in messages), + text=True, + capture_output=True, + timeout=15, + check=True, + ) + replies = {r["id"]: r for r in map(json.loads, result.stdout.splitlines())} + assert replies[1]["result"]["id"] == "pigway-frame-flow" + assert replies[2]["result"]["schema_version"] == 1 + assert replies[3]["error"]["data"] == "UNSUPPORTED" + assert "private-test-token" not in result.stdout + result.stderr + + +def test_auth_and_redirect_are_not_followed(endpoint): + from urllib.error import HTTPError + + device, _, _ = endpoint + for status, code in ((401, "AUTH_FAILED"), (302, "DEVICE_API_ERROR")): + with patch.object(worker, "build_opener") as opener: + opener.return_value.open.side_effect = HTTPError( + "url", status, "private", {}, None + ) + with pytest.raises(worker.AdapterError, match=code): + worker.request(device, "/api/v1/jobs") + + +def test_current_frame_flow_api_contract(tmp_path): + pytest.importorskip("pigway_frame_flow") + from fastapi.testclient import TestClient + from pigway_frame_flow.api.app import create_app + from pigway_frame_flow.settings import Settings + + settings = Settings( + data_dir=tmp_path / "data", + config_dir=tmp_path / "config", + api_token="contract-token", + mock=True, + ) + app = create_app(settings, start_worker=False) + with TestClient(app) as client: + client.headers["Authorization"] = "Bearer contract-token" + + def request(device, path): + response = client.get(path) + assert response.status_code == 200 + return response.json() + + with patch.object(worker, "request", side_effect=request): + snapshot = worker.collect({}) + assert snapshot["alerts_complete"] is True + assert any( + m["key"] == "service_ready" and m["value"] == 1 for m in snapshot["metrics"] + ) + assert any(r["type"] == "device" for r in snapshot["resources"])