Blame for processor/app/routes/devices.py

"""Device heartbeat & polling API — 本地 JSON 存储(无数据库)。

设备存 data/state/devices.json;采集端轮询 /device/tasks 获取
active 项目列表(任务概念已移除,项目即采集任务);采集端「录制完成」的
增量上报存 data/state/device_task_progress.json(见 app/device_progress.py)。
"""

from __future__ import annotations

import asyncio
from datetime import datetime, timezone
from uuid import uuid4

from fastapi import APIRouter, Depends, Query

from app.localstore import (
    STATE_ROOT,
    cached_sessions_for_tasks,
    list_projects,
    scan_sessions,
)
from app import device_progress
from app.device_progress import reported_totals
from app.security import verify_api_key
from app.device_naming import decorate_device_sources, display_names_for_sources

router = APIRouter(prefix="/api/v1", tags=["devices"])

OFFLINE_THRESHOLD_SECONDS = 120
DEVICES_FILE = STATE_ROOT / "devices.json"


def utcnow() -> datetime:
    return datetime.now(timezone.utc)


def _load_devices() -> list[dict]:
    from app.localstore import _read_json
    return _read_json(DEVICES_FILE, [])


def _save_devices(devices: list[dict]) -> None:
    from app.localstore import _write_json
    _write_json(DEVICES_FILE, devices)


def _device_status(dev: dict) -> str:
    last = dev.get("last_seen_at")
    if not last:
        return "unknown"
    try:
        elapsed = (utcnow() - datetime.fromisoformat(last)).total_seconds()
    except Exception:
        return "unknown"
    return "online" if elapsed <= OFFLINE_THRESHOLD_SECONDS else "offline"


# ── Heartbeat ─────────────────────────────────────────

@router.post("/devices/heartbeat")
async def device_heartbeat(body: dict, _: str = Depends(verify_api_key)):
    """采集端心跳。可携带采集能力(capabilities)驱动前端输入卡片:

    {
      "device_name": "DAQ-01",
      "capabilities": {"cameras": ["stereo_left", "stereo_right", "head_fisheye_rgb"],
                       "sensors": ["left_glove", "right_glove"]}
    }

    capabilities 可选:带则更新(采集端当前实际连接的硬件),不带则保持旧值。
    心跳间隔必须 < 120s,否则设备判定 offline、输入卡片不再显示。
    """
    devices = _load_devices()
    incoming_id = str(body.get("device_id") or "").strip()
    incoming_name = str(body.get("device_name") or "").strip()
    if not incoming_id and not incoming_name:
        from fastapi import HTTPException
        raise HTTPException(status_code=400, detail="device_id or device_name required")
    # device_id 是稳定身份；旧采集端只有 name 时继续兼容按 name 识别。
    dev = next((d for d in devices if incoming_id and d.get("device_id") == incoming_id), None)
    if dev is None:
        dev = next((d for d in devices if not incoming_id and d.get("name") == incoming_name), None)
    now = utcnow()
    if not dev:
        dev = {"id": str(uuid4()), "device_id": incoming_id or str(uuid4()),
               "name": incoming_name or incoming_id,
               "meta": body.get("meta"), "first_seen_at": now.isoformat()}
        devices.append(dev)
    elif incoming_name:
        # 改名不产生第二台逻辑设备，历史能力与项目绑定继续归属于同一 id。
        dev["name"] = incoming_name
    dev.setdefault("device_id", dev.get("id") or str(uuid4()))
    dev["last_seen_at"] = now.isoformat()
    if body.get("meta"):
        dev["meta"] = body.get("meta")
    # "capabilities" in body 判断:允许空 dict({} 表示拔光硬件,清空能力)
    if "capabilities" in body and body.get("capabilities") is not None:
        caps = body["capabilities"]
        if isinstance(caps, dict):
            dev["capabilities"] = {
                "cameras": _norm_cap_list(caps.get("cameras")),
                "sensors": _norm_cap_list(caps.get("sensors")),
            }
        else:
            dev["capabilities"] = {"cameras": _norm_cap_list(caps), "sensors": []}
    dev["status"] = "online"
    _save_devices(devices)
    return {"ok": True, "device_id": dev["device_id"],
            "device_name": dev["name"], "status": "online"}


