#!/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()