timelapse-2/apps/device-agent/mock-device-cli.py

#!/usr/bin/env python3
"""
mock-device-cli.py

CLI mô phỏng thiết bị timelapse kết nối server theo 2 mode:

1) claim mode:
   - tạo N device UUID giả
   - gọi POST /v1/devices/claim
   - in claim code ra terminal để approve trên dashboard
   - (optional) poll trạng thái claim

2) heartbeat mode:
   - gửi heartbeat định kỳ cho danh sách device đã provisioned
   - dùng X-API-Key header như thiết bị thật

Ví dụ:
  # tạo 5 device claim giả
  python3 mock-device-cli.py claim --server http://localhost:3001 --count 5

  # heartbeat cho 3 device seed mặc định
  python3 mock-device-cli.py heartbeat --server http://localhost:3001 --demo --interval 5

  # heartbeat từ danh sách custom
  python3 mock-device-cli.py heartbeat \
    --device DEMO-001:dev-api-key-demo-001 \
    --device DEMO-002:dev-api-key-demo-002
"""

from __future__ import annotations

import argparse
import os
import random
import string
import time
from datetime import datetime

import requests


def log(msg: str) -> None:
    print(f"[{datetime.now().strftime('%H:%M:%S')}] {msg}", flush=True)


def random_uuid(prefix: str = "mock") -> str:
    rand = "".join(random.choices(string.ascii_lowercase + string.digits, k=10))
    return f"{prefix}-{rand}"


def claim_devices(server: str, count: int, poll: bool, poll_interval: int) -> None:
    claimed: list[dict] = []

    for i in range(1, count + 1):
        device_uuid = random_uuid("pi4")
        payload = {
            "deviceUuid": device_uuid,
            "deviceName": f"Mock-Pi-{i:03d}",
            "serialNo": f"MOCK-{i:03d}",
        }
        try:
            r = requests.post(f"{server}/v1/devices/claim", json=payload, timeout=10)
            if not r.ok:
                log(f"claim failed [{payload['serialNo']}]: {r.status_code} {r.text[:120]}")
                continue
            data = r.json()
            claim_code = data.get("claimCode", "")
            status = data.get("status", "pending")
            claimed.append({"serial": payload["serialNo"], "claimCode": claim_code})
            log(f"claimed {payload['serialNo']} | code={claim_code} | status={status}")
        except Exception as e:
            log(f"claim error [{payload['serialNo']}]: {e}")

    if not claimed:
        log("no device claimed")
        return

    print("\n=== CLAIM CODES ===")
    for item in claimed:
        print(f"  {item['serial']}: {item['claimCode']}")

    if not poll:
        print("\nTip: approve các claim code trên dashboard rồi chạy lại ở heartbeat mode.")
        return

    log("start polling claim status...")
    pending = {c["claimCode"]: c for c in claimed}
    while pending:
        done_codes = []
        for code, item in pending.items():
            try:
                r = requests.get(f"{server}/v1/devices/claim/{code}/status", timeout=10)
                if not r.ok:
                    log(f"poll failed [{item['serial']}] {r.status_code}")
                    continue
                s = r.json()
                st = s.get("status")
                if st == "approved":
                    log(f"approved {item['serial']} | deviceId={s.get('deviceId')} | apiKey={'yes' if s.get('apiKey') else 'no'}")
                    done_codes.append(code)
                elif st in ("rejected", "expired"):
                    log(f"{st} {item['serial']}")
                    done_codes.append(code)
            except Exception as e:
                log(f"poll error [{item['serial']}]: {e}")

        for c in done_codes:
            pending.pop(c, None)

        if pending:
            time.sleep(poll_interval)

    log("all claims resolved")


def parse_devices(device_args: list[str], demo: bool) -> list[tuple[str, str]]:
    devices: list[tuple[str, str]] = []

    if demo:
        devices.extend([
            ("DEMO-001", "dev-api-key-demo-001"),
            ("DEMO-002", "dev-api-key-demo-002"),
            ("DEMO-003", "dev-api-key-demo-003"),
        ])

    for d in device_args:
        if ":" not in d:
            raise ValueError(f"invalid --device format: {d} (expected DEVICE_ID:API_KEY)")
        device_id, api_key = d.split(":", 1)
        devices.append((device_id.strip(), api_key.strip()))

    return devices


def heartbeat_loop(server: str, devices: list[tuple[str, str]], interval: int, jitter: int) -> None:
    if not devices:
        raise ValueError("no devices provided. use --demo or --device DEVICE_ID:API_KEY")

    log(f"heartbeat mode start | devices={len(devices)} | interval={interval}s")

    while True:
        for device_id, api_key in devices:
            payload = {
                "deviceId": device_id,
                "status": random.choice(["online", "online", "online", "degraded"]),
                "tempC": round(random.uniform(38, 62), 1),
                "batteryPct": random.choice([None, random.randint(30, 100)]),
                "storageFreeGb": random.randint(8, 120),
                "capturesToday": random.randint(0, 240),
                "lastCaptureAt": datetime.utcnow().isoformat(),
                "firmwareVersion": "mock-1.0.0",
                "networkStatus": random.choice(["online", "online", "degraded"]),
            }
            try:
                r = requests.post(
                    f"{server}/v1/devices/{device_id}/heartbeat",
                    json=payload,
                    headers={"X-API-Key": api_key, "Content-Type": "application/json"},
                    timeout=10,
                )
                if r.ok:
                    data = r.json()
                    log(f"hb ok {device_id} | pending={data.get('pendingCommands', 0)}")
                else:
                    log(f"hb fail {device_id} | {r.status_code} {r.text[:100]}")
            except Exception as e:
                log(f"hb err {device_id}: {e}")

        sleep_for = interval + random.randint(0, max(jitter, 0))
        time.sleep(sleep_for)


def main() -> None:
    parser = argparse.ArgumentParser(description="Mock device simulator for timelapse server")
    sub = parser.add_subparsers(dest="mode", required=True)

    p_claim = sub.add_parser("claim", help="simulate fresh devices requesting claim")
    p_claim.add_argument("--server", default=os.environ.get("API_BASE", "http://localhost:3001"))
    p_claim.add_argument("--count", type=int, default=3)
    p_claim.add_argument("--poll", action="store_true", help="poll claim status until resolved")
    p_claim.add_argument("--poll-interval", type=int, default=10)

    p_hb = sub.add_parser("heartbeat", help="simulate provisioned devices sending heartbeat")
    p_hb.add_argument("--server", default=os.environ.get("API_BASE", "http://localhost:3001"))
    p_hb.add_argument("--device", action="append", default=[], help="DEVICE_ID:API_KEY (can repeat)")
    p_hb.add_argument("--demo", action="store_true", help="use demo seed devices DEMO-001..003")
    p_hb.add_argument("--interval", type=int, default=10)
    p_hb.add_argument("--jitter", type=int, default=2)

    args = parser.parse_args()

    if args.mode == "claim":
        claim_devices(args.server.rstrip("/"), args.count, args.poll, args.poll_interval)
        return

    if args.mode == "heartbeat":
        devices = parse_devices(args.device, args.demo)
        heartbeat_loop(args.server.rstrip("/"), devices, args.interval, args.jitter)
        return


if __name__ == "__main__":
    main()