# ── Online device capabilities → Studio 输入卡片 ──────

def _norm_cap_list(value) -> list[str]:
    """归一化能力列表:list[str] 直接返回;逗号分隔字符串按逗号拆分
    (采集端可能上报字符串而非列表,避免逐字符迭代)。"""
    if isinstance(value, str):
        return [v.strip() for v in value.split(",") if v.strip()]
    if isinstance(value, list):
        return [v for v in value if isinstance(v, str)]
    return []


def _online_device_capabilities() -> dict:
    """在线设备上报的采集能力 union(cameras + sensors)。"""
    cameras: set[str] = set()
    sensors: set[str] = set()
    devices: list[dict] = []
    for d in _load_devices():
        if _device_status(d) != "online":
            continue
        caps = d.get("capabilities")
        if isinstance(caps, dict):
            device_cameras = _norm_cap_list(caps.get("cameras"))
            device_sensors = _norm_cap_list(caps.get("sensors"))
            cameras.update(device_cameras)
            sensors.update(device_sensors)
        elif isinstance(caps, list):  # 兼容扁平列表
            device_cameras = _norm_cap_list(caps)
            device_sensors = []
            cameras.update(device_cameras)
        else:
            device_cameras, device_sensors = [], []
        devices.append({
            "device_id": d.get("device_id") or d.get("id"),
            "name": d.get("name"),
            "kind": d.get("kind") or (d.get("meta") or {}).get("kind"),
            "cameras": device_cameras,
            "sensors": device_sensors,
            "last_seen_at": d.get("last_seen_at"),
        })
    return {"cameras": sorted(cameras), "sensors": sorted(sensors), "devices": devices}


@router.get("/devices/input-sources")
def device_input_sources():
    """在线采集设备的可用输入源 → 驱动 Studio 输入面板(无 ?project= 时)。

    只聚合在线设备(120s 内心跳过)上报的 capabilities;模块匹配与
    项目 input-sources 共用 match_input_modules。
    """
    from app.processing.batch import match_input_modules
    from app.api.projects import _device_input_sources, _online_input_sources
    caps = _online_device_capabilities()
    episodes = scan_sessions()
    historical_sources = _device_input_sources(episodes)
    online_sources = _online_input_sources(caps.get("devices", []))
    sources_by_id = {str(source.get("id")): source for source in historical_sources}
    for source in online_sources:
        sources_by_id[str(source.get("id"))] = source
    device_sources = list(sources_by_id.values())
    decorate_device_sources(device_sources)
    camera_names = {
        str(camera).lower()
        for episode in episodes
        for camera in (episode.get("camera_names") or [])
        if not str(camera).lower().endswith("_aux")
    }
    sensors = {
        str(sensor).lower()
        for episode in episodes
        for sensor in (episode.get("sensors") or [])
    }
    available = camera_names | sensors
    available |= {c.lower() for c in caps["cameras"] if not c.lower().endswith("_aux")}
    available |= {s.lower() for s in caps["sensors"]}
    device_names = {
        str(stream): str(name)
        for episode in episodes
        for stream, name in (episode.get("device_names") or {}).items()
        if stream and name
    }
    device_display_names = display_names_for_sources(device_sources)
    devices_by_id = {
        str(device.get("key") or device.get("name")): device
        for episode in episodes
        for device in (episode.get("devices") or [])
        if isinstance(device, dict) and (device.get("key") or device.get("name"))
    }
    for device in caps.get("devices", []):
        devices_by_id[str(device.get("device_id") or device.get("name"))] = device
    return {
        "has_episodes": bool(episodes),
        "has_online_devices": bool(caps["cameras"] or caps["sensors"]),
        "cameras": caps["cameras"],
        "sensors": caps["sensors"],
        "device_names": device_names,
        "device_display_names": device_display_names,
        "devices": list(devices_by_id.values()),
        "device_sources": device_sources,
        "modules": match_input_modules(available),
    }


