#!/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 METRIC_LABELS = { "service_ready": {"zh": "服务就绪", "en": "Service ready"}, "device_online": {"zh": "画框在线", "en": "Frame online"}, "battery_percent": {"zh": "电量", "en": "Battery"}, "job_enabled": {"zh": "任务启用", "en": "Job enabled"}, "last_run_success": {"zh": "最近运行成功", "en": "Last run succeeded"}, "device_count": {"zh": "画框数量", "en": "Frame count"}, "job_count": {"zh": "任务数量", "en": "Job count"}, } 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, "label": METRIC_LABELS[key], } ) def alert(identifier): kind = identifier.split(":", 1)[0] labels = { "SERVICE_NOT_READY": {"zh": "服务未就绪", "en": "Service not ready"}, "DEVICE_OFFLINE": {"zh": "画框离线", "en": "Frame offline"}, "JOB_FAILED": {"zh": "任务最近运行失败", "en": "Last job run failed"}, "JOB_UNKNOWN": { "zh": "任务最近结果待确认", "en": "Last job result unconfirmed", }, } alerts.append({"id": identifier, "severity": 1, "label": labels[kind]}) 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()