From e45a03159b0e95cd46e50c6f1a4a7605ca7ba31c Mon Sep 17 00:00:00 2001 From: kassam Date: Mon, 13 Jul 2026 10:28:20 +0400 Subject: [PATCH] Update 2026-07-13 10:28:18 --- README.md | 20 +- agents/g1/Dockerfile | 45 +- agents/g1/requirements.txt | 7 +- agents/g1/sanad_api_g1.py | 908 +++++++++++++++--- agents/{g1t => g1}/vendor/crc_aarch64.so | Bin agents/{g1t => g1}/vendor/crc_amd64.so | Bin .../unitree_sdk2py-1.0.1-py3-none-any.whl | Bin agents/g1t/.dockerignore | 4 - agents/g1t/.env.example | 43 - agents/g1t/.gitignore | 3 - agents/g1t/Dockerfile | 46 - agents/g1t/README.md | 40 - agents/g1t/docker-compose.yml | 20 - agents/g1t/requirements.txt | 7 - agents/g1t/sanad_api_g1t.py | 572 ----------- agents/go2/sanad_api_go2.py | 9 +- agents/r1/sanad_api_r1.py | 18 +- fleet_install.sh | 147 ++- 18 files changed, 922 insertions(+), 967 deletions(-) rename agents/{g1t => g1}/vendor/crc_aarch64.so (100%) rename agents/{g1t => g1}/vendor/crc_amd64.so (100%) rename agents/{g1t => g1}/vendor/unitree_sdk2py-1.0.1-py3-none-any.whl (100%) delete mode 100644 agents/g1t/.dockerignore delete mode 100644 agents/g1t/.env.example delete mode 100644 agents/g1t/.gitignore delete mode 100644 agents/g1t/Dockerfile delete mode 100644 agents/g1t/README.md delete mode 100644 agents/g1t/docker-compose.yml delete mode 100644 agents/g1t/requirements.txt delete mode 100644 agents/g1t/sanad_api_g1t.py diff --git a/README.md b/README.md index 9c4cc13..ef948c2 100644 --- a/README.md +++ b/README.md @@ -192,15 +192,29 @@ Flow: ### Options +**`install` requires the essentials to be entered explicitly** — the robot's real +serial, the token, and the server: + +```bash +./fleet_install.sh install --sn --token \ + --server-url https://eco.yslootahrobotics.com [--post ] [--name ] +``` + | option | default | meaning | |---|---|---| -| `--sn NAME` | `_` | robot's fleet id (the `sn` field) | -| `--server-ip IP` | auto (route toward robot) | fleet server the robot posts to | -| `--port N` | `8799` | fleet server port | +| `--sn SERIAL` | **required for install** | robot's REAL serial (keys the robot on the server), e.g. `E39N4000Q6D7E70F` | +| `--name NAME` | `_` | friendly display name (e.g. `r1_82`, `g1_58`) | | `--token TOK` | `test-token` | device bearer token | +| `--server-url URL` | — | full fleet server URL (`https://…`) → `VERIFY_TLS=1` | +| `--post PATH` | agent default | ingest POST path (telemetry endpoint; map endpoint for `g1`) | +| `--server-ip IP` | auto (route toward robot) | alternative: local test server by IP (`VERIFY_TLS=0`) | +| `--port N` | `8799` | test-server port (with `--server-ip`) | | `--user USER` | `unitree` | SSH user on the robot | | `--keep-server` | off | (test) leave the workstation test server running | +The interactive flow prompts for the same set: robot type → IP → **SN (required)** +→ display name → SERVER_URL-or-IP → token → POST endpoint. + --- ## 7. The auto-start service (systemd) diff --git a/agents/g1/Dockerfile b/agents/g1/Dockerfile index 2162f1b..bd1a3fd 100644 --- a/agents/g1/Dockerfile +++ b/agents/g1/Dockerfile @@ -1,19 +1,46 @@ -# sanad_api_g1 — G1 fleet MAP uploader. Self-contained, no ROS. +# sanad_api_g1 — G1 fleet agent: TELEMETRY + MAP sync (DDS + web_nav3 map files). +# +# BUILD ON THE G1 JETSON (native arm64). DDS stack: prebuilt CycloneDDS (apt) + +# the CycloneDDS python binding (pip) + the vendored unitree_sdk2py wheel. Run with +# --network host so the robot's DDS is visible (see docker-compose.yml). FROM python:3.10-slim-bookworm -# ca-certificates so outbound HTTPS to the fleet server validates. -RUN apt-get update \ - && apt-get install -y --no-install-recommends ca-certificates \ +# CycloneDDS C lib + idlc (bookworm ships 0.10.2 — matches the robot), build tools +# for the python binding, iproute2 for iface checks, libgomp1 for the TLS guard. +RUN apt-get update && apt-get install -y --no-install-recommends \ + build-essential cmake \ + cyclonedds-dev cyclonedds-tools \ + iproute2 libgomp1 ca-certificates \ && rm -rf /var/lib/apt/lists/* +# aarch64 "static TLS block" guard (harmless on amd64). +ENV LD_PRELOAD=libgomp.so.1 +# cyclonedds build helper looks for libddsc.so under $CYCLONEDDS_HOME/lib; Debian +# installs into the multiarch dir — symlink whatever arch built into /usr/lib. +ENV CYCLONEDDS_HOME=/usr +RUN set -e; for lib in libddsc.so libcycloneddsidl.so; do \ + f=$(ls /usr/lib/*/"$lib" 2>/dev/null | head -1); \ + [ -n "$f" ] && ln -sf "$f" /usr/lib/"$lib" || true; \ + done + WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt +# CycloneDDS python binding, compiled against the C lib above. 0.10.2's build +# helper imports wheel.bdist_wheel (removed in wheel>=0.46) → pin the toolchain. +RUN pip install --no-cache-dir "setuptools<80" "wheel<0.46" \ + && pip install --no-cache-dir --no-build-isolation cyclonedds==0.10.2 + +# Unitree SDK (vendored wheel — not on PyPI) + native crc lib (wheel omits it). +# Both arch crc libs are copied; unitree_sdk2py loads the one matching the image. +COPY vendor/unitree_sdk2py-*.whl /tmp/ +RUN pip install --no-cache-dir --no-deps /tmp/unitree_sdk2py-*.whl +COPY vendor/crc_aarch64.so vendor/crc_amd64.so \ + /usr/local/lib/python3.10/site-packages/unitree_sdk2py/utils/lib/ + COPY sanad_api_g1.py . -# Writable state (upload fingerprints) — override with a volume in compose. -ENV STATE_DIR=/data/state -RUN mkdir -p /data/state - -# Run as the loop by default; override the CMD for --once / --list / --dry-run. +ENV PYTHONUNBUFFERED=1 \ + DDS_INTERFACE=eth0 \ + DDS_DOMAIN=0 ENTRYPOINT ["python", "-u", "sanad_api_g1.py"] diff --git a/agents/g1/requirements.txt b/agents/g1/requirements.txt index b143394..b462572 100644 --- a/agents/g1/requirements.txt +++ b/agents/g1/requirements.txt @@ -1,2 +1,7 @@ -# sanad_api_g1 map uploader — deliberately tiny (no ROS, no DDS). +# sanad_api_r1 telemetry agent. requests>=2.31,<3 +# only needed if R1_POSITION_SOURCE=rosbridge (reads /odom for position): +websocket-client>=1.6,<2 +# DDS stack (cyclonedds + the vendored unitree_sdk2py wheel) is installed by the +# Dockerfile, not from here — see Dockerfile. The agent degrades to heartbeats if +# unitree_sdk2py is unavailable. diff --git a/agents/g1/sanad_api_g1.py b/agents/g1/sanad_api_g1.py index 19378e6..5b30534 100644 --- a/agents/g1/sanad_api_g1.py +++ b/agents/g1/sanad_api_g1.py @@ -1,63 +1,48 @@ #!/usr/bin/env python3 -"""sanad_api_g1 — G1 fleet MAP uploader. +"""sanad_api_g1 — G1 fleet agent: TELEMETRY + MAP sync in ONE service. -Scope (this build): upload the G1's navigation MAP to the YS Lootah fleet -server. Nothing else (no telemetry / commands / logs — those are separate -agents). It is the "Maps sync" row of the fleet spec: +The single G1 agent (one container, one systemd service) that: - POST {SERVER_URL}/api/v1/fleet/ingest/{sn}/map (Bearer device token) + 1. TELEMETRY — every ~2 s POSTs the robot's live status: + POST {SERVER_URL}/api/v1/fleet/ingest/telemetry + { sn, name, mac, brand, type, model, battery, charging, battery_detail, + motor_temp, storage, status, position, faults, map, ts } -WHAT IT UPLOADS ---------------- -The G1's map is produced by the web_nav3 (Nav2 + RTAB-Map) stack and stored on -disk as a RTAB-Map SQLite ``.db`` file (per web_nav3/web/backend.py: -``MAPS_ROOT//.db`` + a ``maps_meta.json`` sidecar), with each map's -named places kept in ``web/data//places/.json``. + 2. MAP — a background loop (every MAP_POLL_INTERVAL, default 30 s) checks the + Sanad dashboard's saved maps (web_nav3: RTAB-Map .db + places) and uploads + each map ONE time (re-upload only if its content changes): + POST {SERVER_URL}/api/v1/fleet/ingest/{sn}/map (multipart: db + meta) -There is NO rendered PNG on disk — the dashboard draws the occupancy grid live -over rosbridge. By design choice this agent uploads the RAW ``.db`` (the actual -map artifact) plus its places, rather than rendering an image. That means the -FLEET SERVER must accept a ``rtabmap_db`` artifact on the map endpoint (see -MAP_UPLOAD_MODE for the two wire formats). The documented spec body -(image_base64 / resolution / origin) is for a rendered map; switch to the -telemetry-agent's live-rosbridge renderer if you need that instead. + The map result is SHOWN inside every telemetry post as the "map" field: + "map": { "uploaded": true|false, + "state": "uploaded" | "no_map" | "failed" | "pending", + "maps_found": N, "last_map": "floor-1"|null, + "error": "no saved map found …"|null, "checked_ts": … } + so the server always sees whether the map made it — and why not. -NO ROS, NO DDS. Pure files + HTTPS, so it drops into any robot as its own -container. It reads the maps/places straight from mounted volumes; the web_nav3 -HTTP API is only used (optionally) to learn which map is "active". +DATA SOURCES (Unitree G1, unitree_hg DDS) +----------------------------------------- + battery / charging : rt/lf/bmsstate (BmsState_) soc, current, voltage, temp, soh, cycles + faults / motor temp: rt/lowstate (LowState_) per-motor temps + staleness + position {x,y} : rt/lf/odommodestate (SportModeState_) firmware odom over DDS + status : derived (charging/moving/idle/offline); optional FSM GET RPC + (G1 ids: 200 walk-ready / 4 stand / 2 squat / 702 lie2stand) + storage : host disk via the read-only /:/host mount (+ Sanad data dir size) + maps : MAPS_DIR//*.db (+ maps_meta.json), places under + DATA_DIR//places/.json — the web_nav3 stores -CHANGE DETECTION ----------------- -Each map's ``.db`` is fingerprinted (size + mtime fast-path, then sha256). A map -is (re)uploaded only when its fingerprint changes — matching the spec's "send -the nav map on change". State persists in ``STATE_DIR/uploaded.json`` so a -restart doesn't re-push unchanged maps. +Read-only toward the robot: never commands motion. Degrades to heartbeats if +unitree_sdk2py is unavailable; --simulate fakes only the DDS side (the map scan +stays real). -CONFIG — all via environment (see .env.example) ------------------------------------------------ - SERVER_URL base URL of the fleet server (required) e.g. https://fleet.example.com - DEVICE_TOKEN per-robot bearer token (required) - SN this robot's fleet id (path key) default g1_7892 - ROBOT web_nav3 robot name (maps subdir + header) default sanad - MAPS_DIR mounted web_nav3 maps/ dir (contains /*.db and/or *.db) - DATA_DIR mounted web_nav3 web/data dir (per-map places live under /places/) - LEGACY_PLACES optional path to a legacy places.json (per-robot, no map scoping) - WEB_NAV3_URL optional http://127.0.0.1:8765 — used only to read the ACTIVE map - MAP_SELECT all | active | newest default all - MAP_UPLOAD_MODE multipart | base64json default multipart - MAP_ENDPOINT path template, {sn} substituted default /api/v1/fleet/ingest/{sn}/map - POLL_INTERVAL seconds between scans (loop mode) default 30 - STATE_DIR writable dir for upload state default /data/state - VERIFY_TLS 1|0 verify server TLS default 1 - HTTP_TIMEOUT per-request timeout seconds default 30 +CONFIG — environment (see .env.example). Key vars: + SERVER_URL, DEVICE_TOKEN, SN (required at install), ROBOT_NAME, + DDS_INTERFACE (eth0), POLL_INTERVAL (2), TELEMETRY_ENDPOINT, + ROBOT (web_nav3 robot name, default sanad), MAPS_DIR, DATA_DIR, STATE_DIR, + MAP_SELECT (all|active|newest), MAP_UPLOAD_MODE (multipart|base64json), + MAP_ENDPOINT, MAP_POLL_INTERVAL (30), VERIFY_TLS, HTTP_TIMEOUT -CLI ---- - python sanad_api_g1.py # run the loop (default) - python sanad_api_g1.py --once # one scan+upload pass, then exit - python sanad_api_g1.py --list # list discovered maps (no upload) - python sanad_api_g1.py --dry-run # build payloads + report, never POST - python sanad_api_g1.py --force # ignore state; upload even if unchanged +CLI: --simulate | --once | --dry-run | --force (map re-upload) | --list (maps) | -v """ from __future__ import annotations @@ -68,8 +53,11 @@ import json import logging import math import os +import shutil import sys +import threading import time +import uuid from dataclasses import dataclass, field from pathlib import Path from typing import Any, Dict, List, Optional @@ -80,7 +68,7 @@ log = logging.getLogger("sanad_api_g1") # --------------------------------------------------------------------------- # -# tiny .env loader (so it also runs bare, outside docker) — no dependency +# env helpers # --------------------------------------------------------------------------- # def _load_dotenv(path: str = ".env") -> None: p = Path(path) @@ -91,8 +79,7 @@ def _load_dotenv(path: str = ".env") -> None: if not line or line.startswith("#") or "=" not in line: continue k, _, v = line.partition("=") - k, v = k.strip(), v.strip().strip('"').strip("'") - os.environ.setdefault(k, v) + os.environ.setdefault(k.strip(), v.strip().strip('"').strip("'")) def _env(name: str, default: str = "") -> str: @@ -100,8 +87,7 @@ def _env(name: str, default: str = "") -> str: def _env_bool(name: str, default: bool) -> bool: - v = _env(name, "1" if default else "0").lower() - return v in ("1", "true", "yes", "on") + return _env(name, "1" if default else "0").lower() in ("1", "true", "yes", "on") # --------------------------------------------------------------------------- # @@ -112,16 +98,35 @@ class Config: server_url: str device_token: str sn: str + name: str + brand: str + robot_type: str + model: str + storage_path: str + data_path: str + # telemetry / DDS + dds_interface: str + dds_domain: int + mac_interface: str + read_fsm: bool + position_source: str + rosbridge_url: str + low_soc: int + motor_temp_max: float + poll_interval: float + telemetry_endpoint: str + # map sync robot: str maps_dir: Path - data_dir: Optional[Path] + web_data_dir: Optional[Path] legacy_places: Optional[Path] web_nav3_url: str map_select: str - upload_mode: str - endpoint_tmpl: str - poll_interval: float + map_upload_mode: str + map_endpoint_tmpl: str + map_poll_interval: float state_dir: Path + # transport verify_tls: bool http_timeout: float @@ -132,50 +137,339 @@ class Config: missing = [n for n, v in (("SERVER_URL", server), ("DEVICE_TOKEN", token)) if not v] if missing: raise SystemExit(f"[config] missing required env: {', '.join(missing)}") + iface = _env("DDS_INTERFACE", "eth0") data_dir = _env("DATA_DIR") legacy = _env("LEGACY_PLACES") return cls( server_url=server, device_token=token, - sn=_env("SN", "g1_7892"), + sn=_env("SN", "g1_0000"), + name=_env("ROBOT_NAME", "") or _env("SN", "g1_0000"), + brand=_env("ROBOT_BRAND", "unitree"), + robot_type=_env("ROBOT_TYPE", "humanoid"), + model=_env("ROBOT_MODEL", "g1"), + storage_path=_env("STORAGE_PATH", ""), + data_path=_env("STORAGE_DATA_PATH", ""), + dds_interface=iface, + dds_domain=int(_env("DDS_DOMAIN", "0")), + mac_interface=_env("MAC_INTERFACE", iface), + read_fsm=_env_bool("G1_READ_FSM", False), + position_source=_env("G1_POSITION_SOURCE", "odom").lower(), + rosbridge_url=_env("ROSBRIDGE_URL", "ws://127.0.0.1:9090"), + low_soc=int(_env("LOW_SOC", "15")), + motor_temp_max=float(_env("MOTOR_TEMP_MAX", "85")), + poll_interval=float(_env("POLL_INTERVAL", "2")), + telemetry_endpoint=_env("TELEMETRY_ENDPOINT", "/api/v1/fleet/ingest/telemetry"), robot=_env("ROBOT", "sanad"), maps_dir=Path(_env("MAPS_DIR", "/data/maps")), - data_dir=Path(data_dir) if data_dir else None, + web_data_dir=Path(data_dir) if data_dir else None, legacy_places=Path(legacy) if legacy else None, web_nav3_url=_env("WEB_NAV3_URL", "").rstrip("/"), map_select=_env("MAP_SELECT", "all").lower(), - upload_mode=_env("MAP_UPLOAD_MODE", "multipart").lower(), - endpoint_tmpl=_env("MAP_ENDPOINT", "/api/v1/fleet/ingest/{sn}/map"), - poll_interval=float(_env("POLL_INTERVAL", "30")), + map_upload_mode=_env("MAP_UPLOAD_MODE", "multipart").lower(), + map_endpoint_tmpl=_env("MAP_ENDPOINT", "/api/v1/fleet/ingest/{sn}/map"), + map_poll_interval=float(_env("MAP_POLL_INTERVAL", "30")), state_dir=Path(_env("STATE_DIR", "/data/state")), verify_tls=_env_bool("VERIFY_TLS", True), http_timeout=float(_env("HTTP_TIMEOUT", "30")), ) + def telemetry_url(self) -> str: + return self.server_url + self.telemetry_endpoint + def map_url(self) -> str: - return self.server_url + self.endpoint_tmpl.format(sn=self.sn) + return self.server_url + self.map_endpoint_tmpl.format(sn=self.sn) def auth_headers(self) -> Dict[str, str]: return {"Authorization": f"Bearer {self.device_token}"} # --------------------------------------------------------------------------- # -# map artifact +# mac + storage +# --------------------------------------------------------------------------- # +def read_mac(interface: str) -> str: + p = Path(f"/sys/class/net/{interface}/address") + try: + mac = p.read_text().strip() + if mac and mac != "00:00:00:00:00:00": + return mac.lower() + except Exception: + pass + n = uuid.getnode() + return ":".join(f"{(n >> b) & 0xff:02x}" for b in range(40, -1, -8)) + + +_data_size_cache: Dict[str, Any] = {"ts": 0.0, "kb": None} + + +def read_storage(cfg: Config) -> Optional[Dict[str, Any]]: + """Disk usage of the robot's root fs + optional Sanad data-dir size. + + In docker, bind-mount the host root read-only at /host (the installer does) + so this reports the HOST disk, not the container overlay.""" + root = cfg.storage_path or ("/host" if os.path.isdir("/host") else "/") + try: + du = shutil.disk_usage(root) + out: Dict[str, Any] = { + "total_gb": round(du.total / 1e9, 2), + "free_gb": round(du.free / 1e9, 2), + "used_percent": round(du.used / du.total * 100, 1), + } + except Exception: + return None + if cfg.data_path and os.path.isdir(cfg.data_path): + now = time.monotonic() + if _data_size_cache["kb"] is None or now - _data_size_cache["ts"] > 60: + try: + total = 0 + for r, _, files in os.walk(cfg.data_path): + for f in files: + try: + total += os.path.getsize(os.path.join(r, f)) + except OSError: + pass + _data_size_cache.update(ts=now, kb=round(total / 1024, 1)) + except Exception: + pass + if _data_size_cache["kb"] is not None: + out["data_kb"] = _data_size_cache["kb"] + return out + + +# --------------------------------------------------------------------------- # +# DDS reader (telemetry side — degrades to heartbeats if SDK absent) +# --------------------------------------------------------------------------- # +class DDSReader: + """Subscribes rt/lf/bmsstate + rt/lowstate (+ rt/lf/odommodestate for position). + Passive reads; the only RPC ever issued is GET_FSM_ID.""" + + def __init__(self, cfg: Config): + self.cfg = cfg + self._lock = threading.Lock() + self._bms: Optional[Dict[str, Any]] = None + self._bms_ts = 0.0 + self._low_ts = 0.0 + self._temps: List[float] = [] + self._max_dq = 0.0 + self._xy: Optional[Dict[str, float]] = None + self._loco = None + self.ok = False + self._start() + + def _start(self) -> None: + try: + from unitree_sdk2py.core.channel import ( + ChannelFactoryInitialize, ChannelSubscriber) + from unitree_sdk2py.idl.unitree_hg.msg.dds_ import LowState_ + try: + from unitree_sdk2py.idl.unitree_hg.msg.dds_ import BmsState_ + except Exception: + BmsState_ = None + SportModeState_ = None + if self.cfg.position_source == "odom": + try: + # G1 firmware publishes odom as unitree_go SportModeState_. + from unitree_sdk2py.idl.unitree_go.msg.dds_ import SportModeState_ + except Exception: + SportModeState_ = None + except Exception as e: + log.warning("unitree_sdk2py unavailable (%s) — telemetry runs in heartbeat mode", e) + return + try: + ChannelFactoryInitialize(self.cfg.dds_domain, self.cfg.dds_interface) + self._low_sub = ChannelSubscriber("rt/lowstate", LowState_) + self._low_sub.Init(self._on_low, 10) + if BmsState_ is not None: + self._bms_sub = ChannelSubscriber("rt/lf/bmsstate", BmsState_) + self._bms_sub.Init(self._on_bms, 10) + else: + log.warning("BmsState_ not in this unitree_sdk2py — battery will be null") + if SportModeState_ is not None: + self._odom_sub = ChannelSubscriber("rt/lf/odommodestate", SportModeState_) + self._odom_sub.Init(self._on_odom, 10) + if self.cfg.read_fsm: + self._init_loco() + self.ok = True + log.info("DDS up: domain=%d iface=%s (bmsstate + lowstate%s)", + self.cfg.dds_domain, self.cfg.dds_interface, + " + odom" if SportModeState_ is not None else "") + except Exception as e: + log.warning("DDS init failed (%s) — heartbeat mode", e) + + def _init_loco(self) -> None: + try: + from unitree_sdk2py.rpc.client import Client # type: ignore + except Exception as e: + log.warning("loco RPC client unavailable (%s) — status from BMS/motion only", e) + return + try: + c = Client("loco", 0); c.Init(); c.SetTimeout(3.0) + self._loco = c + log.info("loco FSM read enabled (GET-only, no motion)") + except Exception as e: + log.warning("loco client init failed (%s) — status from BMS/motion only", e) + self._loco = None + + def _on_bms(self, msg) -> None: + try: + soc = int(getattr(msg, "soc", 0) or 0) + cur_mA = int(getattr(msg, "current", 0) or 0) + # Pack voltage: prefer bmsvoltage[0] (mV); else sum of cell voltages. + volt_mv = 0 + bv = getattr(msg, "bmsvoltage", None) + try: + if bv is not None and len(bv) and int(bv[0]): + volt_mv = int(bv[0]) + except Exception: + volt_mv = 0 + if not volt_mv: + cv = getattr(msg, "cell_vol", None) + if cv is not None: + try: + volt_mv = int(sum(int(x) for x in cv if x)) + except Exception: + volt_mv = 0 + # Max plausible pack temperature (int16 °C). + temp_c = None + tt = getattr(msg, "temperature", None) + if tt is not None: + try: + vals = [int(x) for x in tt if -40 <= int(x) <= 150] + if vals: + temp_c = max(vals) + except Exception: + temp_c = None + with self._lock: + self._bms = { + "soc": max(0, min(100, soc)), + "current_a": round(cur_mA / 1000.0, 2), + "voltage_v": round(volt_mv / 1000.0, 1) if volt_mv else None, + "temp_c": temp_c, + "soh": int(getattr(msg, "soh", 0) or 0), + "cycles": int(getattr(msg, "cycle", 0) or 0), + } + self._bms_ts = time.monotonic() + except Exception: + pass + + def _on_low(self, msg) -> None: + try: + temps: List[float] = [] + max_dq = 0.0 + for m in (getattr(msg, "motor_state", None) or []): + t = getattr(m, "temperature", None) + if t is not None: + try: + vals = [float(x) for x in t] if hasattr(t, "__iter__") else [float(t)] + # 0 = slot not reporting (unpopulated motor), not a real temp + temps.extend(v for v in vals if 0 < v <= 200) + except Exception: + pass + dq = getattr(m, "dq", None) + if dq is not None: + try: + max_dq = max(max_dq, abs(float(dq))) + except Exception: + pass + with self._lock: + self._low_ts = time.monotonic() + self._temps = temps + self._max_dq = max_dq + except Exception: + pass + + def _on_odom(self, msg) -> None: + try: + pos = getattr(msg, "position", None) + if pos is not None and len(pos) >= 2: + with self._lock: + self._xy = {"x": round(float(pos[0]), 3), "y": round(float(pos[1]), 3)} + except Exception: + pass + + def snapshot(self) -> Dict[str, Any]: + with self._lock: + now = time.monotonic() + return { + "bms": dict(self._bms) if self._bms else None, + "low_age": (now - self._low_ts) if self._low_ts else None, + "temps": list(self._temps), + "max_dq": self._max_dq, + "xy": dict(self._xy) if self._xy else None, + } + + def fsm_id(self) -> Optional[int]: + if not self._loco: + return None + try: + code, data = self._loco._Call(7001, "{}") # GET_FSM_ID — read-only + if code == 0 and data: + return int(json.loads(data).get("data", data)) if data.strip().startswith("{") else int(data) + except Exception as e: + log.debug("fsm read failed: %s", e) + return None + + +# G1 FSM ids (differ from the R1's): 200 balance/walk-ready, 4 StandUp, 2 Squat, 702 Lie2Stand. +_FSM_STATUS = {200: "ready", 4: "standing", 2: "squat", 702: "lie2stand"} + + +class RosbridgePosition: + def __init__(self, cfg: Config): + self.cfg = cfg + self._xy: Optional[Dict[str, float]] = None + self._lock = threading.Lock() + self._stop = False + try: + import websocket # noqa: F401 + except Exception as e: + log.warning("websocket-client absent (%s) — rosbridge position disabled", e) + self._ok = False + return + self._ok = True + threading.Thread(target=self._run, daemon=True).start() + + def _run(self) -> None: + import websocket + sub = json.dumps({"op": "subscribe", "topic": "/odom", + "type": "nav_msgs/Odometry", "throttle_rate": 500}) + while not self._stop: + try: + ws = websocket.create_connection(self.cfg.rosbridge_url, timeout=5) + ws.send(sub) + while not self._stop: + msg = json.loads(ws.recv()) + pos = (((msg.get("msg") or {}).get("pose") or {}).get("pose") or {}).get("position") + if pos: + with self._lock: + self._xy = {"x": round(float(pos["x"]), 3), "y": round(float(pos["y"]), 3)} + except Exception as e: + log.debug("rosbridge position reconnect: %s", e) + time.sleep(3) + + def get(self) -> Optional[Dict[str, float]]: + with self._lock: + return dict(self._xy) if self._xy else None + + +# --------------------------------------------------------------------------- # +# map sync (web_nav3 saved maps → fleet server, uploaded ONCE per content) # --------------------------------------------------------------------------- # @dataclass class MapArtifact: - path: Path # absolute path to the .db - name: str # file name, e.g. floor-1.db - stem: str # name without .db, e.g. floor-1 + path: Path # .db (rtabmap) or .yaml (slam_toolbox set) + name: str + stem: str size: int mtime: int + fmt: str = "rtabmap_db" # rtabmap_db | slam_toolbox + files: Dict[str, Path] = field(default_factory=dict) # slam set: pgm/yaml/posegraph/data description: str = "" sha256: str = "" points: List[Dict[str, Any]] = field(default_factory=list) def fingerprint(self) -> str: - # cheap identity for the change check before we hash the whole file - return f"{self.size}:{self.mtime}" + return f"{self.fmt}:{self.size}:{self.mtime}" def _sha256(path: Path) -> str: @@ -186,11 +480,125 @@ def _sha256(path: Path) -> str: return h.hexdigest() -# --------------------------------------------------------------------------- # -# discovery — read maps + places straight from the mounted web_nav3 files -# --------------------------------------------------------------------------- # +def _sha256_set(paths: List[Path]) -> str: + h = hashlib.sha256() + for p in sorted(paths): + with p.open("rb") as f: + for chunk in iter(lambda: f.read(1024 * 1024), b""): + h.update(chunk) + return h.hexdigest() + + +# ---- slam_toolbox map set (office.yaml + office.pgm [+ .posegraph .data]) ---- +def _parse_map_yaml(path: Path) -> Dict[str, Any]: + """Tiny parser for a ROS map_server yaml (image/resolution/origin) — no pyyaml.""" + out: Dict[str, Any] = {} + for line in path.read_text().splitlines(): + line = line.split("#", 1)[0].strip() + if ":" not in line: + continue + k, _, v = line.partition(":") + k, v = k.strip(), v.strip() + if k == "image": + out["image"] = v + elif k == "resolution": + try: + out["resolution"] = float(v) + except ValueError: + pass + elif k == "origin": + try: + nums = [float(x) for x in v.strip("[]").split(",")] + out["origin"] = {"x": nums[0], "y": nums[1], + "yaw": nums[2] if len(nums) > 2 else 0.0} + except Exception: + pass + return out + + +def _read_pgm(path: Path) -> Optional[Dict[str, Any]]: + """Parse a binary PGM (P5): returns {width, height, maxval, pixels(bytes)}.""" + try: + data = path.read_bytes() + if not data.startswith(b"P5"): + return None + # tokenize header (magic, width, height, maxval), skipping comments + tokens: List[bytes] = [] + i = 2 + while len(tokens) < 3 and i < len(data): + c = data[i:i + 1] + if c in b" \t\r\n": + i += 1 + elif c == b"#": + i = data.index(b"\n", i) + 1 + else: + j = i + while j < len(data) and data[j:j + 1] not in b" \t\r\n": + j += 1 + tokens.append(data[i:j]) + i = j + w, h, maxval = int(tokens[0]), int(tokens[1]), int(tokens[2]) + pixels = data[i + 1: i + 1 + w * h] + if len(pixels) < w * h: + return None + return {"width": w, "height": h, "maxval": maxval, "pixels": pixels} + except Exception: + return None + + +def _pgm_to_png_b64(pgm: Dict[str, Any]) -> str: + """Grayscale 8-bit PNG from parsed PGM — pure stdlib (zlib + struct).""" + import struct + import zlib + + w, h, pixels = pgm["width"], pgm["height"], pgm["pixels"] + + def chunk(tag: bytes, body: bytes) -> bytes: + return (struct.pack(">I", len(body)) + tag + body + + struct.pack(">I", zlib.crc32(tag + body) & 0xFFFFFFFF)) + + ihdr = struct.pack(">IIBBBBB", w, h, 8, 0, 0, 0, 0) # 8-bit grayscale + raw = b"".join(b"\x00" + pixels[y * w:(y + 1) * w] for y in range(h)) + png = (b"\x89PNG\r\n\x1a\n" + chunk(b"IHDR", ihdr) + + chunk(b"IDAT", zlib.compress(raw, 6)) + chunk(b"IEND", b"")) + return base64.b64encode(png).decode("ascii") + + +def _discover_slam_sets(cfg: Config) -> List[MapArtifact]: + """Find slam_toolbox / map_server map sets: .yaml + .pgm + (+ optional .posegraph/.data) under the maps roots and maps_slam/.""" + roots = [cfg.maps_dir, cfg.maps_dir / cfg.robot, cfg.maps_dir / "maps_slam"] + seen: set = set() + out: List[MapArtifact] = [] + for root in roots: + if not root.exists(): + continue + for y in sorted(root.glob("*.yaml")): + meta = _parse_map_yaml(y) + img = meta.get("image", "") + pgm = (y.parent / img) if img else y.with_suffix(".pgm") + if not pgm.exists(): + pgm = y.with_suffix(".pgm") + if not pgm.exists(): + continue # yaml without a raster — not a map set + rp = str(y.resolve()) + if rp in seen: + continue + seen.add(rp) + files: Dict[str, Path] = {"yaml": y, "pgm": pgm} + for ext in ("posegraph", "data"): + p = y.with_suffix("." + ext) + if p.exists(): + files[ext] = p + size = sum(p.stat().st_size for p in files.values()) + mtime = max(int(p.stat().st_mtime) for p in files.values()) + out.append(MapArtifact(path=y, name=y.name, stem=y.stem, + size=size, mtime=mtime, + fmt="slam_toolbox", files=files)) + return out + + def _map_key(stem: str) -> str: - """Mirror backend._map_key: safe filename stem for the per-map places file.""" stem = Path(stem).name if stem.endswith(".db"): stem = stem[:-3] @@ -205,8 +613,6 @@ def _read_json(path: Path, default: Any) -> Any: def _yaw_from_pose(pose: Dict[str, Any]) -> float: - """Yaw (rad) from a places-pose dict. Supports full quaternion, planar - (qz,qw), or an explicit yaw field.""" if "qw" in pose or "qz" in pose: qx = float(pose.get("qx", 0.0)); qy = float(pose.get("qy", 0.0)) qz = float(pose.get("qz", 0.0)); qw = float(pose.get("qw", 1.0)) @@ -216,21 +622,16 @@ def _yaw_from_pose(pose: Dict[str, Any]) -> float: def _places_files_for(cfg: Config, stem: str) -> List[Path]: - """Candidate on-disk places files for a map, most-specific first.""" out: List[Path] = [] key = _map_key(stem) - if cfg.data_dir: - out.append(cfg.data_dir / cfg.robot / "places" / f"{key}.json") + if cfg.web_data_dir: + out.append(cfg.web_data_dir / cfg.robot / "places" / f"{key}.json") if cfg.legacy_places: out.append(cfg.legacy_places) return out def load_points(cfg: Config, stem: str) -> List[Dict[str, Any]]: - """Return the map's saved places as fleet points: {name, type, x, y, yaw}. - - Places store shape (web_nav3): {"": {x,y,z,qx,qy,qz,qw}}. - """ for pf in _places_files_for(cfg, stem): data = _read_json(pf, None) if pf.exists() else None if isinstance(data, dict) and data: @@ -248,20 +649,17 @@ def load_points(cfg: Config, stem: str) -> List[Dict[str, Any]]: }) except (KeyError, TypeError, ValueError): continue - log.debug("points for %s: %d (from %s)", stem, len(pts), pf) return pts return [] def discover_maps(cfg: Config) -> List[MapArtifact]: - """Find every .db under MAPS_DIR// and MAPS_DIR/ (legacy root).""" roots = [cfg.maps_dir / cfg.robot, cfg.maps_dir] meta: Dict[str, Any] = {} meta_file = cfg.maps_dir / cfg.robot / "maps_meta.json" if meta_file.exists(): meta = _read_json(meta_file, {}) or {} - - seen: set[str] = set() + seen: set = set() out: List[MapArtifact] = [] for root in roots: if not root.exists(): @@ -273,19 +671,17 @@ def discover_maps(cfg: Config) -> List[MapArtifact]: seen.add(rp) st = p.stat() out.append(MapArtifact( - path=p, - name=p.name, - stem=p.stem, - size=st.st_size, - mtime=int(st.st_mtime), + path=p, name=p.name, stem=p.stem, + size=st.st_size, mtime=int(st.st_mtime), description=(meta.get(p.name) or {}).get("description", ""), )) + # slam_toolbox map sets (office.yaml + office.pgm …) live alongside + out.extend(_discover_slam_sets(cfg)) out.sort(key=lambda m: m.mtime, reverse=True) return out def _active_map_name(cfg: Config) -> Optional[str]: - """Ask web_nav3 which map is currently loaded (optional; None if not set/up).""" if not cfg.web_nav3_url: return None try: @@ -295,8 +691,7 @@ def _active_map_name(cfg: Config) -> Optional[str]: r.raise_for_status() am = (r.json() or {}).get("active_map") return _map_key(am) if am else None - except requests.RequestException as e: - log.debug("active-map query failed: %s", e) + except requests.RequestException: return None @@ -311,14 +706,10 @@ def select_maps(cfg: Config, maps: List[MapArtifact]) -> List[MapArtifact]: picked = [m for m in maps if _map_key(m.stem) == active] if picked: return picked - log.warning("active map %r not found on disk; falling back to newest", active) return maps[:1] return maps # "all" -# --------------------------------------------------------------------------- # -# state (which fingerprints already uploaded) -# --------------------------------------------------------------------------- # def _state_file(cfg: Config) -> Path: return cfg.state_dir / "uploaded.json" @@ -332,18 +723,15 @@ def save_state(cfg: Config, state: Dict[str, str]) -> None: cfg.state_dir.mkdir(parents=True, exist_ok=True) _state_file(cfg).write_text(json.dumps(state, indent=2)) except Exception as e: - log.warning("could not persist state: %s", e) + log.warning("could not persist map state: %s", e) -# --------------------------------------------------------------------------- # -# upload -# --------------------------------------------------------------------------- # def build_meta(cfg: Config, m: MapArtifact) -> Dict[str, Any]: return { "sn": cfg.sn, "name": m.stem, "file": m.name, - "format": "rtabmap_db", + "format": m.fmt, "size_bytes": m.size, "sha256": m.sha256, "mtime": m.mtime, @@ -352,11 +740,45 @@ def build_meta(cfg: Config, m: MapArtifact) -> Dict[str, Any]: } +def _upload_slam_map(cfg: Config, m: MapArtifact, session: requests.Session) -> bool: + """slam_toolbox map → the spec's image JSON: PNG (from the pgm) + resolution + + origin + width/height + points. This is what the dashboard renders.""" + url = cfg.map_url() + ymeta = _parse_map_yaml(m.files["yaml"]) + pgm = _read_pgm(m.files["pgm"]) + if pgm is None: + log.error("map %s: cannot parse %s (not binary P5?)", m.stem, m.files["pgm"].name) + return False + body = build_meta(cfg, m) + body.update({ + "resolution": ymeta.get("resolution"), + "origin": ymeta.get("origin"), + "width": pgm["width"], + "height": pgm["height"], + "image_base64": _pgm_to_png_b64(pgm), + }) + try: + resp = session.post(url, json=body, headers=cfg.auth_headers(), + timeout=cfg.http_timeout, verify=cfg.verify_tls) + except requests.RequestException as e: + log.error("map upload %s FAILED (transport): %s", m.stem, e) + return False + if not resp.ok: + log.error("map upload %s FAILED: HTTP %s %s", m.stem, resp.status_code, resp.text[:300]) + return False + log.info("map uploaded: %s (slam_toolbox %dx%d @ %sm, %d points) -> HTTP %s", + m.stem, pgm["width"], pgm["height"], ymeta.get("resolution"), + len(m.points), resp.status_code) + return True + + def upload_map(cfg: Config, m: MapArtifact, session: requests.Session) -> bool: + if m.fmt == "slam_toolbox": + return _upload_slam_map(cfg, m, session) url = cfg.map_url() meta = build_meta(cfg, m) try: - if cfg.upload_mode == "base64json": + if cfg.map_upload_mode == "base64json": body = dict(meta) body["db_base64"] = base64.b64encode(m.path.read_bytes()).decode("ascii") resp = session.post(url, json=body, headers=cfg.auth_headers(), @@ -369,111 +791,291 @@ def upload_map(cfg: Config, m: MapArtifact, session: requests.Session) -> bool: headers=cfg.auth_headers(), timeout=cfg.http_timeout, verify=cfg.verify_tls) except requests.RequestException as e: - log.error("upload %s FAILED (transport): %s", m.name, e) + log.error("map upload %s FAILED (transport): %s", m.name, e) return False - if not resp.ok: - detail = resp.text[:300] - log.error("upload %s FAILED: HTTP %s %s", m.name, resp.status_code, detail) + log.error("map upload %s FAILED: HTTP %s %s", m.name, resp.status_code, resp.text[:300]) return False - log.info("uploaded %s (%.2f MB, %d points) -> HTTP %s", + log.info("map uploaded: %s (%.2f MB, %d points) -> HTTP %s", m.name, m.size / 1024 / 1024, len(m.points), resp.status_code) return True -# --------------------------------------------------------------------------- # -# one pass -# --------------------------------------------------------------------------- # -def run_once(cfg: Config, *, force: bool, dry_run: bool, - session: requests.Session) -> int: - maps = select_maps(cfg, discover_maps(cfg)) +# Shared map status — SHOWN in every telemetry post ("map" field). +_MAP_STATUS_LOCK = threading.Lock() +_MAP_STATUS: Dict[str, Any] = { + "uploaded": False, "state": "pending", "maps_found": 0, + "last_map": None, "error": None, "checked_ts": None, +} + + +def _set_map_status(**kw: Any) -> None: + with _MAP_STATUS_LOCK: + _MAP_STATUS.update(kw) + _MAP_STATUS["checked_ts"] = int(time.time()) + + +def get_map_status() -> Dict[str, Any]: + with _MAP_STATUS_LOCK: + return dict(_MAP_STATUS) + + +def map_sync_once(cfg: Config, session: requests.Session, + force: bool = False, dry_run: bool = False) -> int: + """One map pass: scan the Sanad dashboard maps and upload anything new. + Always updates the shared map status (visible in telemetry).""" + try: + maps = select_maps(cfg, discover_maps(cfg)) + except Exception as e: + _set_map_status(state="failed", uploaded=False, error=f"map scan failed: {e}") + return 0 if not maps: - log.info("no .db maps found under %s (robot=%s)", cfg.maps_dir, cfg.robot) + _set_map_status(state="no_map", uploaded=False, maps_found=0, last_map=None, + error=f"no saved map found in Sanad dashboard " + f"(maps_dir={cfg.maps_dir}, robot={cfg.robot})") return 0 state = load_state(cfg) - uploaded = 0 + uploaded = failed = 0 + last_err: Optional[str] = None + now = time.time() for m in maps: prev = state.get(str(m.path.resolve())) if not force and prev == m.fingerprint(): - log.debug("unchanged, skip: %s", m.name) + continue # already uploaded this exact content — one-time rule + # stability guard: a db modified in the last 120 s is still being + # written (active mapping) — wait until it settles before uploading + if not force and (now - m.mtime) < 120: + log.info("map %s still changing (mapping in progress) — waiting to settle", m.name) continue - # confirm change with a real content hash (mtime can shift without edits) - m.sha256 = _sha256(m.path) + m.sha256 = (_sha256_set(list(m.files.values())) + if m.fmt == "slam_toolbox" else _sha256(m.path)) m.points = load_points(cfg, m.stem) - if dry_run: - log.info("[dry-run] would upload %s (%.2f MB, sha=%s…, %d points)", - m.name, m.size / 1024 / 1024, m.sha256[:12], len(m.points)) - uploaded += 1 + log.info("[dry-run] would upload map %s (%.2f MB, %d points)", + m.name, m.size / 1024 / 1024, len(m.points)) continue - if upload_map(cfg, m, session): state[str(m.path.resolve())] = m.fingerprint() save_state(cfg, state) uploaded += 1 + else: + failed += 1 + last_err = f"upload failed for {m.name} (see agent log)" - if uploaded == 0: - log.info("nothing to upload (%d map(s) already current)", len(maps)) + if failed: + _set_map_status(state="failed", uploaded=False, maps_found=len(maps), + last_map=maps[0].stem, error=last_err) + else: + # every discovered map is on the server (just now or previously) + _set_map_status(state="uploaded", uploaded=True, maps_found=len(maps), + last_map=maps[0].stem, error=None) return uploaded +def map_loop(cfg: Config, session: requests.Session) -> None: + while True: + try: + map_sync_once(cfg, session) + except Exception as e: + log.exception("map pass failed: %s", e) + time.sleep(cfg.map_poll_interval) + + +# --------------------------------------------------------------------------- # +# telemetry assembly +# --------------------------------------------------------------------------- # +def derive_faults(cfg: Config, snap: Dict[str, Any]) -> List[Dict[str, Any]]: + # STRINGS, not objects — the fleet ingest 500s on fault objects. + faults: List[str] = [] + bms = snap.get("bms") + if bms and bms.get("soc", 100) <= cfg.low_soc: + faults.append(f"LOW_BATTERY: battery {bms['soc']}% (warning)") + temps = snap.get("temps") or [] + if temps and max(temps) >= cfg.motor_temp_max: + faults.append(f"MOTOR_OVERTEMP: motor temp {max(temps):.0f}C (warning)") + if snap.get("low_age") is not None and snap["low_age"] > 3.0: + faults.append(f"COMMS_STALE: no rt/lowstate for {snap['low_age']:.0f}s (critical)") + return faults + + +def derive_status(cfg: Config, snap: Dict[str, Any], fsm: Optional[int]) -> str: + base = _FSM_STATUS.get(fsm) if fsm is not None else None + bms = snap.get("bms") + charging = bool(bms and bms.get("current_a", 0.0) > 0.05) + alive = snap.get("low_age") is not None and snap["low_age"] <= 3.0 + if not alive and bms is None: + return "offline" + if charging: + return "charging" + if snap.get("max_dq", 0.0) > 0.15: + return "moving" + return base or "idle" + + +def build_telemetry(cfg: Config, mac: str, reader: Optional[DDSReader], + pos: Optional[RosbridgePosition], + sim: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: + if sim is not None: + snap = {"bms": {"soc": sim["battery"], "current_a": 0.5 if sim["charging"] else -0.3, + "voltage_v": 47.5, "temp_c": 36, "soh": 100, "cycles": 45}, + "low_age": 0.1, "temps": [sim.get("temp", 45)], "max_dq": sim.get("max_dq", 0.0), + "xy": sim.get("position")} + fsm = sim.get("fsm") + else: + snap = reader.snapshot() if reader else {"bms": None, "low_age": None, "temps": [], "max_dq": 0.0, "xy": None} + fsm = reader.fsm_id() if (reader and cfg.read_fsm) else None + + bms = snap.get("bms") + battery = bms["soc"] if bms else None + charging = bool(bms and bms.get("current_a", 0.0) > 0.05) + status = derive_status(cfg, snap, fsm) + faults = derive_faults(cfg, snap) + + battery_detail = None + if bms: + battery_detail = {"voltage_v": bms.get("voltage_v"), "current_a": bms.get("current_a"), + "temp_c": bms.get("temp_c"), "soh": bms.get("soh"), + "cycles": bms.get("cycles")} + + temps = snap.get("temps") or [] + motor_temp = ({"max": round(max(temps), 1), "avg": round(sum(temps) / len(temps), 1), + "min": round(min(temps), 1)} if temps else None) + + position = snap.get("xy") + if position is None and pos is not None: + position = pos.get() + + return { + "sn": cfg.sn, + "name": cfg.name, # friendly display name (e.g. g1_58) + "mac": mac, + "brand": cfg.brand, + "type": cfg.robot_type, # humanoid + "model": cfg.model, # g1 + "battery": battery, # null = couldn't read (heartbeat) + "charging": charging, + "battery_detail": battery_detail, + "motor_temp": motor_temp, # null = not receiving + "storage": read_storage(cfg), + "status": status, + "position": position, # null when no odom/localization source + "faults": faults, + "map": get_map_status(), # SHOWS whether the saved map made it to the server + "ts": int(time.time()), + } + + +def post_telemetry(cfg: Config, payload: Dict[str, Any], session: requests.Session) -> bool: + try: + r = session.post(cfg.telemetry_url(), json=payload, headers=cfg.auth_headers(), + timeout=cfg.http_timeout, verify=cfg.verify_tls) + except requests.RequestException as e: + log.error("telemetry POST failed (transport): %s", e) + return False + if not r.ok: + log.error("telemetry POST failed: HTTP %s %s", r.status_code, r.text[:200]) + return False + mp = payload.get("map") or {} + log.info("telemetry ok: battery=%s charging=%s status=%s pos=%s faults=%d map=%s -> HTTP %s", + payload["battery"], payload["charging"], payload["status"], + payload["position"], len(payload["faults"]), mp.get("state"), r.status_code) + return True + + +def _sim_state(i: int) -> Dict[str, Any]: + charging = (i % 6) in (0, 1) + battery = max(5, 90 - (i % 40)) + moving = (i % 3) == 2 and not charging + return {"battery": battery, "charging": charging, "temp": 45 + (i % 10), + "max_dq": 0.4 if moving else 0.0, "fsm": 200 if moving else 4, + "position": {"x": round(1.0 + 0.1 * i, 2), "y": round(2.0 - 0.05 * i, 2)}} + + def cmd_list(cfg: Config) -> None: maps = discover_maps(cfg) - active = _active_map_name(cfg) if not maps: print(f"(no .db maps under {cfg.maps_dir} for robot '{cfg.robot}')") return print(f"{len(maps)} map(s) under {cfg.maps_dir} (robot={cfg.robot}):") for m in maps: pts = load_points(cfg, m.stem) - flag = " <-- active" if active and _map_key(m.stem) == active else "" - print(f" {m.name:<28} {m.size/1024/1024:6.2f} MB {len(pts):>3} points" - f" {m.description}{flag}") + print(f" {m.name:<28} {m.size/1024/1024:6.2f} MB {len(pts):>3} points {m.description}") # --------------------------------------------------------------------------- # # main # --------------------------------------------------------------------------- # def main(argv: Optional[List[str]] = None) -> int: - ap = argparse.ArgumentParser(description="G1 fleet map uploader") - ap.add_argument("--once", action="store_true", help="one scan+upload pass, then exit") + ap = argparse.ArgumentParser(description="G1 fleet agent: telemetry + map sync") + ap.add_argument("--simulate", action="store_true", help="synthetic DDS state (map scan stays real)") + ap.add_argument("--once", action="store_true", help="one map pass + one telemetry post, then exit") + ap.add_argument("--dry-run", action="store_true", help="build payloads, never POST") + ap.add_argument("--force", action="store_true", help="re-upload maps even if unchanged") ap.add_argument("--list", action="store_true", help="list discovered maps and exit") - ap.add_argument("--dry-run", action="store_true", help="build payloads but never POST") - ap.add_argument("--force", action="store_true", help="upload even if unchanged") - ap.add_argument("--interval", type=float, default=None, help="override POLL_INTERVAL seconds") + ap.add_argument("--interval", type=float, default=None, help="override telemetry POLL_INTERVAL") ap.add_argument("-v", "--verbose", action="store_true") args = ap.parse_args(argv) - logging.basicConfig( - level=logging.DEBUG if args.verbose else logging.INFO, - format="%(asctime)s %(levelname)s %(name)s: %(message)s", - ) + logging.basicConfig(level=logging.DEBUG if args.verbose else logging.INFO, + format="%(asctime)s %(levelname)s %(name)s: %(message)s") _load_dotenv() cfg = Config.from_env() if args.interval is not None: cfg.poll_interval = args.interval - log.info("sanad_api_g1 map uploader — sn=%s robot=%s server=%s mode=%s select=%s", - cfg.sn, cfg.robot, cfg.server_url, cfg.upload_mode, cfg.map_select) - log.info("maps_dir=%s data_dir=%s web_nav3=%s", - cfg.maps_dir, cfg.data_dir, cfg.web_nav3_url or "(disabled)") - if args.list: cmd_list(cfg) return 0 + mac = read_mac(cfg.mac_interface) + log.info("sanad_api_g1 — sn=%s name=%s mac=%s server=%s iface=%s pos=%s " + "map_dir=%s map_every=%.0fs%s", + cfg.sn, cfg.name, mac, cfg.server_url, cfg.dds_interface, + cfg.position_source, cfg.maps_dir, cfg.map_poll_interval, + " [SIMULATE]" if args.simulate else "") + + reader = None + pos = None + if not args.simulate: + reader = DDSReader(cfg) + if cfg.position_source == "rosbridge": + pos = RosbridgePosition(cfg) + time.sleep(1.0) + session = requests.Session() + tick = 0 + + def one_telemetry() -> None: + nonlocal tick + sim = _sim_state(tick) if args.simulate else None + payload = build_telemetry(cfg, mac, reader, pos, sim=sim) + if args.dry_run: + log.info("[dry-run] %s", json.dumps(payload)) + else: + post_telemetry(cfg, payload, session) + tick += 1 + if args.once or args.dry_run: - run_once(cfg, force=args.force, dry_run=args.dry_run, session=session) + # one map pass first so the telemetry "map" field reflects it + map_sync_once(cfg, session, force=args.force, dry_run=args.dry_run) + if args.once: + one_telemetry() + return 0 + for _ in range(3): + one_telemetry() + time.sleep(min(cfg.poll_interval, 1.0)) return 0 - log.info("loop every %.0fs (Ctrl-C to stop)", cfg.poll_interval) + # loop mode: map sync in a background thread, telemetry in the main loop + threading.Thread(target=map_loop, args=(cfg, session), daemon=True).start() + log.info("telemetry every %.1fs; map check every %.0fs (Ctrl-C to stop)", + cfg.poll_interval, cfg.map_poll_interval) while True: try: - run_once(cfg, force=args.force, dry_run=False, session=session) - except Exception as e: # never let the loop die - log.exception("pass failed: %s", e) + one_telemetry() + except Exception as e: + log.exception("telemetry tick failed: %s", e) try: time.sleep(cfg.poll_interval) except KeyboardInterrupt: diff --git a/agents/g1t/vendor/crc_aarch64.so b/agents/g1/vendor/crc_aarch64.so similarity index 100% rename from agents/g1t/vendor/crc_aarch64.so rename to agents/g1/vendor/crc_aarch64.so diff --git a/agents/g1t/vendor/crc_amd64.so b/agents/g1/vendor/crc_amd64.so similarity index 100% rename from agents/g1t/vendor/crc_amd64.so rename to agents/g1/vendor/crc_amd64.so diff --git a/agents/g1t/vendor/unitree_sdk2py-1.0.1-py3-none-any.whl b/agents/g1/vendor/unitree_sdk2py-1.0.1-py3-none-any.whl similarity index 100% rename from agents/g1t/vendor/unitree_sdk2py-1.0.1-py3-none-any.whl rename to agents/g1/vendor/unitree_sdk2py-1.0.1-py3-none-any.whl diff --git a/agents/g1t/.dockerignore b/agents/g1t/.dockerignore deleted file mode 100644 index bce1cad..0000000 --- a/agents/g1t/.dockerignore +++ /dev/null @@ -1,4 +0,0 @@ -.env -__pycache__/ -*.pyc -README.md diff --git a/agents/g1t/.env.example b/agents/g1t/.env.example deleted file mode 100644 index 9771d6a..0000000 --- a/agents/g1t/.env.example +++ /dev/null @@ -1,43 +0,0 @@ -# sanad_api_g1t — copy to .env and fill in. (G1 telemetry; sibling of the G1 map agent.) - -# ── fleet server (REQUIRED — YS Lootah gives you these two) ────────────────── -SERVER_URL=https://fleet.example.com -DEVICE_TOKEN=REPLACE_WITH_DEVICE_TOKEN - -# ── identity ───────────────────────────────────────────────────────────────── -# Use the SAME sn as the G1 map agent — it's the same physical robot. -SN=g1_7892 - -# ── DDS (reading robot state — unitree_hg, same family as R1) ──────────────── -# G1 DDS link is usually eth0. -DDS_INTERFACE=eth0 -DDS_DOMAIN=0 -# MAC_INTERFACE=eth0 - -# ── status / position ──────────────────────────────────────────────────────── -# Read the loco FSM for a richer status (READ-ONLY GET RPC, never moves). -# 0 = derive status from battery + joint motion (safe default). -G1_READ_FSM=0 -# Position source: -# odom (default) — firmware leg-odometry x,y from rt/lf/odommodestate (drifts, resets at boot) -# rosbridge — map-frame /odom over rosbridge (needs the nav bringup + ROSBRIDGE_URL) -# none — omit position -G1_POSITION_SOURCE=odom -ROSBRIDGE_URL=ws://127.0.0.1:9090 - -# ── fault thresholds ───────────────────────────────────────────────────────── -LOW_SOC=15 -MOTOR_TEMP_MAX=85 - -# ── cadence / transport ────────────────────────────────────────────────────── -POLL_INTERVAL=2 -VERIFY_TLS=1 -HTTP_TIMEOUT=10 - -# ── robot identity card (shown on the dashboard) ───────────────────────────── -ROBOT_BRAND=unitree -ROBOT_TYPE=humanoid -ROBOT_MODEL=g1 -# Optional: Sanad data dir whose size is reported in storage (as /host/ -# when the installer's read-only /:/host mount is used). Empty = omit. -STORAGE_DATA_PATH= diff --git a/agents/g1t/.gitignore b/agents/g1t/.gitignore deleted file mode 100644 index cff5543..0000000 --- a/agents/g1t/.gitignore +++ /dev/null @@ -1,3 +0,0 @@ -.env -__pycache__/ -*.pyc diff --git a/agents/g1t/Dockerfile b/agents/g1t/Dockerfile deleted file mode 100644 index 931fa83..0000000 --- a/agents/g1t/Dockerfile +++ /dev/null @@ -1,46 +0,0 @@ -# sanad_api_g1t — G1 fleet telemetry agent (DDS: rt/lf/bmsstate + rt/lowstate + odom). -# -# BUILD ON THE G1 JETSON (native arm64). DDS stack: prebuilt CycloneDDS (apt) + -# the CycloneDDS python binding (pip) + the vendored unitree_sdk2py wheel. Run with -# --network host so the robot's DDS is visible (see docker-compose.yml). -FROM python:3.10-slim-bookworm - -# CycloneDDS C lib + idlc (bookworm ships 0.10.2 — matches the robot), build tools -# for the python binding, iproute2 for iface checks, libgomp1 for the TLS guard. -RUN apt-get update && apt-get install -y --no-install-recommends \ - build-essential cmake \ - cyclonedds-dev cyclonedds-tools \ - iproute2 libgomp1 ca-certificates \ - && rm -rf /var/lib/apt/lists/* - -# aarch64 "static TLS block" guard (harmless on amd64). -ENV LD_PRELOAD=libgomp.so.1 -# cyclonedds build helper looks for libddsc.so under $CYCLONEDDS_HOME/lib; Debian -# installs into the multiarch dir — symlink whatever arch built into /usr/lib. -ENV CYCLONEDDS_HOME=/usr -RUN set -e; for lib in libddsc.so libcycloneddsidl.so; do \ - f=$(ls /usr/lib/*/"$lib" 2>/dev/null | head -1); \ - [ -n "$f" ] && ln -sf "$f" /usr/lib/"$lib" || true; \ - done - -WORKDIR /app -COPY requirements.txt . -RUN pip install --no-cache-dir -r requirements.txt -# CycloneDDS python binding, compiled against the C lib above. 0.10.2's build -# helper imports wheel.bdist_wheel (removed in wheel>=0.46) → pin the toolchain. -RUN pip install --no-cache-dir "setuptools<80" "wheel<0.46" \ - && pip install --no-cache-dir --no-build-isolation cyclonedds==0.10.2 - -# Unitree SDK (vendored wheel — not on PyPI) + native crc lib (wheel omits it). -# Both arch crc libs are copied; unitree_sdk2py loads the one matching the image. -COPY vendor/unitree_sdk2py-*.whl /tmp/ -RUN pip install --no-cache-dir --no-deps /tmp/unitree_sdk2py-*.whl -COPY vendor/crc_aarch64.so vendor/crc_amd64.so \ - /usr/local/lib/python3.10/site-packages/unitree_sdk2py/utils/lib/ - -COPY sanad_api_g1t.py . - -ENV PYTHONUNBUFFERED=1 \ - DDS_INTERFACE=eth0 \ - DDS_DOMAIN=0 -ENTRYPOINT ["python", "-u", "sanad_api_g1t.py"] diff --git a/agents/g1t/README.md b/agents/g1t/README.md deleted file mode 100644 index 212833f..0000000 --- a/agents/g1t/README.md +++ /dev/null @@ -1,40 +0,0 @@ -# sanad_api_g1t — G1 fleet **telemetry** agent - -Streams the Unitree **G1**'s live status to the YS Lootah fleet server. It's the -**telemetry** sibling of the G1 **map** agent (`../g1`) — the G1 runs both, each -its own container/service, both keyed by the **same `sn`** (same physical robot). - -``` -POST {SERVER_URL}/api/v1/fleet/ingest/telemetry Authorization: Bearer -{ "sn":"g1_7892", "mac":"…", "battery":74, "charging":false, - "status":"idle", "position":{"x":…,"y":…}|null, "faults":[], "ts":… } -``` - -Every ~2 s; a **heartbeat** (`battery:null, status:offline`) when DDS is silent. - -## Data sources (G1, `unitree_hg` DDS — same family as R1) - -| field | source | -|---|---| -| `battery`, `charging` | `rt/lf/bmsstate` (`BmsState_`): `soc` 0–100; charging from `current` sign | -| `faults[]` | `rt/lowstate` (`LowState_`): motor temps + staleness | -| `position` `{x,y}` | **`rt/lf/odommodestate`** (`SportModeState_`) — firmware leg-odometry over DDS, no ROS (odom frame: drifts, resets at boot). Optional map-frame via `rosbridge`. | -| `status` | derived (charging/moving/idle/offline); optional loco **FSM** (`G1_READ_FSM=1`, ids **200** ready / 4 stand / 2 squat / 702 lie2stand — GET-only) | -| `mac` | primary NIC | - -> **Safety:** read-only — never commands motion. - -## Run / test - -```bash -# via the fleet installer (adds a 'g1t' type): -../../fleet_install.sh install g1t --sn g1_7892 -../../fleet_install.sh data g1t - -# bare simulate (no robot): -SERVER_URL=… DEVICE_TOKEN=… SN=g1_7892 python3 sanad_api_g1t.py --simulate -v -``` - -Build on the G1 Jetson (arm64); bundles CycloneDDS + the vendored `unitree_sdk2py` -wheel (`vendor/`). `network_mode: host` for DDS. Coexists with the map agent and -with a running Sanad (DDS allows many readers of the same topics). diff --git a/agents/g1t/docker-compose.yml b/agents/g1t/docker-compose.yml deleted file mode 100644 index 470080d..0000000 --- a/agents/g1t/docker-compose.yml +++ /dev/null @@ -1,20 +0,0 @@ -# sanad_api_g1t — standalone G1 fleet TELEMETRY agent. -# (The fleet installer deploys via docker build/run + systemd; this compose is -# for manual local use.) -# -# cp .env.example .env # SERVER_URL + DEVICE_TOKEN + SN + DDS_INTERFACE -# docker compose up -d --build -# -# Reads the G1's DDS state (rt/lf/bmsstate + rt/lowstate + rt/lf/odommodestate) -# and POSTs telemetry every ~2s. network_mode: host is REQUIRED for DDS. -services: - sanad-api-g1t: - build: - context: . - dockerfile: Dockerfile - image: "${SANAD_API_G1T_IMAGE:-sanad-api-g1t:latest}" - container_name: sanad-api-g1t - network_mode: host - restart: unless-stopped - env_file: .env - # command: ["--simulate"] # synthetic state, no robot diff --git a/agents/g1t/requirements.txt b/agents/g1t/requirements.txt deleted file mode 100644 index b462572..0000000 --- a/agents/g1t/requirements.txt +++ /dev/null @@ -1,7 +0,0 @@ -# sanad_api_r1 telemetry agent. -requests>=2.31,<3 -# only needed if R1_POSITION_SOURCE=rosbridge (reads /odom for position): -websocket-client>=1.6,<2 -# DDS stack (cyclonedds + the vendored unitree_sdk2py wheel) is installed by the -# Dockerfile, not from here — see Dockerfile. The agent degrades to heartbeats if -# unitree_sdk2py is unavailable. diff --git a/agents/g1t/sanad_api_g1t.py b/agents/g1t/sanad_api_g1t.py deleted file mode 100644 index 5f931d7..0000000 --- a/agents/g1t/sanad_api_g1t.py +++ /dev/null @@ -1,572 +0,0 @@ -#!/usr/bin/env python3 -"""sanad_api_g1t — G1 fleet TELEMETRY agent. - -Pushes the Unitree **G1**'s live status to the YS Lootah fleet server: - - POST {SERVER_URL}/api/v1/fleet/ingest/telemetry (Bearer device token) - body: { "sn", "mac", "battery", "charging", "status", "position":{x,y}, "faults":[] } - -This is the telemetry sibling of the G1 MAP uploader (sanad_api_g1) — the G1 runs -both, each its own container/service, both keyed by the SAME `sn` (same robot). - -DATA SOURCES (Unitree G1, unitree_hg DDS — same family as the R1) ----------------------------------------------------------------- - battery / charging : rt/lf/bmsstate (BmsState_) soc 0-100; charging = current>+0.05A - faults / liveness : rt/lowstate (LowState_) motor temps + message staleness - position {x,y} : rt/lf/odommodestate (SportModeState_, unitree_go idl) — - firmware leg-odometry x,y over DDS, no ROS. Odom frame: - resets at boot, drifts. (Map-frame via rosbridge/TF optional.) - status : G1 loco FSM via GET RPC 7001 (ids 200 walk-ready / 4 stand / - 2 squat / 702 lie2stand) — READ-ONLY, optional (G1_READ_FSM=1). - Default derives status from BMS + motion. - mac : primary NIC hardware address. - -NO ROS. DDS via unitree_sdk2py. Degrades to heartbeats if unitree_sdk2py is -unavailable; --simulate tests the upload path without a robot. Read-only: never -commands motion (only the GET_FSM_ID RPC is ever issued). - -CONFIG — environment (see .env.example) ---------------------------------------- - SERVER_URL, DEVICE_TOKEN fleet base URL + bearer token (required) - SN this robot's fleet id (SAME as the G1 map agent) default g1_0000 - DDS_INTERFACE robot network iface for DDS default eth0 - DDS_DOMAIN DDS domain id default 0 - MAC_INTERFACE iface whose MAC to report default = DDS_INTERFACE - G1_READ_FSM 1 = read loco FSM for status default 0 - G1_POSITION_SOURCE odom | rosbridge | none default odom - ROSBRIDGE_URL ws://127.0.0.1:9090 (position) default ws://127.0.0.1:9090 - LOW_SOC / MOTOR_TEMP_MAX fault thresholds default 15 / 85 - POLL_INTERVAL seconds between posts default 2 - VERIFY_TLS / HTTP_TIMEOUT TLS verify (1) / timeout (10) - -CLI: --simulate | --once | --dry-run | --interval N | -v -""" -from __future__ import annotations - -import argparse -import json -import logging -import os -import shutil -import sys -import threading -import time -import uuid -from dataclasses import dataclass -from pathlib import Path -from typing import Any, Dict, List, Optional - -import requests - -log = logging.getLogger("sanad_api_g1t") - - -def _load_dotenv(path: str = ".env") -> None: - p = Path(path) - if not p.exists(): - return - for line in p.read_text().splitlines(): - line = line.strip() - if not line or line.startswith("#") or "=" not in line: - continue - k, _, v = line.partition("=") - os.environ.setdefault(k.strip(), v.strip().strip('"').strip("'")) - - -def _env(name: str, default: str = "") -> str: - return os.environ.get(name, default).strip() - - -def _env_bool(name: str, default: bool) -> bool: - return _env(name, "1" if default else "0").lower() in ("1", "true", "yes", "on") - - -@dataclass -class Config: - server_url: str - device_token: str - sn: str - name: str - brand: str - robot_type: str - model: str - storage_path: str - data_path: str - dds_interface: str - dds_domain: int - mac_interface: str - read_fsm: bool - position_source: str - rosbridge_url: str - low_soc: int - motor_temp_max: float - poll_interval: float - endpoint: str - verify_tls: bool - http_timeout: float - - @classmethod - def from_env(cls) -> "Config": - server = _env("SERVER_URL").rstrip("/") - token = _env("DEVICE_TOKEN") - missing = [n for n, v in (("SERVER_URL", server), ("DEVICE_TOKEN", token)) if not v] - if missing: - raise SystemExit(f"[config] missing required env: {', '.join(missing)}") - iface = _env("DDS_INTERFACE", "eth0") - return cls( - server_url=server, device_token=token, - sn=_env("SN", "g1_0000"), - name=_env("ROBOT_NAME", "") or _env("SN", "g1_0000"), - brand=_env("ROBOT_BRAND", "unitree"), - robot_type=_env("ROBOT_TYPE", "humanoid"), - model=_env("ROBOT_MODEL", "g1"), - storage_path=_env("STORAGE_PATH", ""), - data_path=_env("STORAGE_DATA_PATH", ""), - dds_interface=iface, dds_domain=int(_env("DDS_DOMAIN", "0")), - mac_interface=_env("MAC_INTERFACE", iface), - read_fsm=_env_bool("G1_READ_FSM", False), - position_source=_env("G1_POSITION_SOURCE", "odom").lower(), - rosbridge_url=_env("ROSBRIDGE_URL", "ws://127.0.0.1:9090"), - low_soc=int(_env("LOW_SOC", "15")), - motor_temp_max=float(_env("MOTOR_TEMP_MAX", "85")), - poll_interval=float(_env("POLL_INTERVAL", "2")), - endpoint=_env("TELEMETRY_ENDPOINT", "/api/v1/fleet/ingest/telemetry"), - verify_tls=_env_bool("VERIFY_TLS", True), - http_timeout=float(_env("HTTP_TIMEOUT", "10")), - ) - - def telemetry_url(self) -> str: - return self.server_url + self.endpoint - - def auth_headers(self) -> Dict[str, str]: - return {"Authorization": f"Bearer {self.device_token}"} - - -def read_mac(interface: str) -> str: - p = Path(f"/sys/class/net/{interface}/address") - try: - mac = p.read_text().strip() - if mac and mac != "00:00:00:00:00:00": - return mac.lower() - except Exception: - pass - n = uuid.getnode() - return ":".join(f"{(n >> b) & 0xff:02x}" for b in range(40, -1, -8)) - - -_data_size_cache: Dict[str, Any] = {"ts": 0.0, "kb": None} - - -def read_storage(cfg: Config) -> Optional[Dict[str, Any]]: - """Disk usage of the robot's root fs + optional Sanad data-dir size. - - In docker, bind-mount the host root read-only at /host (the installer does) - so this reports the HOST disk, not the container overlay.""" - root = cfg.storage_path or ("/host" if os.path.isdir("/host") else "/") - try: - du = shutil.disk_usage(root) - out: Dict[str, Any] = { - "total_gb": round(du.total / 1e9, 2), - "free_gb": round(du.free / 1e9, 2), - "used_percent": round(du.used / du.total * 100, 1), - } - except Exception: - return None - if cfg.data_path and os.path.isdir(cfg.data_path): - now = time.monotonic() - if _data_size_cache["kb"] is None or now - _data_size_cache["ts"] > 60: - try: - total = 0 - for r, _, files in os.walk(cfg.data_path): - for f in files: - try: - total += os.path.getsize(os.path.join(r, f)) - except OSError: - pass - _data_size_cache.update(ts=now, kb=round(total / 1024, 1)) - except Exception: - pass - if _data_size_cache["kb"] is not None: - out["data_kb"] = _data_size_cache["kb"] - return out - - -class DDSReader: - """Subscribes rt/lf/bmsstate + rt/lowstate (+ rt/lf/odommodestate for position). - Passive reads; the only RPC ever issued is GET_FSM_ID.""" - - def __init__(self, cfg: Config): - self.cfg = cfg - self._lock = threading.Lock() - self._bms: Optional[Dict[str, Any]] = None - self._bms_ts = 0.0 - self._low_ts = 0.0 - self._temps: List[float] = [] - self._max_dq = 0.0 - self._xy: Optional[Dict[str, float]] = None - self._loco = None - self.ok = False - self._start() - - def _start(self) -> None: - try: - from unitree_sdk2py.core.channel import ( - ChannelFactoryInitialize, ChannelSubscriber) - from unitree_sdk2py.idl.unitree_hg.msg.dds_ import LowState_ - try: - from unitree_sdk2py.idl.unitree_hg.msg.dds_ import BmsState_ - except Exception: - BmsState_ = None - SportModeState_ = None - if self.cfg.position_source == "odom": - try: - # G1 firmware publishes odom as unitree_go SportModeState_. - from unitree_sdk2py.idl.unitree_go.msg.dds_ import SportModeState_ - except Exception: - SportModeState_ = None - except Exception as e: - log.warning("unitree_sdk2py unavailable (%s) — telemetry runs in heartbeat mode", e) - return - try: - ChannelFactoryInitialize(self.cfg.dds_domain, self.cfg.dds_interface) - self._low_sub = ChannelSubscriber("rt/lowstate", LowState_) - self._low_sub.Init(self._on_low, 10) - if BmsState_ is not None: - self._bms_sub = ChannelSubscriber("rt/lf/bmsstate", BmsState_) - self._bms_sub.Init(self._on_bms, 10) - else: - log.warning("BmsState_ not in this unitree_sdk2py — battery will be null") - if SportModeState_ is not None: - self._odom_sub = ChannelSubscriber("rt/lf/odommodestate", SportModeState_) - self._odom_sub.Init(self._on_odom, 10) - if self.cfg.read_fsm: - self._init_loco() - self.ok = True - log.info("DDS up: domain=%d iface=%s (bmsstate + lowstate%s)", - self.cfg.dds_domain, self.cfg.dds_interface, - " + odom" if SportModeState_ is not None else "") - except Exception as e: - log.warning("DDS init failed (%s) — heartbeat mode", e) - - def _init_loco(self) -> None: - try: - from unitree_sdk2py.rpc.client import Client # type: ignore - except Exception as e: - log.warning("loco RPC client unavailable (%s) — status from BMS/motion only", e) - return - try: - c = Client("loco", 0); c.Init(); c.SetTimeout(3.0) - self._loco = c - log.info("loco FSM read enabled (GET-only, no motion)") - except Exception as e: - log.warning("loco client init failed (%s) — status from BMS/motion only", e) - self._loco = None - - def _on_bms(self, msg) -> None: - try: - soc = int(getattr(msg, "soc", 0) or 0) - cur_mA = int(getattr(msg, "current", 0) or 0) - # Pack voltage: prefer bmsvoltage[0] (mV); else sum of cell voltages. - volt_mv = 0 - bv = getattr(msg, "bmsvoltage", None) - try: - if bv is not None and len(bv) and int(bv[0]): - volt_mv = int(bv[0]) - except Exception: - volt_mv = 0 - if not volt_mv: - cv = getattr(msg, "cell_vol", None) - if cv is not None: - try: - volt_mv = int(sum(int(x) for x in cv if x)) - except Exception: - volt_mv = 0 - # Max plausible pack temperature (int16 °C). - temp_c = None - tt = getattr(msg, "temperature", None) - if tt is not None: - try: - vals = [int(x) for x in tt if -40 <= int(x) <= 150] - if vals: - temp_c = max(vals) - except Exception: - temp_c = None - with self._lock: - self._bms = { - "soc": max(0, min(100, soc)), - "current_a": round(cur_mA / 1000.0, 2), - "voltage_v": round(volt_mv / 1000.0, 1) if volt_mv else None, - "temp_c": temp_c, - "soh": int(getattr(msg, "soh", 0) or 0), - "cycles": int(getattr(msg, "cycle", 0) or 0), - } - self._bms_ts = time.monotonic() - except Exception: - pass - - def _on_low(self, msg) -> None: - try: - temps: List[float] = [] - max_dq = 0.0 - for m in (getattr(msg, "motor_state", None) or []): - t = getattr(m, "temperature", None) - if t is not None: - try: - vals = [float(x) for x in t] if hasattr(t, "__iter__") else [float(t)] - # 0 = slot not reporting (unpopulated motor), not a real temp - temps.extend(v for v in vals if 0 < v <= 200) - except Exception: - pass - dq = getattr(m, "dq", None) - if dq is not None: - try: - max_dq = max(max_dq, abs(float(dq))) - except Exception: - pass - with self._lock: - self._low_ts = time.monotonic() - self._temps = temps - self._max_dq = max_dq - except Exception: - pass - - def _on_odom(self, msg) -> None: - try: - pos = getattr(msg, "position", None) - if pos is not None and len(pos) >= 2: - with self._lock: - self._xy = {"x": round(float(pos[0]), 3), "y": round(float(pos[1]), 3)} - except Exception: - pass - - def snapshot(self) -> Dict[str, Any]: - with self._lock: - now = time.monotonic() - return { - "bms": dict(self._bms) if self._bms else None, - "low_age": (now - self._low_ts) if self._low_ts else None, - "temps": list(self._temps), - "max_dq": self._max_dq, - "xy": dict(self._xy) if self._xy else None, - } - - def fsm_id(self) -> Optional[int]: - if not self._loco: - return None - try: - code, data = self._loco._Call(7001, "{}") # GET_FSM_ID — read-only - if code == 0 and data: - return int(json.loads(data).get("data", data)) if data.strip().startswith("{") else int(data) - except Exception as e: - log.debug("fsm read failed: %s", e) - return None - - -# G1 FSM ids (differ from the R1's): 200 balance/walk-ready, 4 StandUp, 2 Squat, 702 Lie2Stand. -_FSM_STATUS = {200: "ready", 4: "standing", 2: "squat", 702: "lie2stand"} - - -class RosbridgePosition: - def __init__(self, cfg: Config): - self.cfg = cfg - self._xy: Optional[Dict[str, float]] = None - self._lock = threading.Lock() - self._stop = False - try: - import websocket # noqa: F401 - except Exception as e: - log.warning("websocket-client absent (%s) — rosbridge position disabled", e) - self._ok = False - return - self._ok = True - threading.Thread(target=self._run, daemon=True).start() - - def _run(self) -> None: - import websocket - sub = json.dumps({"op": "subscribe", "topic": "/odom", - "type": "nav_msgs/Odometry", "throttle_rate": 500}) - while not self._stop: - try: - ws = websocket.create_connection(self.cfg.rosbridge_url, timeout=5) - ws.send(sub) - while not self._stop: - msg = json.loads(ws.recv()) - pos = (((msg.get("msg") or {}).get("pose") or {}).get("pose") or {}).get("position") - if pos: - with self._lock: - self._xy = {"x": round(float(pos["x"]), 3), "y": round(float(pos["y"]), 3)} - except Exception as e: - log.debug("rosbridge position reconnect: %s", e) - time.sleep(3) - - def get(self) -> Optional[Dict[str, float]]: - with self._lock: - return dict(self._xy) if self._xy else None - - -def derive_faults(cfg: Config, snap: Dict[str, Any]) -> List[Dict[str, Any]]: - faults: List[Dict[str, Any]] = [] - bms = snap.get("bms") - if bms and bms.get("soc", 100) <= cfg.low_soc: - faults.append({"code": "LOW_BATTERY", "severity": "warning", "message": f"battery {bms['soc']}%"}) - temps = snap.get("temps") or [] - if temps and max(temps) >= cfg.motor_temp_max: - faults.append({"code": "MOTOR_OVERTEMP", "severity": "warning", "message": f"motor temp {max(temps):.0f}C"}) - if snap.get("low_age") is not None and snap["low_age"] > 3.0: - faults.append({"code": "COMMS_STALE", "severity": "critical", "message": f"no rt/lowstate for {snap['low_age']:.0f}s"}) - return faults - - -def derive_status(cfg: Config, snap: Dict[str, Any], fsm: Optional[int]) -> str: - base = _FSM_STATUS.get(fsm) if fsm is not None else None - bms = snap.get("bms") - charging = bool(bms and bms.get("current_a", 0.0) > 0.05) - alive = snap.get("low_age") is not None and snap["low_age"] <= 3.0 - if not alive and bms is None: - return "offline" - if charging: - return "charging" - if snap.get("max_dq", 0.0) > 0.15: - return "moving" - return base or "idle" - - -def build_telemetry(cfg: Config, mac: str, reader: Optional[DDSReader], - pos: Optional[RosbridgePosition], - sim: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: - if sim is not None: - snap = {"bms": {"soc": sim["battery"], "current_a": 0.5 if sim["charging"] else -0.3, - "voltage_v": 47.5, "temp_c": 36, "soh": 100, "cycles": 45}, - "low_age": 0.1, "temps": [sim.get("temp", 45)], "max_dq": sim.get("max_dq", 0.0), - "xy": sim.get("position")} - fsm = sim.get("fsm") - else: - snap = reader.snapshot() if reader else {"bms": None, "low_age": None, "temps": [], "max_dq": 0.0, "xy": None} - fsm = reader.fsm_id() if (reader and cfg.read_fsm) else None - - bms = snap.get("bms") - battery = bms["soc"] if bms else None - charging = bool(bms and bms.get("current_a", 0.0) > 0.05) - status = derive_status(cfg, snap, fsm) - faults = derive_faults(cfg, snap) - - # Battery detail (voltage / current / pack temp / health / cycles). - battery_detail = None - if bms: - battery_detail = {"voltage_v": bms.get("voltage_v"), "current_a": bms.get("current_a"), - "temp_c": bms.get("temp_c"), "soh": bms.get("soh"), - "cycles": bms.get("cycles")} - - # Motor temperature stats; null = temps not receiving. - temps = snap.get("temps") or [] - motor_temp = ({"max": round(max(temps), 1), "avg": round(sum(temps) / len(temps), 1), - "min": round(min(temps), 1)} if temps else None) - - position = snap.get("xy") - if position is None and pos is not None: - position = pos.get() - - return { - "sn": cfg.sn, - "name": cfg.name, # friendly display name (e.g. g1_58) - "mac": mac, - "brand": cfg.brand, - "type": cfg.robot_type, # humanoid | dog - "model": cfg.model, # r1 | g1 | go2 - "battery": battery, "charging": charging, - "battery_detail": battery_detail, - "motor_temp": motor_temp, # null = not receiving - "storage": read_storage(cfg), - "status": status, - "position": position, "faults": faults, "ts": int(time.time()), - } - - -def post_telemetry(cfg: Config, payload: Dict[str, Any], session: requests.Session) -> bool: - try: - r = session.post(cfg.telemetry_url(), json=payload, headers=cfg.auth_headers(), - timeout=cfg.http_timeout, verify=cfg.verify_tls) - except requests.RequestException as e: - log.error("telemetry POST failed (transport): %s", e) - return False - if not r.ok: - log.error("telemetry POST failed: HTTP %s %s", r.status_code, r.text[:200]) - return False - log.info("telemetry ok: battery=%s charging=%s status=%s pos=%s faults=%d -> HTTP %s", - payload["battery"], payload["charging"], payload["status"], - payload["position"], len(payload["faults"]), r.status_code) - return True - - -def _sim_state(i: int) -> Dict[str, Any]: - charging = (i % 6) in (0, 1) - battery = max(5, 90 - (i % 40)) - moving = (i % 3) == 2 and not charging - return {"battery": battery, "charging": charging, "temp": 45 + (i % 10), - "max_dq": 0.4 if moving else 0.0, "fsm": 200 if moving else 4, - "position": {"x": round(1.0 + 0.1 * i, 2), "y": round(2.0 - 0.05 * i, 2)}} - - -def main(argv: Optional[List[str]] = None) -> int: - ap = argparse.ArgumentParser(description="G1 fleet telemetry agent") - ap.add_argument("--simulate", action="store_true") - ap.add_argument("--once", action="store_true") - ap.add_argument("--dry-run", action="store_true") - ap.add_argument("--interval", type=float, default=None) - ap.add_argument("-v", "--verbose", action="store_true") - args = ap.parse_args(argv) - - logging.basicConfig(level=logging.DEBUG if args.verbose else logging.INFO, - format="%(asctime)s %(levelname)s %(name)s: %(message)s") - _load_dotenv() - cfg = Config.from_env() - if args.interval is not None: - cfg.poll_interval = args.interval - - mac = read_mac(cfg.mac_interface) - log.info("sanad_api_g1t telemetry — sn=%s mac=%s server=%s iface=%s domain=%d pos=%s%s", - cfg.sn, mac, cfg.server_url, cfg.dds_interface, cfg.dds_domain, - cfg.position_source, " [SIMULATE]" if args.simulate else "") - - reader = None - pos = None - if not args.simulate: - reader = DDSReader(cfg) - if cfg.position_source == "rosbridge": - pos = RosbridgePosition(cfg) - time.sleep(1.0) - - session = requests.Session() - tick = 0 - - def one() -> None: - nonlocal tick - sim = _sim_state(tick) if args.simulate else None - payload = build_telemetry(cfg, mac, reader, pos, sim=sim) - if args.dry_run: - log.info("[dry-run] %s", json.dumps(payload)) - else: - post_telemetry(cfg, payload, session) - tick += 1 - - if args.once: - one(); return 0 - if args.dry_run: - for _ in range(3): - one(); time.sleep(min(cfg.poll_interval, 1.0)) - return 0 - - log.info("loop every %.1fs (Ctrl-C to stop)", cfg.poll_interval) - while True: - try: - one() - except Exception as e: - log.exception("tick failed: %s", e) - try: - time.sleep(cfg.poll_interval) - except KeyboardInterrupt: - log.info("stopped"); return 0 - - -if __name__ == "__main__": - sys.exit(main()) diff --git a/agents/go2/sanad_api_go2.py b/agents/go2/sanad_api_go2.py index 582979f..5542c78 100644 --- a/agents/go2/sanad_api_go2.py +++ b/agents/go2/sanad_api_go2.py @@ -346,15 +346,16 @@ class RosbridgePosition: def derive_faults(cfg: Config, snap: Dict[str, Any]) -> List[Dict[str, Any]]: - faults: List[Dict[str, Any]] = [] + # STRINGS, not objects — the fleet ingest 500s on fault objects. + faults: List[str] = [] bms = snap.get("bms") if bms and bms.get("soc", 100) <= cfg.low_soc: - faults.append({"code": "LOW_BATTERY", "severity": "warning", "message": f"battery {bms['soc']}%"}) + faults.append(f"LOW_BATTERY: battery {bms['soc']}% (warning)") temps = snap.get("temps") or [] if temps and max(temps) >= cfg.motor_temp_max: - faults.append({"code": "MOTOR_OVERTEMP", "severity": "warning", "message": f"motor temp {max(temps):.0f}C"}) + faults.append(f"MOTOR_OVERTEMP: motor temp {max(temps):.0f}C (warning)") if snap.get("low_age") is not None and snap["low_age"] > 3.0: - faults.append({"code": "COMMS_STALE", "severity": "critical", "message": f"no rt/lowstate for {snap['low_age']:.0f}s"}) + faults.append(f"COMMS_STALE: no rt/lowstate for {snap['low_age']:.0f}s (critical)") return faults diff --git a/agents/r1/sanad_api_r1.py b/agents/r1/sanad_api_r1.py index 0733b77..3cb4e47 100644 --- a/agents/r1/sanad_api_r1.py +++ b/agents/r1/sanad_api_r1.py @@ -435,21 +435,17 @@ class RosbridgePosition: # --------------------------------------------------------------------------- # # telemetry assembly # --------------------------------------------------------------------------- # -def derive_faults(cfg: Config, snap: Dict[str, Any]) -> List[Dict[str, Any]]: - faults: List[Dict[str, Any]] = [] +def derive_faults(cfg: Config, snap: Dict[str, Any]) -> List[str]: + # STRINGS, not objects — the fleet ingest 500s on fault objects. + faults: List[str] = [] bms = snap.get("bms") if bms and bms.get("soc", 100) <= cfg.low_soc: - faults.append({"code": "LOW_BATTERY", "severity": "warning", - "message": f"battery {bms['soc']}%"}) + faults.append(f"LOW_BATTERY: battery {bms['soc']}% (warning)") temps = snap.get("temps") or [] - if temps: - hot = max(temps) - if hot >= cfg.motor_temp_max: - faults.append({"code": "MOTOR_OVERTEMP", "severity": "warning", - "message": f"motor temp {hot:.0f}C"}) + if temps and max(temps) >= cfg.motor_temp_max: + faults.append(f"MOTOR_OVERTEMP: motor temp {max(temps):.0f}C (warning)") if snap.get("low_age") is not None and snap["low_age"] > 3.0: - faults.append({"code": "COMMS_STALE", "severity": "critical", - "message": f"no rt/lowstate for {snap['low_age']:.0f}s"}) + faults.append(f"COMMS_STALE: no rt/lowstate for {snap['low_age']:.0f}s (critical)") return faults diff --git a/fleet_install.sh b/fleet_install.sh index a2b263f..d4a1861 100755 --- a/fleet_install.sh +++ b/fleet_install.sh @@ -6,11 +6,13 @@ # → pick robot type → enter IP → it detects whether the agent is already # installed and offers the right actions (install / uninstall / data / ...). # -# SCRIPTABLE: -# ./fleet_install.sh install [--sn NAME] [--server-ip IP] [--port N] [--token TOK] [--user U] +# SCRIPTABLE — install requires: ip, --sn, --token, --server-url (and optionally --post): +# ./fleet_install.sh install --sn --token \ +# --server-url https://fleet.example.com [--post /api/v1/fleet/ingest/telemetry] \ +# [--name NAME] [--user U] # ./fleet_install.sh uninstall [--user U] # ./fleet_install.sh status -# ./fleet_install.sh data # show the data it is sending +# ./fleet_install.sh data # show the data it is sending # ./fleet_install.sh logs # ./fleet_install.sh test [--server-ip IP] [--keep-server] # @@ -22,15 +24,14 @@ set -uo pipefail SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)" AGENTS="$SCRIPT_DIR/agents" -TYPES="g1 g1t r1 go2" -declare -A KIND=( [g1]="MAP uploader" [g1t]="TELEMETRY (G1, unitree_hg)" [r1]="TELEMETRY (unitree_hg)" [go2]="TELEMETRY (unitree_go)" ) +TYPES="g1 r1 go2" +declare -A KIND=( [g1]="TELEMETRY + MAP (unitree_hg)" [r1]="TELEMETRY (unitree_hg)" [go2]="TELEMETRY (unitree_go)" ) die(){ echo "ERROR: $*" >&2; exit 1; } img_of(){ echo "sanad-api-$1"; } # image + container + unit share this base rdir_of(){ echo "sanad_api_$1"; } unit_of(){ echo "sanad-api-$1.service"; } -# G1 map + G1 telemetry are the SAME physical robot → share one sn. -sn_default(){ case "$1" in g1|g1t) echo "g1_${IP##*.}";; *) echo "${1}_${IP##*.}";; esac; } +sn_default(){ echo "${1}_${IP##*.}"; } SSH_OPTS=(-o BatchMode=yes -o ConnectTimeout=8 -o StrictHostKeyChecking=accept-new) rmt(){ ssh -n "${SSH_OPTS[@]}" "$USER_@$IP" "$@"; } # -n: never read our stdin @@ -39,15 +40,33 @@ detect_server_ip(){ [ -n "$SERVER_IP" ] && { echo "$SERVER_IP"; return; } ip -o route get "$IP" 2>/dev/null | grep -oP 'src \K\S+' | head -1; } # ---- per-type docker create args (volumes differ) ---- +# g1: find where THIS robot's Sanad dashboard keeps its maps/places on the host +# (native web_nav3 install, or Package_4's host-mounted nav data). Falls back to +# the agent's own dirs (agent then reports map: no_map until maps are exposed). +G1_MAPS_HOST=""; G1_DATA_HOST="" +probe_g1_dirs(){ + G1_MAPS_HOST="/home/$USER_/$(rdir_of g1)/maps" + for c in "/home/$USER_/marcus_nav2_test/maps" \ + "/home/$USER_/sanad_deploy/Sanad_Package_4/nav/data/ros"; do + if rmt "test -d $c" 2>/dev/null; then G1_MAPS_HOST="$c"; break; fi + done + G1_DATA_HOST="/home/$USER_/$(rdir_of g1)/web_data" + for c in "/home/$USER_/marcus_nav2_test/web/data" \ + "/home/$USER_/sanad_deploy/Sanad_Package_4/nav/data/web"; do + if rmt "test -d $c" 2>/dev/null; then G1_DATA_HOST="$c"; break; fi + done +} + run_args(){ - local base="--network host --env-file /home/$USER_/$(rdir_of "$1")/.env" + # all agents: host root read-only at /host → real disk-usage stats + local base="--network host --env-file /home/$USER_/$(rdir_of "$1")/.env -v /:/host:ro" if [ "$1" = g1 ]; then - echo "$base -v /home/$USER_/$(rdir_of "$1")/maps:/data/maps:ro \ - -v /home/$USER_/$(rdir_of "$1")/web_data:/data/web_data:ro \ - -v /home/$USER_/$(rdir_of "$1")/state:/data/state" + # g1 also mounts the Sanad dashboard map stores + its upload-state dir + echo "$base -v ${G1_MAPS_HOST:-/home/$USER_/$(rdir_of g1)/maps}:/data/maps:ro \ + -v ${G1_DATA_HOST:-/home/$USER_/$(rdir_of g1)/web_data}:/data/web_data:ro \ + -v /home/$USER_/$(rdir_of g1)/state:/data/state" else - # telemetry agents: host root read-only at /host → real disk-usage stats - echo "$base -v /:/host:ro" + echo "$base" fi } @@ -61,34 +80,20 @@ verify_tls(){ [ -n "$VERIFY_TLS_OPT" ] && echo "$VERIFY_TLS_OPT" || { [ -n "$SER push_env(){ local t="$1" sip="$2" rdir surl vtls; rdir="$(rdir_of "$t")" surl="$(server_url "$sip")"; vtls="$(verify_tls)" - if [ "$t" = g1 ]; then - rmt_in "cat > ~/$rdir/.env" <_ (e.g. r1_82, g1_58) - local btype=humanoid model="$t" - [ "$t" = g1t ] && model=g1 - [ "$t" = go2 ] && btype=dog - local rname="${NAME:-${model}_${IP##*.}}" - # optional Sanad data dir (its size is shown on the dashboard) — probe - # the known per-robot locations; path is as seen through /host (ro mount) - local dpath="" - for c in "/home/$USER_/SanadR1/data" "/home/$USER_/sanad_deploy/Sanad_Package_4/data"; do - if rmt "test -d $c" 2>/dev/null; then dpath="/host$c"; break; fi - done - rmt_in "cat > ~/$rdir/.env" <_ (e.g. r1_82, g1_58) + local iface=eth0 btype=humanoid model="$t" + [ "$t" = r1 ] && iface=eth10 + [ "$t" = go2 ] && btype=dog + local rname="${NAME:-${model}_${IP##*.}}" + # optional Sanad data dir (its size is shown on the dashboard) — probe + # the known per-robot locations; path is as seen through /host (ro mount) + local dpath="" + for c in "/home/$USER_/SanadR1/data" "/home/$USER_/sanad_deploy/Sanad_Package_4/data"; do + if rmt "test -d $c" 2>/dev/null; then dpath="/host$c"; break; fi + done + # common telemetry env (all agents) + rmt_in "cat > ~/$rdir/.env" <> ~/$rdir/.env" < (e.g. --sn E39N4000Q6D7E70F)" if [ -z "$SERVER_URL_OVERRIDE" ]; then sip="$(detect_server_ip)"; [ -n "$sip" ] || die "cannot detect server IP toward $IP (use --server-ip or --server-url)" fi @@ -121,6 +141,11 @@ do_install(){ -e "ssh ${SSH_OPTS[*]}" "$AGENTS/$t/" "$USER_@$IP:~/$rdir/" || die "rsync failed" rmt "mkdir -p ~/$rdir/maps/sanad ~/$rdir/web_data ~/$rdir/state ~/.config/systemd/user" push_env "$t" "$sip" + if [ "$t" = g1 ]; then + probe_g1_dirs + echo ">> Sanad dashboard maps dir: $G1_MAPS_HOST" + echo ">> places dir: $G1_DATA_HOST" + fi echo ">> docker build on robot (native arm64; R1/Go2 compile DDS on first build) ..." rmt "cd ~/$rdir && docker build -t $img:latest ." || die "docker build failed" echo ">> creating container (systemd will own its lifecycle) ..." @@ -192,10 +217,15 @@ do_test(){ [ "$rc" = 200 ] || die "robot cannot reach workstation server (http $rc) — firewall on $sip:$PORT?" echo " reachability OK (HTTP $rc)" if [ "$t" = g1 ]; then - rmt "mkdir -p ~/$(rdir_of g1)/maps/sanad ~/$(rdir_of g1)/web_data/sanad/places - head -c 4096 /dev/urandom > ~/$(rdir_of g1)/maps/sanad/floor-test.db - printf '%s' '{\"dock\":{\"x\":1.2,\"y\":3.4,\"qz\":0,\"qw\":1}}' > ~/$(rdir_of g1)/web_data/sanad/places/floor-test.json - rm -f ~/$(rdir_of g1)/state/uploaded.json" + probe_g1_dirs + if [ "$G1_MAPS_HOST" = "/home/$USER_/$(rdir_of g1)/maps" ]; then + # no real Sanad maps dir on this robot — seed a fixture to prove the path + rmt "mkdir -p ~/$(rdir_of g1)/maps/sanad ~/$(rdir_of g1)/web_data/sanad/places + head -c 4096 /dev/urandom > ~/$(rdir_of g1)/maps/sanad/floor-test.db + touch -d '10 minutes ago' ~/$(rdir_of g1)/maps/sanad/floor-test.db + printf '%s' '{\"dock\":{\"x\":1.2,\"y\":3.4,\"qz\":0,\"qw\":1}}' > ~/$(rdir_of g1)/web_data/sanad/places/floor-test.json + rm -f ~/$(rdir_of g1)/state/uploaded.json" + fi rmt "docker run --rm $(run_args g1) $img:latest --once --force" || true else echo ">> real DDS --once:"; rmt "docker run --rm $(run_args "$t") $img:latest --once" || true @@ -223,7 +253,7 @@ PY # --------------------------- arg parsing --------------------------- # CMD=""; ROBOT=""; IP=""; SERVER_IP=""; PORT=8799; TOKEN="test-token"; SN=""; NAME=""; USER_="unitree"; KEEP_SERVER=0 -SERVER_URL_OVERRIDE=""; VERIFY_TLS_OPT="" +SERVER_URL_OVERRIDE=""; VERIFY_TLS_OPT=""; POST_ENDPOINT="" POSA=() while [ $# -gt 0 ]; do case "$1" in --server-ip) SERVER_IP="$2"; shift 2;; @@ -233,6 +263,7 @@ while [ $# -gt 0 ]; do case "$1" in --token) TOKEN="$2"; shift 2;; --sn) SN="$2"; shift 2;; --name) NAME="$2"; shift 2;; # friendly display name (default _) + --post) POST_ENDPOINT="$2"; shift 2;; # ingest POST path (telemetry or map endpoint) --user) USER_="$2"; shift 2;; --keep-server) KEEP_SERVER=1; shift;; -h|--help) grep -E '^#( |$)' "$0" | sed 's/^# \{0,1\}//'; exit 0;; @@ -274,12 +305,24 @@ interactive(){ } prompt_install(){ - local def; def="$(sn_default "$ROBOT")" - read -rp "Robot name / SN [$def]: " s; SN="${s:-$def}" + # SN = the robot's REAL serial; it keys the robot on the fleet server → required. + while [ -z "$SN" ]; do + read -rp "Robot serial SN (required, e.g. E39N4000PB89GF88): " SN + done + local defname="${ROBOT}_${IP##*.}" + read -rp "Display name [$defname]: " nm; NAME="${nm:-$defname}" + # Server: full URL (https://… → TLS verified) or a bare IP (local test server). local autos; autos="$(detect_server_ip)" - read -rp "Fleet server IP [${autos:-required}]: " si; [ -n "$si" ] && SERVER_IP="$si" || SERVER_IP="${SERVER_IP:-$autos}" - read -rp "Server port [$PORT]: " p; [ -n "$p" ] && PORT="$p" + read -rp "SERVER_URL (https://…) or server IP [${autos:-required}]: " si + si="${si:-$autos}" + case "$si" in + http://*|https://*) SERVER_URL_OVERRIDE="$si";; + "") die "server URL or IP required";; + *) SERVER_IP="$si" + read -rp "Server port [$PORT]: " p; [ -n "$p" ] && PORT="$p";; + esac read -rp "Device token [$TOKEN]: " tk; [ -n "$tk" ] && TOKEN="$tk" + read -rp "POST endpoint [agent default]: " pe; [ -n "$pe" ] && POST_ENDPOINT="$pe" } # --------------------------- dispatch --------------------------- # @@ -287,7 +330,9 @@ prompt_install(){ [ -n "$ROBOT" ] && [ -n "$IP" ] || die "usage: $0 [opts] (or run with no args for interactive)" echo "$TYPES" | grep -qw "$ROBOT" || die "robot must be one of: $TYPES" [ -d "$AGENTS/$ROBOT" ] || die "agent dir missing: $AGENTS/$ROBOT" -[ -z "$SN" ] && SN="$(sn_default "$ROBOT")" +# SN: REQUIRED for install (the robot's real serial — checked in do_install); +# for test-only runs a placeholder default is fine. +if [ -z "$SN" ] && [ "$CMD" != install ]; then SN="$(sn_default "$ROBOT")"; fi case "$CMD" in install) do_install "$ROBOT";;