# ── Device list (Web UI) ──────────────────────────────

@router.get("/devices")
async def list_devices():
    devices = _load_devices()
    devices.sort(key=lambda d: d.get("last_seen_at") or "", reverse=True)
    return {"devices": [
        {
            "id": d.get("id"),
            "device_id": d.get("device_id") or d.get("id"),
            "name": d.get("name"),
            "status": _device_status(d),
            "meta": d.get("meta"),
            "first_seen_at": d.get("first_seen_at"),
            "last_seen_at": d.get("last_seen_at"),
            "capabilities": d.get("capabilities") or {"cameras": [], "sensors": []},
        }
        for d in devices
    ]}


# ── Device task polling(项目即采集任务)───────────────

@router.get("/device/tasks")
async def device_tasks(
    device_name: str = Query(..., min_length=1),
    _: str = Depends(verify_api_key),
):
    """采集端轮询此接口获取 active 项目(DAQ client 格式)。

    current_count 口径:各设备上报的录制完成数之和(见 app/device_progress.py);
    只有从未收到过上报的项目才回退到「项目目录下的 session 数」,让升级前就
    存在的历史项目不归零。session 是**上传**的产物,早先直接拿它当进度会让
    同一条数据被算两次(录制一次、上传一次)。
    """
    project_rows = await asyncio.to_thread(list_projects)
    projects = [p for p in project_rows if p.get("status", "active") == "active"]
    reported = await asyncio.to_thread(reported_totals)
    # 只有需要 session 兜底时才扫目录(用了上报口径的项目不再触发远程遍历)
    needs_sessions = device_progress.needs_session_fallback(projects, reported)
    episodes: list = []
    if needs_sessions:
        # Prefer the last complete snapshot.  A cache miss only happens during a
        # cold start, and even that first remote scan is kept off the event loop.
        episodes = cached_sessions_for_tasks()
        if episodes is None:
            episodes = await asyncio.to_thread(scan_sessions)

    tasks = []
    for p in projects:
        sessions = (sum(1 for e in episodes if e.get("project") == p.get("name"))
                    if needs_sessions else None)
        cur, source = device_progress.resolve_count(p, reported, sessions or 0)
        params = p.get("params") or {}
        if not isinstance(params, dict):
            params = {}
        tasks.append({
            "id": p.get("id"),
            "name": p.get("name"),
            "description": p.get("description"),
            "status": p.get("status", "active"),
            "total_required": int(params.get("target_episodes") or 0),
            "current_count": cur,
            "assigned_at": p.get("created_at"),
            "params": params,
            # 供 Web UI/排查用:进度口径与该项目实际已上传的 session 数
            "progress_source": source,
            "session_count": sessions,
        })

    return {"tasks": tasks, "updated_at": utcnow().isoformat()}


@router.post("/device/tasks/progress")
async def device_task_progress(body: dict | None = None,
                               _: str = Depends(verify_api_key)):
    """采集端「录制完成」增量上报(幂等)。

    body::

        {"task_id": "<项目 id>", "device_name": "EGO_001",
         "session_id": "EGO_001:<task_id>:<本地水位>", "increment": 1}

    返回 ``{"completed_count": <该项目最新进度>}`` —— 采集端把这个数当后端
    权威数落进自己的分账(backend_count)。同一 (device_name, session_id)
    重放不重复计数:采集端崩溃/断网后重发同一个水位是安全的。

    错误:参数缺失/非法、任务不存在 → 400。**刻意不用 404** —— 采集端把 404
    当成「后端还没实现该端点」并永久降级(整机不再上报任何进度)。
    """
    from fastapi import HTTPException

    projects = await asyncio.to_thread(list_projects)
    known = {str(p.get("id") or "") for p in projects}
    try:
        return await asyncio.to_thread(
            device_progress.apply_report, body or {}, known)
    except ValueError as exc:
        raise HTTPException(status_code=400, detail=str(exc))