From 2757229abcea1857cdda51683a29283189aff70b Mon Sep 17 00:00:00 2001 From: Sidra Date: Tue, 4 Aug 2026 15:14:59 +0400 Subject: [PATCH] Update 2026-08-04 15:14 --- .gitignore | 5 + README.md | 230 +++++ agent/.dockerignore | 5 + agent/.env.example | 166 +++ agent/AGENT_README.md | 174 ++++ agent/Dockerfile | 43 + agent/entrypoint.sh | 19 + agent/requirements.txt | 10 + agent/sanad_api_x2.py | 2199 ++++++++++++++++++++++++++++++++++++++++ docs/X2_INTERFACE.md | 190 ++++ tools/probe_x2.sh | 159 +++ 11 files changed, 3200 insertions(+) create mode 100644 .gitignore create mode 100644 README.md create mode 100644 agent/.dockerignore create mode 100644 agent/.env.example create mode 100644 agent/AGENT_README.md create mode 100644 agent/Dockerfile create mode 100644 agent/entrypoint.sh create mode 100644 agent/requirements.txt create mode 100644 agent/sanad_api_x2.py create mode 100644 docs/X2_INTERFACE.md create mode 100644 tools/probe_x2.sh diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..838f105 --- /dev/null +++ b/.gitignore @@ -0,0 +1,5 @@ +.env +__pycache__/ +*.pyc +state/ +maps/ diff --git a/README.md b/README.md new file mode 100644 index 0000000..3baa25e --- /dev/null +++ b/README.md @@ -0,0 +1,230 @@ +# AGI Fleet — AGIBOT X2 agent + +On-robot agent that reports the **AGIBOT X2**'s live state to the **YS Lootah +fleet server**. Self-contained: it shares no code with the Unitree (G1 / R1 / +Go2) agents and is deployed and managed independently. + +> **Status: LIVE on production.** Running on `10.255.254.84`, posting to +> `https://eco.yslootahrobotics.com` every 2 s. + +``` +telemetry ok: battery=68 charging=True status=charging pos={'x': 1.222, 'y': 0.437} -> HTTP 200 +``` + +--- + +## 1. Layout + +``` +agi_fleet/ +├── README.md ← this file +├── agent/ +│ ├── sanad_api_x2.py ← THE AGENT (deployed as-is — do not edit casually) +│ ├── .env.example ← every setting, documented +│ ├── requirements.txt ← requests (+ websocket-client, optional) +│ ├── AGENT_README.md ← agent internals: backends, field mapping +│ ├── Dockerfile ← optional container path (NOT used — see §7) +│ ├── entrypoint.sh ← container entrypoint (unused) +│ └── .dockerignore +├── tools/ +│ └── probe_x2.sh ← read-only robot discovery +└── docs/ + └── X2_INTERFACE.md ← what the X2 actually exposes (topics, QoS, msgs) +``` + +The agent filename stays `sanad_api_x2.py` because that is exactly what is +deployed and running on the robot (`~/sanad_api_x2/`, unit `sanad-api-x2`). +Renaming it would mean redeploying a working production feed for cosmetic +reasons — not worth it. + +--- + +## 2. The robot + +| | | +|---|---| +| Host | `10.255.254.84` (NIC `wifi0`) | +| SSH | `agi@10.255.254.84`, key `~/.ssh/agibot_x2_ed25519` (alias `agix2`) | +| Serial (`SN`) | `X230028C5Z0058` | +| Display name | `x2_84` | +| Hardware | NVIDIA Jetson Orin NX, arm64, L4T R36.4.3 | +| OS | Ubuntu 22.04.5 LTS, kernel 5.15.148-tegra | +| ROS | Humble (`/opt/ros/humble`), `ROS_DOMAIN_ID=0` | +| Install dir | `~/sanad_api_x2/` | +| Service | `systemctl --user sanad-api-x2` | + +--- + +## 3. How it gets its data + +Two independent sources, running at the same time: + +| what | source | +|---|---| +| battery, charging, voltage, current, temp, cycles | **HTTP** — the robot's *AGIBOT X2 Control Dash* at `http://127.0.0.1:8770/api/state` | +| position `{x, y}` | **ROS 2** — `/aima/mc/leg_odometry` (`nav_msgs/msg/Odometry`) | +| locomotion → `status: moving` | `joints.leg[*].velocity` from the same dash payload | +| OS, kernel, arch, board, L4T, storage, MAC | read directly from the host | + +That split matters: the Control Dash carries no odometry, and ROS carries no +tidy battery percentage — so `X2_SOURCE=http` and `X2_POSITION_SOURCE=ros2` run +side by side. Neither depends on the other. + +### The field mapping actually in use + +```ini +X2_SOURCE=http +X2_STATE_URL=http://127.0.0.1:8770/api/state +X2_FIELD_SOC=battery_pct X2_FIELD_VOLTAGE=battery_voltage +X2_FIELD_CURRENT=battery_current X2_FIELD_TEMP=battery_temp +X2_FIELD_CYCLES=battery_cycles X2_FIELD_VEL=joints.leg[*].velocity +X2_SOC_SCALE=percent X2_CURRENT_SIGN=1 + +X2_POSITION_SOURCE=ros2 +X2_TOPIC_ODOM=/aima/mc/leg_odometry +X2_ROS_QOS=best_effort # ← must stay best_effort, see docs/X2_INTERFACE.md + +MAC_INTERFACE=wifi0 +REMOTE_PORTS=8770,8001,8000,8080 +``` + +Nothing above is hard-coded in the agent — it is all `.env`. Retargeting is an +`.env` edit + restart, never a code change. + +--- + +## 4. What it sends + +One JSON object every 2 s to `POST /api/v1/fleet/ingest/telemetry` with +`Authorization: Bearer `, plus `/{sn}/alert`, `/{sn}/logs` and +`/{sn}/remote`. Verified field-by-field against the robot's own readings: + +| robot ground truth | agent sends | +|---|---| +| dash `battery_pct: 68.0` | `battery: 68` | +| dash `battery_voltage: 51.996` | `voltage_v: 52.0` | +| dash `battery_current: 3.544` | `current_a: 3.54` | +| dash `battery_temp: 41.6` · `cycles: 14` | `temp_c: 42` · `cycles: 14` | +| dash `charging: true` | `charging: true`, `status: "charging"` | +| ROS odom `x: 1.2202 y: 0.4363` | `position: {x: 1.222, y: 0.437}` | + +Also registers the Control Dash (`http://10.255.254.84:8770`) and +`ssh agi@10.255.254.84` for the fleet UI. + +### Fields that are always `null` — and why + +- **`motor_temp`** — the X2 publishes **no per-motor temperatures at all**. + `PmuState.msg` is the only `aimdk_msgs` definition containing any temperature + field and it is the *power unit*, not the motors. Reporting `pmu_temp` there + would be fabricating data. +- **`map`** — `no_map`; there are no saved maps on this robot. +- **`project_logs` / `control`** — these are Sanad-app integrations; the X2 does + not run that app, so they degrade to `null` automatically. + +`null` means "not available", never "zero". + +--- + +## 5. Operating it + +```bash +# live log +ssh agix2 'journalctl --user -u sanad-api-x2 -f' + +# what it is sending right now +ssh agix2 "journalctl --user -u sanad-api-x2 -n 20 --no-pager | grep -oE 'telemetry ok:.*'" + +# service control +ssh agix2 'systemctl --user restart sanad-api-x2' +ssh agix2 'systemctl --user status sanad-api-x2' + +# change a setting (then restart) +ssh agix2 'nano ~/sanad_api_x2/.env && systemctl --user restart sanad-api-x2' +``` + +### Updating the agent code + +```bash +scp -i ~/.ssh/agibot_x2_ed25519 agent/sanad_api_x2.py \ + agi@10.255.254.84:~/sanad_api_x2/sanad_api_x2.py +ssh agix2 'systemctl --user restart sanad-api-x2' +``` + +`.env` is never overwritten by this, so the device token stays put. + +### Rotating the device token + +The fleet server issues a **Universal Connector** token — regenerating it on the +server invalidates the previous one for *every* robot that uses it. + +```bash +ssh agix2 "sed -i 's#^DEVICE_TOKEN=.*#DEVICE_TOKEN=#' ~/sanad_api_x2/.env \ + && chmod 600 ~/sanad_api_x2/.env && systemctl --user restart sanad-api-x2" +``` + +### Re-discovering the robot's interface + +```bash +FLEET_SSH_KEY=~/.ssh/agibot_x2_ed25519 bash tools/probe_x2.sh 10.255.254.84 agi +``` + +Read-only — no install, no writes, safe on a live robot. Every line it prints is +labelled with the `.env` variable it feeds. + +--- + +## 6. ⚠ Open item — reboot survival + +`linger` is **off**, so the agent will **not** start again after a power cycle. +It does survive crashes (systemd `Restart=always`, verified by `kill -9`). + +polkit denies `loginctl enable-linger` for a non-root user on this image, so it +needs one privileged command **on the robot**: + +```bash +ssh agix2 'sudo loginctl enable-linger agi' # prompts for the password +ssh agix2 'loginctl show-user agi | grep Linger' # expect Linger=yes +``` + +--- + +## 7. Why it runs natively, not in Docker + +Docker is installed on the X2 but the daemon is **`inactive` and `masked`**, and +`agi` has no passwordless sudo — so the container path cannot start. The agent's +only hard dependency is `requests`, so it runs directly under a user-level +systemd unit instead. Same auto-start model, same `.env`, no root. + +The unit sources the ROS overlay first so `rclpy` is importable for position: + +```ini +ExecStart=/bin/bash -c 'set +u; . /opt/ros/humble/setup.bash >/dev/null 2>&1 || true; \ + exec /usr/bin/python3 -u /home/agi/sanad_api_x2/sanad_api_x2.py' +``` + +`set +u` is required — ROS's `setup.bash` reads unbound variables and would abort +the unit under `set -u`. + +`agent/Dockerfile` and `agent/entrypoint.sh` are kept for the day dockerd is +unmasked; nothing currently uses them. + +--- + +## 8. Security + +- **Outbound only.** No inbound port is opened on the robot; every call is an + HTTPS POST carrying `Authorization: Bearer `, keyed by `sn`. +- **Read-only toward the robot.** It reads an HTTP status page and subscribes to + one ROS topic. It never commands motion. +- **Token handling.** Lives only in `~/sanad_api_x2/.env` on the robot, mode + `600`, and is `.gitignore`d here. Only `.env.example` (placeholders) is in this + repo. +- **TLS verified** (`VERIFY_TLS=1`) against the production server. + +--- + +## 9. Relationship to the Unitree fleet + +None. This project shares no files with `fleet/` (the G1 / R1 / Go2 agents). +`sanad_api_x2.py` is standalone — it is not generated by `fleet/tools/gen_agents.py` +and contains zero Unitree/DDS code. The two can be changed independently. diff --git a/agent/.dockerignore b/agent/.dockerignore new file mode 100644 index 0000000..ed287fb --- /dev/null +++ b/agent/.dockerignore @@ -0,0 +1,5 @@ +.env +data/ +__pycache__/ +*.pyc +README.md diff --git a/agent/.env.example b/agent/.env.example new file mode 100644 index 0000000..aa1a71f --- /dev/null +++ b/agent/.env.example @@ -0,0 +1,166 @@ +# sanad_api_x2 — AGIBOT X2. Copy to .env and fill in. +# ONE agent = telemetry + map + logs + alerts + remote (see fleet/README.md). +# +# READ THIS FIRST — the X2 exposes no fixed topic contract, so EVERY topic name +# and field path used to read robot state is an env var in this file. Nothing is +# hard-coded in the agent. Discover the real ones with: +# +# ./fleet_install.sh probe x2 (or: tools/probe_x2.sh @) +# +# then set them below and restart — no code change, no image rebuild. + +# ── fleet server (REQUIRED — YS Lootah gives you these two) ────────────────── +SERVER_URL=https://eco.yslootahrobotics.com +DEVICE_TOKEN=REPLACE_WITH_DEVICE_TOKEN + +# ── identity ───────────────────────────────────────────────────────────────── +# The robot's REAL serial — keys the robot on the server ({sn} routes). +SN=REPLACE_WITH_X2_SERIAL +# Friendly display name shown on the dashboard (default _). +ROBOT_NAME=x2_10 +ROBOT_BRAND=agibot +ROBOT_TYPE=humanoid +ROBOT_MODEL=x2 +# Maps subdir (//…) + X-Robot-Name header. +ROBOT=sanad +# Optional: app 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= + +# ═════════════════════════════════════════════════════════════════════════════ +# X2 STATE SOURCE — the one robot-specific section +# ═════════════════════════════════════════════════════════════════════════════ +# auto = http if X2_STATE_URL is set, else ros2 if rclpy imports, else aimrt if +# aimrt_py imports, else none (heartbeat: battery null, status offline). +# http = poll X2_STATE_URL for one JSON object; map fields with X2_FIELD_* . +# ros2 = subscribe X2_TOPIC_* (needs a ROS base image — see Dockerfile). +# aimrt = AgiBot's runtime. NOT natively bound (channel/message defs are not +# public). Enable AimRT's ROS 2 plugin and use X2_SOURCE=ros2 instead. +# none = never read; always heartbeat. Useful to bring a robot online on the +# dashboard (identity, storage, maps, logs) before the state map is known. +X2_SOURCE=auto + +# ── backend: http ──────────────────────────────────────────────────────────── +# A vendor endpoint returning one JSON object with the robot's state. +X2_STATE_URL= +X2_HTTP_INTERVAL=1 + +# ── backend: ros2 ──────────────────────────────────────────────────────────── +# Topic names from `ros2 topic list` on the robot. Blank = don't subscribe. +# The TYPE must match `ros2 topic info `; custom vendor messages work as +# long as their python package is importable inside the container. +X2_TOPIC_BATTERY=/battery_state +X2_TYPE_BATTERY=sensor_msgs/msg/BatteryState +X2_TOPIC_JOINTS=/joint_states +X2_TYPE_JOINTS=sensor_msgs/msg/JointState +X2_TOPIC_ODOM=/odom +X2_TYPE_ODOM=nav_msgs/msg/Odometry +# Must match the robot's domain or ROS 2 discovery silently sees nothing. +ROS_DOMAIN_ID=0 + +# ── field mapping (applies to EVERY backend) ───────────────────────────────── +# Dotted paths into the message/JSON. They work over both ROS message objects +# and plain dicts, support list indices ("bms.cell_temp[0]"), and yield null for +# any missing link — a wrong path degrades a field, it never crashes the agent. +# Defaults below are the STANDARD ROS messages: correct for a stock ros2 setup, +# almost certainly wrong for a vendor http payload. Empty = field unavailable. +X2_FIELD_SOC=percentage +X2_FIELD_VOLTAGE=voltage +X2_FIELD_CURRENT=current +X2_FIELD_TEMP=temperature +X2_FIELD_SOH= +X2_FIELD_CYCLES= +# Standard sensor_msgs/JointState has NO temperature array — leave empty unless +# the X2 publishes a custom message that carries one (motor_temp reports null). +X2_FIELD_TEMPS= +# Joint velocities → the "moving" status (max |velocity| > 0.15). +X2_FIELD_VEL=velocity +X2_FIELD_X=pose.pose.position.x +X2_FIELD_Y=pose.pose.position.y +# http backend only: an FSM id and a {name: version} firmware dict, if exposed. +X2_FIELD_FSM= +X2_FIELD_FW= + +# ── unit conventions (verify these on the real robot!) ─────────────────────── +# auto = a value in (0, 1] is treated as a 0..1 fraction and scaled to %. +# fraction = always scale x100. percent = never scale. +X2_SOC_SCALE=auto +# +1 = positive current means CHARGING (the ROS BatteryState convention). +# Set -1 if the X2 reports the opposite sign, or "charging" will be inverted. +X2_CURRENT_SIGN=1 +# The telemetry schema is VOLTS and AMPS. BMS firmware often publishes mV/mA +# instead — set 0.001 for those, so battery_detail isn't off by 1000x. +X2_VOLTAGE_SCALE=1 +X2_CURRENT_SCALE=1 + +# ── position / status ──────────────────────────────────────────────────────── +# none | ros2 | rosbridge. ros2 is INDEPENDENT of X2_SOURCE: the Control Dash +# carries no odometry, but ROS publishes it — so http battery + ros2 position +# run side by side. Needs rclpy, i.e. the unit must source the ROS overlay. +# On the X2 as deployed: /aima/mc/leg_odometry (nav_msgs/msg/Odometry). +X2_POSITION_SOURCE=ros2 +ROSBRIDGE_URL=ws://127.0.0.1:9090 + +# Subscription reliability. MUST stay best_effort unless you know otherwise: +# the X2 publishes /aima/mc/leg_odometry as BEST_EFFORT, and a RELIABLE +# subscriber receives NOTHING from a BEST_EFFORT publisher — the subscription +# is created, no error is raised, and position silently stays null forever. +# A BEST_EFFORT subscriber reads from either kind of publisher. +X2_ROS_QOS=best_effort +# There is no known X2 loco FSM RPC — leave 0. +X2_READ_FSM=0 + +# ── identity / networking ──────────────────────────────────────────────────── +# Which NIC's MAC is reported as the robot identity. +MAC_INTERFACE=eth0 + +# ── fault thresholds ───────────────────────────────────────────────────────── +LOW_SOC=50 +# Only meaningful when X2_FIELD_TEMPS is mapped; otherwise motor_temp is null. +MOTOR_TEMP_MAX=85 + +# ── cadence / transport ────────────────────────────────────────────────────── +POLL_INTERVAL=2 +VERIFY_TLS=1 +HTTP_TIMEOUT=30 + +# ── map sync (uploaded ONCE per content; status shown in telemetry "map") ──── +# pgm+yaml sets are rendered to PNG with pure stdlib; RTAB-Map .db is sent +# as-is (skipped above the server's upload cap). +MAPS_DIR=/data/maps +EXTRA_MAP_DIRS=/data/nav2_maps +DATA_DIR=/data/web_data +STATE_DIR=/data/state +MAP_SELECT=all +MAP_UPLOAD_MODE=multipart +MAP_POLL_INTERVAL=30 +MAP_MAX_UPLOAD_MB=7 + +# ── logs + alerts ──────────────────────────────────────────────────────────── +LOGS_INTERVAL=60 +# auto = find a RUNNING sanad* container via the /host mount and tail its +# json-log. If the X2 doesn't run the Sanad app, telemetry.project_logs is null +# and only the agent's own lines ship — no configuration needed. +PROJECT_LOG_CONTAINER=auto +PROJECT_LOG_PATH= +PROJECT_LOG_LABEL= +ALERT_SCAN_INTERVAL=10 +ALERT_LOG_COOLDOWN=300 + +# ── remote (register a dashboard URL + ssh for the fleet UI) ───────────────── +# If the X2 exposes no dashboard, remote.web is null and only ssh registers. +REMOTE_ENABLE=1 +REMOTE_PORTS=8001,8014,8011,8012,8013,8000,8080 +REMOTE_KIND=web +REMOTE_INTERVAL=60 +SSH_REGISTER=1 +# The X2's LOGIN user. The installer writes the user it deployed with; set it +# here for manual runs. +# Left empty, ssh registration is skipped rather than registering a wrong user. +SSH_USER= + +# ── control panel (READ-ONLY; remote mode-SWITCH is off by design) ─────────── +# Blank URL = use the auto-discovered dashboard port. No known X2 FSM id map, so +# control.mode reports "unknown" until the ids are filled into the agent. +CONTROL_STATUS_URL= +CONTROL_ENABLE=0 diff --git a/agent/AGENT_README.md b/agent/AGENT_README.md new file mode 100644 index 0000000..ea727fb --- /dev/null +++ b/agent/AGENT_README.md @@ -0,0 +1,174 @@ +# sanad_api_x2 — AGIBOT X2 fleet agent + +Reports the **AGIBOT X2**'s live state to the YS Lootah fleet server over the +fleet's five ingest endpoints, with the standard 27-field payload and per-robot +Bearer-token auth. Deploy it with the fleet installer (`../../fleet_install.sh`). + +Generated from the canonical `agents/g1/` by `tools/gen_agents.py` — **do not +hand-edit this folder's `sanad_api_x2.py`**. Everything except the state source +is byte-identical logic to the G1/R1/Go2 agents. + +--- + +## The one thing that is different + +**The X2 exposes no fixed topic contract**, so this agent hard-codes nothing: +it picks a state-source backend at runtime and reads every topic name and field +path from `.env`. + +Retargeting it at the real interface is an **`.env` edit + restart** — no code +change, no image rebuild. + +| `X2_SOURCE` | what it does | needs | +|---|---|---| +| `auto` *(default)* | `http` if `X2_STATE_URL` is set → else `ros2` if `rclpy` imports → else `aimrt` if `aimrt_py` imports → else `none` | — | +| `http` | polls `X2_STATE_URL` for one JSON object | nothing (slim image) | +| `ros2` | subscribes `X2_TOPIC_*` | a ROS base image (see below) | +| `aimrt` | **not natively bound** — see below | — | +| `none` | never reads; always heartbeat | — | + +Whatever the backend, `AgiBotSource` fills one six-key snapshot — `bms`, +`state_age`, `temps`, `max_vel`, `xy`, `fw` — which is the entire seam between +this robot and the shared telemetry / fault / status / alert pipeline. + +### On the `aimrt` backend + +AgiBot's open-source X1 stack is built on **AimRT**, and it is the likely +middleware here. It is deliberately **not** implemented natively: the X2's AimRT +channel names and message definitions aren't public, and guessing them yields an +agent that imports cleanly, builds, and posts perfect-looking heartbeats with +`battery: null` forever — a failure you'd only find on the robot. + +The supported path today is **AimRT's ROS 2 plugin**: enable it and run with +`X2_SOURCE=ros2` against the bridged topics. If you get the native channel +definitions from AgiBot, implement `_start_aimrt()` in `tools/gen_agents.py` +(not here — this file is generated). The only contract to satisfy is calling +`_ingest_battery` / `_ingest_joints` / `_ingest_odom` with the incoming messages. + +--- + +## Quick start + +**1. Discover the robot's interface** (read-only — safe on a live robot): + +```bash +./fleet_install.sh probe x2 --user # or: tools/probe_x2.sh +``` + +It prints the platform, serial-number candidates, NICs, ROS topics **with their +message types**, listening ports, running containers and map directories — each +labelled with the `.env` variable it feeds. + +**2. Install**, using the real serial from the probe: + +```bash +./fleet_install.sh install x2 --sn \ + --token --server-url https://eco.yslootahrobotics.com +``` + +For the `ros2` backend, build against a base matching the **robot's** ROS distro +(a mismatch means ROS 2 discovery silently sees no topics): + +```bash +X2_BASE_IMAGE=ros:humble-ros-base ./fleet_install.sh install x2 --sn ... +``` + +**3. Map the fields** you found, then restart: + +```bash +ssh @ 'nano ~/sanad_api_x2/.env' +ssh @ 'systemctl --user restart sanad-api-x2' +./fleet_install.sh data x2 +``` + +**Updating later?** Always pass `--keep-token`, or `--token` defaults to +`test-token` and the feed dies with `401`. + +--- + +## Field mapping + +Paths are dotted and work over **both** ROS message objects and plain JSON +dicts, with list indices: `pose.pose.position.x`, `battery.cell_temp[0]`, +`percentage`. A missing link yields `null` — a wrong path degrades one field, it +never crashes the agent. + +| var | default (standard ROS msg) | feeds | +|---|---|---| +| `X2_FIELD_SOC` | `percentage` | `battery` | +| `X2_FIELD_VOLTAGE` / `_CURRENT` | `voltage` / `current` | `battery_detail`, `charging` | +| `X2_FIELD_TEMP` / `_SOH` / `_CYCLES` | `temperature` / — / — | `battery_detail` | +| `X2_FIELD_TEMPS` | *(empty)* | `motor_temp`, `MOTOR_OVERTEMP` | +| `X2_FIELD_VEL` | `velocity` | `status: moving` | +| `X2_FIELD_X` / `_Y` | `pose.pose.position.{x,y}` | `position` | +| `X2_FIELD_FSM` / `_FW` | *(empty)* | `control.fsm_id`, `firmware` *(http only)* | + +> **`motor_temp` is `null` on the X2, permanently.** `PmuState.msg` is the only +> `aimdk_msgs` definition with any temperature field, and it is the *power unit* +> temperature — not per-motor. `aimdk_msgs/msg/JointState` is +> `name/position/velocity/effort/error_code`. Reporting `pmu_temp` as +> `motor_temp` would be fabricating data, so it stays `null` = "not receiving". + +## Position — a second, independent source + +The Control Dash exposes no odometry, but ROS does. `X2_POSITION_SOURCE=ros2` +runs **alongside** `X2_SOURCE=http`: battery over HTTP, position over ROS 2. + +``` +X2_POSITION_SOURCE=ros2 +X2_TOPIC_ODOM=/aima/mc/leg_odometry # nav_msgs/msg/Odometry +X2_ROS_QOS=best_effort +``` + +⚠ **`X2_ROS_QOS` must stay `best_effort`.** The X2 publishes that topic as +`BEST_EFFORT`, and a `RELIABLE` subscriber receives **nothing** from a +`BEST_EFFORT` publisher — rclpy creates the subscription, raises no error, and +position silently stays `null`. A `BEST_EFFORT` subscriber reads from either +kind of publisher, which is why it's the default. + +This requires `rclpy`, so the systemd unit sources the ROS overlay before +starting the agent: + +```ini +ExecStart=/bin/bash -c 'set +u; . /opt/ros/humble/setup.bash >/dev/null 2>&1 || true; \ + exec /usr/bin/python3 -u /home/agi/sanad_api_x2/sanad_api_x2.py' +``` + +### Verify these two on real hardware + +Both are unit conventions that look plausible when wrong: + +| var | symptom if wrong | fix | +|---|---|---| +| `X2_SOC_SCALE` | battery reads `0` or `100` when it should be `47` | `fraction` (0–1 source) or `percent` (0–100 source) | +| `X2_CURRENT_SIGN` | `charging` is inverted | `-1` | +| `X2_VOLTAGE_SCALE` / `X2_CURRENT_SCALE` | `voltage_v: 48200` instead of `48.2` | `0.001` for mV/mA sources | + +--- + +## What works without any mapping + +These need no X2-specific configuration and are live from the first install — +so the robot is useful on the dashboard **before** the state map is known: + +- **identity** — `sn`, `name`, `mac`, `brand: agibot`, `type`, `model` +- **software / firmware** — OS, kernel, arch, python, board, L4T (via `/host`) +- **storage** — real host disk usage +- **maps** — `pgm`+`yaml` sets rendered to PNG, RTAB-Map `.db` sent as-is +- **logs** — the agent's own lines, shipped every 60 s with requeue on failure +- **remote** — `ssh @` registration +- **alerts** — log-pattern scanning + +Fields that depend on the **Sanad app** degrade to `null` automatically if the +X2 doesn't run it: `project_logs`, `remote.web`, `control`. No configuration +needed — the schema stays identical, the dashboard just shows fewer cards. + +--- + +## Status + +⚠ **Unverified on hardware.** The full pipeline is verified end-to-end against +`fleet_test_server.py` — all five endpoints, correct Bearer auth, 27-field +payload, field mapping, unit scaling, fault derivation and map upload. What is +*not* verified is which backend and which field paths the real X2 needs; that is +what `probe_x2.sh` and the `.env` mapping exist to resolve. diff --git a/agent/Dockerfile b/agent/Dockerfile new file mode 100644 index 0000000..552273b --- /dev/null +++ b/agent/Dockerfile @@ -0,0 +1,43 @@ +# sanad_api_x2 — AGIBOT X2 fleet agent: TELEMETRY + MAP sync. +# +# BUILD ON THE X2 COMPUTE (native arm64). There is no middleware to compile into +# the image: the state source is chosen at RUNTIME +# (X2_SOURCE=http|ros2|aimrt|none), so the image only needs whatever the chosen +# backend requires. That is what BASE_IMAGE selects: +# +# http / none backend (default) — tiny, no ROS at all: +# docker build -t sanad-api-x2:latest . +# +# ros2 backend — rclpy comes from the ROS base; match the ROBOT's distro: +# docker build --build-arg BASE_IMAGE=ros:humble-ros-base -t sanad-api-x2:latest . +# (or ros:foxy-ros-base, ros:jazzy-ros-base, … — must match the robot, or +# ROS 2 discovery silently fails to see the robot's topics) +# +# Run with --network host so the robot's middleware traffic is visible (ROS 2 +# discovery does not cross a NAT bridge) and the agent can reach vendor APIs on +# 127.0.0.1 (the installer does this). +ARG BASE_IMAGE=python:3.10-slim-bookworm +FROM ${BASE_IMAGE} + +# iproute2 for iface checks; ca-certificates for outbound HTTPS to the fleet +# server. Kept minimal — the ROS bases already carry python3 + rclpy. +RUN apt-get update && apt-get install -y --no-install-recommends \ + iproute2 ca-certificates python3 python3-pip \ + && rm -rf /var/lib/apt/lists/* + +WORKDIR /app +COPY requirements.txt . +# ROS bases ship a Debian-managed python (PEP 668), so retry with +# --break-system-packages rather than failing the build. +RUN python3 -m pip install --no-cache-dir -r requirements.txt \ + || python3 -m pip install --no-cache-dir --break-system-packages -r requirements.txt + +COPY sanad_api_x2.py entrypoint.sh ./ +RUN chmod +x entrypoint.sh + +ENV PYTHONUNBUFFERED=1 \ + X2_SOURCE=auto \ + MAC_INTERFACE=eth0 +# entrypoint.sh sources a ROS overlay when the base has one (no-op otherwise), +# then execs the agent — so ONE image definition serves every backend. +ENTRYPOINT ["/app/entrypoint.sh"] diff --git a/agent/entrypoint.sh b/agent/entrypoint.sh new file mode 100644 index 0000000..8896275 --- /dev/null +++ b/agent/entrypoint.sh @@ -0,0 +1,19 @@ +#!/bin/sh +# sanad_api_x2 entrypoint. +# +# The X2 agent runs on either a slim python base (http backend) or a ROS 2 base +# (ros2 backend). rclpy is only importable after the ROS overlay is sourced, and +# sourcing it must NOT be a hard requirement — so: source it if it exists, carry +# on if it doesn't. One image definition, every backend. +set -e + +if [ -z "${ROS_DISTRO:-}" ] && [ -d /opt/ros ]; then + ROS_DISTRO="$(ls /opt/ros 2>/dev/null | head -1)" +fi +if [ -n "${ROS_DISTRO:-}" ] && [ -f "/opt/ros/${ROS_DISTRO}/setup.sh" ]; then + # shellcheck disable=SC1090 + . "/opt/ros/${ROS_DISTRO}/setup.sh" + echo "entrypoint: sourced ROS ${ROS_DISTRO} (rclpy available)" >&2 +fi + +exec python3 -u /app/sanad_api_x2.py "$@" diff --git a/agent/requirements.txt b/agent/requirements.txt new file mode 100644 index 0000000..703401e --- /dev/null +++ b/agent/requirements.txt @@ -0,0 +1,10 @@ +# sanad_api_x2 — AGIBOT X2 fleet agent. +requests>=2.31,<3 +# only needed if X2_POSITION_SOURCE=rosbridge (reads /odom over a websocket): +websocket-client>=1.6,<2 +# NOT installed here: +# rclpy — comes from the ROS base image (see Dockerfile BASE_IMAGE); it is +# not on PyPI and must match the robot's ROS distro. +# aimrt_py — AgiBot's runtime, installed with their stack if you use it. +# The agent degrades to heartbeat mode when no state source is reachable, so a +# missing backend never stops telemetry. diff --git a/agent/sanad_api_x2.py b/agent/sanad_api_x2.py new file mode 100644 index 0000000..54aae56 --- /dev/null +++ b/agent/sanad_api_x2.py @@ -0,0 +1,2199 @@ +#!/usr/bin/env python3 +"""sanad_api_x2 — AGIBOT X2 fleet agent: TELEMETRY + MAP sync in ONE service. + +The single X2 agent (one container, one systemd service) that: + + 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 } + + 2. MAP — a background loop (every MAP_POLL_INTERVAL, default 30 s) checks the + robot's saved nav maps (pgm+yaml sets and RTAB-Map .db) 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) + + 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. + +DATA SOURCES (AGIBOT X2 — pluggable, configured in .env) +-------------------------------------------------------- + backend : X2_SOURCE = auto | http | ros2 | aimrt | none + battery / charging : X2_FIELD_SOC/VOLTAGE/CURRENT/TEMP/SOH/CYCLES + faults / motor temp: X2_FIELD_TEMPS (+ staleness of the last successful read) + position {x,y} : X2_FIELD_X / X2_FIELD_Y + status : derived (charging/moving/idle/offline) + storage : host disk via the read-only /:/host mount (+ data dir size) + maps : MAPS_DIR/**.pgm+yaml and *.db, places under + DATA_DIR//places/.json — unchanged from g1 + +Nothing above is hard-coded: run tools/probe_x2.sh on the robot to discover the +real topics/fields, then set them in .env. No code change, no image rebuild. + +Read-only toward the robot: never commands motion. Degrades to heartbeats when +no state source is reachable; --simulate fakes only the state side (the map scan +stays real). + +CONFIG — environment (see .env.example). Key vars: + SERVER_URL, DEVICE_TOKEN, SN (required at install), ROBOT_NAME, + X2_SOURCE + X2_TOPIC_*/X2_FIELD_* (state source), POLL_INTERVAL (2), + MAC_INTERFACE (which NIC's MAC to report), 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: --simulate | --once | --dry-run | --force (map re-upload) | --list (maps) | -v +""" +from __future__ import annotations + +import argparse +import base64 +import datetime as _dt +import hashlib +import json +import logging +import math +import os +import platform +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 + +import requests + +log = logging.getLogger("sanad_api_x2") + + +# --------------------------------------------------------------------------- # +# env helpers +# --------------------------------------------------------------------------- # +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") + + +def _now_str() -> str: + """Full local date+time with UTC offset (e.g. 2026-07-13 14:20:33+04:00). + TZ_OFFSET_HOURS (default +4, Dubai) keeps container clocks honest without tzdata.""" + off = float(_env("TZ_OFFSET_HOURS", "4")) + tz = _dt.timezone(_dt.timedelta(hours=off)) + return _dt.datetime.now(tz).isoformat(sep=" ", timespec="seconds") + + +# --------------------------------------------------------------------------- # +# config +# --------------------------------------------------------------------------- # +@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 + # telemetry / state source + net_interface: str + domain_id: int + mac_interface: str + read_fsm: bool + position_source: str + rosbridge_url: str + low_soc: int + motor_temp_max: float + alert_log_patterns: str + alert_log_cooldown: float + alert_scan_interval: float + alert_backfill_bytes: int + poll_interval: float + telemetry_endpoint: str + # map sync + robot: str + maps_dir: Path + extra_map_dirs: List[Path] # extra roots to scan for pgm+yaml sets (e.g. Nav2/Pudu maps) + web_data_dir: Optional[Path] + legacy_places: Optional[Path] + web_nav3_url: str + map_select: str + map_upload_mode: str + map_endpoint_tmpl: str + map_poll_interval: float + map_max_upload_mb: float + state_dir: Path + # logs + alerts + alert_endpoint: str + logs_endpoint: str + logs_interval: float + # remote dashboard (register the Sanad UI URL for the fleet to embed) + remote_enable: bool + remote_endpoint: str + remote_kind: str + remote_host: str + remote_ports: str + remote_url: str + remote_interval: float + ssh_enable: bool + ssh_user: str + ssh_port: int + control_url: str + control_enable: bool + # project logs (e.g. the robot's Sanad app) shipped alongside agent logs + project_log_container: str + project_log_path: str + project_log_label: str + project_log_backfill: int + project_log_exclude: str + ros_distro: str + # transport + 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("X2_INTERFACE", "eth0") + data_dir = _env("DATA_DIR") + legacy = _env("LEGACY_PLACES") + return cls( + server_url=server, + device_token=token, + sn=_env("SN", "x2_0000"), + name=_env("ROBOT_NAME", "") or _env("SN", "x2_0000"), + brand=_env("ROBOT_BRAND", "agibot"), + robot_type=_env("ROBOT_TYPE", "humanoid"), + model=_env("ROBOT_MODEL", "x2"), + storage_path=_env("STORAGE_PATH", ""), + data_path=_env("STORAGE_DATA_PATH", ""), + net_interface=iface, + domain_id=int(_env("ROS_DOMAIN_ID", "0")), + mac_interface=_env("MAC_INTERFACE", iface), + read_fsm=_env_bool("X2_READ_FSM", False), + position_source=_env("X2_POSITION_SOURCE", "none").lower(), + rosbridge_url=_env("ROSBRIDGE_URL", "ws://127.0.0.1:9090"), + low_soc=int(_env("LOW_SOC", "50")), + motor_temp_max=float(_env("MOTOR_TEMP_MAX", "85")), + # log-driven alerts: "CODE=regex" entries separated by ";;" (regex may + # contain '|'). Scanned against the robot's project logs (sanadr1). + # NOTE: matching is CASE-SENSITIVE (log levels are uppercase); use an + # inline (?i) prefix for case-insensitive text (Gemini messages). + alert_log_patterns=_env("ALERT_LOG_PATTERNS", + "GEMINI_BILLING=(?i)prepayment credits.{0,40}deplet|Please go to AI Studio|Failed to connect to Gemini" + ";;ROBOT_ERROR=\\bERROR\\b|\\bCRITICAL\\b|^Traceback"), + alert_log_cooldown=float(_env("ALERT_LOG_COOLDOWN", "300")), # per-signature re-alert gap + alert_scan_interval=float(_env("ALERT_SCAN_INTERVAL", "10")), + alert_backfill_bytes=int(_env("ALERT_BACKFILL_BYTES", str(8 * 1024 * 1024))), + 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")), + # colon-separated extra roots (mounted Nav2/Pudu map dirs). Any *.yaml+*.pgm + # set found here is rendered to PNG and uploaded like a slam_toolbox map. + extra_map_dirs=[Path(p) for p in _env("EXTRA_MAP_DIRS", "").split(":") if p.strip()], + 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(), + 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")), + map_max_upload_mb=float(_env("MAP_MAX_UPLOAD_MB", "7")), + state_dir=Path(_env("STATE_DIR", "/data/state")), + alert_endpoint=_env("ALERT_ENDPOINT", "/api/v1/fleet/ingest/{sn}/alert"), + logs_endpoint=_env("LOGS_ENDPOINT", "/api/v1/fleet/ingest/{sn}/logs"), + logs_interval=float(_env("LOGS_INTERVAL", "60")), + remote_enable=_env_bool("REMOTE_ENABLE", True), + remote_endpoint=_env("REMOTE_ENDPOINT", "/api/v1/fleet/ingest/{sn}/remote"), + remote_kind=_env("REMOTE_KIND", "web"), + remote_host=_env("REMOTE_HOST", ""), + remote_ports=_env("REMOTE_PORTS", "8001,8014,8011,8012,8013,8000,8080"), + remote_url=_env("REMOTE_URL", ""), # explicit URL (e.g. a public tunnel) wins + remote_interval=float(_env("REMOTE_INTERVAL", "60")), + ssh_enable=_env_bool("SSH_REGISTER", True), + ssh_user=_env("SSH_USER", ""), + ssh_port=int(_env("SSH_PORT", "22")), + control_url=_env("CONTROL_STATUS_URL", ""), # Sanad /api/controller/status + control_enable=_env_bool("CONTROL_ENABLE", False), # remote mode-SWITCH (motion!) — off + project_log_container=_env("PROJECT_LOG_CONTAINER", "auto"), + project_log_path=_env("PROJECT_LOG_PATH", ""), + project_log_label=_env("PROJECT_LOG_LABEL", ""), + project_log_backfill=int(_env("PROJECT_LOG_BACKFILL", "100")), + # drop uvicorn/access-log noise so shipped lines match the project's + # own LIVE LOGS panel exactly (app-logger lines + tracebacks only) + project_log_exclude=_env("PROJECT_LOG_EXCLUDE", + r"^(INFO|WARNING|ERROR|DEBUG|CRITICAL):\s"), + ros_distro=_env("SOFTWARE_ROS", ""), + 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.map_endpoint_tmpl.format(sn=self.sn) + + def alert_url(self) -> str: + return self.server_url + self.alert_endpoint.format(sn=self.sn) + + def logs_url(self) -> str: + return self.server_url + self.logs_endpoint.format(sn=self.sn) + + def remote_url_ep(self) -> str: + return self.server_url + self.remote_endpoint.format(sn=self.sn) + + def auth_headers(self) -> Dict[str, str]: + return {"Authorization": f"Bearer {self.device_token}"} + + +# --------------------------------------------------------------------------- # +# 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)) + + +_AGENT_VERSION = "2026.07.13" + +_software_cache: Optional[Dict[str, Any]] = None + + +def read_software(cfg: Config) -> Dict[str, Any]: + """Robot software/OS card: ROS distro, host OS (via /host), kernel, arch. + The container shares the HOST kernel; the host OS comes from + /host/etc/os-release (the read-only /:/host mount).""" + global _software_cache + if _software_cache is not None: + return _software_cache + sw: Dict[str, Any] = {} + # ROS distro: pinned via SOFTWARE_ROS, else detect host installs /opt/ros/* + ros = cfg.ros_distro + if not ros: + try: + for base in ("/host/opt/ros", "/opt/ros"): + p = Path(base) + if p.is_dir(): + distros = sorted(d.name for d in p.iterdir() if d.is_dir()) + if distros: + ros = ",".join(distros) + break + except Exception: + pass + sw["ros"] = ros or None + # host OS (firmware/OS card) + os_name = os_ver = None + for rel in ("/host/etc/os-release", "/etc/os-release"): + try: + kv = {} + for line in Path(rel).read_text().splitlines(): + if "=" in line: + k, _, v = line.partition("=") + kv[k] = v.strip().strip('"') + os_name = kv.get("PRETTY_NAME") or kv.get("NAME") + os_ver = kv.get("VERSION_ID") + break + except Exception: + continue + sw["os"] = os_name # e.g. "Ubuntu 20.04.6 LTS" + sw["os_version"] = os_ver # e.g. "20.04" + # platform.uname(), not os.uname(): identical .release/.machine on the robot, + # but os.uname() is Unix-only and this must also import on a Windows + # workstation (dry-run / --simulate during development). + u = platform.uname() + sw["kernel"] = u.release # host kernel (shared with the container) + sw["arch"] = u.machine # e.g. aarch64 + sw["python"] = ".".join(map(str, sys.version_info[:3])) + sw["agent"] = f"{log.name} {_AGENT_VERSION}" + _software_cache = sw + return sw + + +_firmware_cache: Optional[Dict[str, Any]] = None + + +def read_firmware_static() -> Dict[str, Any]: + """Board-level firmware card (static, via /host): compute board model + + Jetson L4T/BSP release + kernel. The state source adds live fw versions.""" + global _firmware_cache + if _firmware_cache is not None: + return dict(_firmware_cache) + fw: Dict[str, Any] = {} + # board model, e.g. "NVIDIA Orin NX Developer Kit" + for p in ("/host/sys/firmware/devicetree/base/model", + "/host/proc/device-tree/model", + "/sys/firmware/devicetree/base/model", + "/proc/device-tree/model"): + try: + fw["board"] = Path(p).read_bytes().decode().strip("\x00 \n") + break + except Exception: + continue + # Jetson L4T/BSP: "# R35 (release), REVISION: 3.1, ..." -> "R35.3.1" + for p in ("/host/etc/nv_tegra_release", "/etc/nv_tegra_release"): + try: + head = Path(p).read_text().splitlines()[0] + import re + m = re.search(r"(R\d+).*?REVISION:\s*([\d.]+)", head) + fw["l4t"] = f"{m.group(1)}.{m.group(2)}" if m else head.lstrip("# ").strip() + break + except Exception: + continue + fw["kernel"] = platform.uname().release + _firmware_cache = fw + return dict(fw) + + +_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 + + +# --------------------------------------------------------------------------- # +# state source (telemetry side — degrades to heartbeats if unreachable) +# --------------------------------------------------------------------------- # +def _dig(obj: Any, path: str) -> Any: + """Walk a dotted path over dicts OR message objects. + + "pose.pose.position.x" nested attribute / key + "bms.cell_temp[0]" list index + "joints.leg[*].velocity" WILDCARD — collect that field from EVERY + element, returning a flat list + + The wildcard matters because real robot payloads publish joints as an array + of objects, not an array of numbers: without it, a velocity/temperature + mapping resolves to a list of dicts and silently yields nothing. + Several wildcards compose ("a[*].b[*].c" flattens both levels). + + Returns None for any missing link, so a wrong mapping degrades that one + field to null — never an exception.""" + if not path: + return None + cur = obj + for part in path.split("."): + if cur is None: + return None + idx = None + star = False + if part.endswith("]") and "[" in part: + part, _, raw = part[:-1].partition("[") + if raw == "*": + star = True + else: + try: + idx = int(raw) + except ValueError: + return None + + def _get(o: Any, name: str) -> Any: + if not name: + return o + return o.get(name) if isinstance(o, dict) else getattr(o, name, None) + + if isinstance(cur, list) and part: + # a previous [*] already fanned out — keep mapping across the list + cur = [_get(o, part) for o in cur] + else: + cur = _get(cur, part) + + if star: + if cur is None: + return None + try: + cur = list(cur) + except TypeError: + return None + elif idx is not None: + try: + cur = cur[idx] + except Exception: + return None + if isinstance(cur, list) and cur and isinstance(cur[0], list): + cur = [v for sub in cur for v in (sub if isinstance(sub, list) else [sub])] + return cur + + +def _import_msg(spec: str): + """"sensor_msgs/msg/BatteryState" -> the message class.""" + import importlib + parts = [p for p in spec.replace(".", "/").split("/") if p] + return getattr(importlib.import_module(".".join(parts[:-1])), parts[-1]) + + +def _qos(depth: int = 10): + """Subscription QoS for reading robot telemetry. + + Defaults to BEST_EFFORT, which is the only setting that works against BOTH + kinds of publisher: a RELIABLE subscriber gets NOTHING from a BEST_EFFORT + publisher (incompatible), while a BEST_EFFORT subscriber happily reads from + either. The X2 publishes /aima/mc/leg_odometry as BEST_EFFORT, so the rclpy + default (RELIABLE) silently delivers zero messages — the subscription is + created, no error is raised, and the field just stays null forever. + + Override with X2_ROS_QOS=reliable if a topic ever requires it.""" + from rclpy.qos import HistoryPolicy, QoSProfile, ReliabilityPolicy + want = _env("X2_ROS_QOS", "best_effort").lower() + rel = (ReliabilityPolicy.RELIABLE if want in ("reliable", "rel") + else ReliabilityPolicy.BEST_EFFORT) + return QoSProfile(reliability=rel, history=HistoryPolicy.KEEP_LAST, depth=depth) + + +def _seq(v: Any) -> List[Any]: + if v is None: + return [] + if isinstance(v, (str, bytes)): + return [] + return list(v) if hasattr(v, "__iter__") else [v] + + +class AgiBotSource: + """AGIBOT X2 state source — the ONE robot-specific class in this agent. + + The X2 exposes no fixed topic contract, so NOTHING here is hard-coded: + choose a backend with X2_SOURCE and point it at the real interface with + env vars. + + X2_SOURCE=auto (default) http if X2_STATE_URL is set, else ros2 if rclpy + imports, else aimrt if aimrt_py imports, else none + X2_SOURCE=http poll X2_STATE_URL (JSON); map fields with X2_FIELD_* + X2_SOURCE=ros2 subscribe X2_TOPIC_* (types from X2_TYPE_*, defaulting to + the standard sensor_msgs / nav_msgs ones) + X2_SOURCE=aimrt AimRT channels — see _start_aimrt() below + X2_SOURCE=none never read; always heartbeat + + Whatever the backend, it fills one six-key snapshot — + {bms, state_age, temps, max_vel, xy, fw} — which is the entire seam between + this robot and the shared telemetry / fault / status / alert pipeline. + + Every read is wrapped: a wrong mapping yields null fields and heartbeat mode, + never a crashed loop.""" + + # Defaults match the STANDARD ROS 2 messages. They are almost certainly + # right for a ros2 backend and almost certainly wrong for a vendor http + # payload — which is exactly why they are env-overridable. + _DEFAULTS = { + "soc": "percentage", # sensor_msgs/BatteryState + "voltage": "voltage", + "current": "current", + "temp": "temperature", + "soh": "", + "cycles": "", + "temps": "", # standard JointState has NO temperatures + "vel": "velocity", # sensor_msgs/JointState + "x": "pose.pose.position.x", # nav_msgs/Odometry + "y": "pose.pose.position.y", + "fsm": "", + "fw": "", + } + + 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._state_ts = 0.0 + self._temps: List[float] = [] + self._max_vel = 0.0 + self._xy: Optional[Dict[str, float]] = None + self._fw: Dict[str, Any] = {} + self._fsm: Optional[int] = None + self._loco = None + self._stop = False + self.backend = "none" + self.ok = False + # field map: X2_FIELD_SOC, X2_FIELD_VOLTAGE, X2_FIELD_X, … (empty value + # = that field is unavailable on this robot and reports null) + self._map = {k: _env("X2_FIELD_" + k.upper(), d) for k, d in self._DEFAULTS.items()} + self._soc_scale = _env("X2_SOC_SCALE", "auto").lower() + + def _f(name: str, default: str) -> float: + try: + return float(_env(name, default) or default) + except ValueError: + log.warning("bad %s — using %s", name, default) + return float(default) + + # +1 = positive current means CHARGING (the ROS BatteryState + # convention). Set -1 if the X2 reports the opposite sign. + self._cur_sign = _f("X2_CURRENT_SIGN", "1") + # Unit scaling: the telemetry schema is VOLTS and AMPS. BMS firmware + # commonly publishes mV/mA instead — set 0.001 for those. + self._v_scale = _f("X2_VOLTAGE_SCALE", "1") + self._i_scale = _f("X2_CURRENT_SCALE", "1") + self._start() + + # ---------------- backend selection ---------------- + def _start(self) -> None: + want = (_env("X2_SOURCE", "auto").lower() or "auto") + order = ["http", "ros2", "aimrt"] if want == "auto" else [want] + for b in order: + try: + if b == "none": + break + if b == "http" and self._start_http(): + self.backend = "http" + break + if b == "ros2" and self._start_ros2(): + self.backend = "ros2" + break + if b == "aimrt" and self._start_aimrt(): + self.backend = "aimrt" + break + except Exception as e: + log.warning("X2 source %r failed to start (%s)", b, e) + if self.backend == "none": + log.warning("no X2 state source active (X2_SOURCE=%s) — telemetry runs in " + "heartbeat mode. Set X2_STATE_URL (http) or X2_TOPIC_* (ros2); " + "run tools/probe_x2.sh on the robot to find the real names.", want) + else: + self.ok = True + log.info("X2 source up: backend=%s", self.backend) + + def _start_http(self) -> bool: + """Poll a vendor state endpoint returning one JSON object.""" + url = _env("X2_STATE_URL") + if not url: + return False + self._http_url = url + try: + self._http_period = max(0.2, float(_env("X2_HTTP_INTERVAL", "1") or "1")) + except ValueError: + self._http_period = 1.0 + threading.Thread(target=self._http_loop, daemon=True).start() + log.info("X2 http source: %s every %.1fs", url, self._http_period) + return True + + def _http_loop(self) -> None: + while not self._stop: + try: + r = requests.get(self._http_url, timeout=4) + if r.ok: + self._ingest_all(r.json()) + else: + log.debug("X2 http state: HTTP %s", r.status_code) + except Exception as e: + log.debug("X2 http poll failed: %s", e) + time.sleep(self._http_period) + + def _start_ros2(self) -> bool: + """Subscribe the configured ROS 2 topics. Types are resolved by name, so + a custom vendor message works as long as its package is on PYTHONPATH.""" + try: + import rclpy + from rclpy.node import Node + except Exception as e: + log.debug("rclpy unavailable (%s)", e) + return False + wanted = [ + (_env("X2_TOPIC_BATTERY", "/battery_state"), + _env("X2_TYPE_BATTERY", "sensor_msgs/msg/BatteryState"), self._ingest_battery), + (_env("X2_TOPIC_JOINTS", "/joint_states"), + _env("X2_TYPE_JOINTS", "sensor_msgs/msg/JointState"), self._ingest_joints), + (_env("X2_TOPIC_ODOM", "/odom"), + _env("X2_TYPE_ODOM", "nav_msgs/msg/Odometry"), self._ingest_odom), + ] + if not any(t for t, _, _ in wanted): + return False + rclpy.init(args=None) + node = Node("sanad_api_x2") + n = 0 + for topic, spec, cb in wanted: + if not topic or not spec: + continue + try: + cls = _import_msg(spec) + except Exception as e: + log.warning("X2 ros2: cannot import %s for %s (%s) — skipped", spec, topic, e) + continue + node.create_subscription(cls, topic, cb, _qos()) + log.info("X2 ros2: subscribed %s (%s)", topic, spec) + n += 1 + if not n: + try: + rclpy.shutdown() + except Exception: + pass + return False + self._node = node + threading.Thread(target=lambda: rclpy.spin(node), daemon=True).start() + return True + + def _start_aimrt(self) -> bool: + """AgiBot's own runtime (the X1 stack is built on it). + + NOT implemented natively, on purpose: the X2's AimRT channel names and + message definitions are not public, and guessing them would produce an + agent that imports cleanly, builds, posts perfect-looking heartbeats + with battery:null forever, and only reveals the mistake on the robot. + + The supported path today is AimRT's ROS 2 plugin: enable it on the robot + and run this agent with X2_SOURCE=ros2 against the bridged topics. If you + get the native channel definitions from AgiBot, implement them here — the + only contract to satisfy is calling self._ingest_battery / _ingest_joints + / _ingest_odom with the incoming messages.""" + try: + import aimrt_py # noqa: F401 + except Exception as e: + log.debug("aimrt_py unavailable (%s)", e) + return False + log.warning("aimrt_py is installed, but no native X2 channel binding is built " + "(channel/message definitions are not public). Enable AimRT's ROS 2 " + "plugin and set X2_SOURCE=ros2 with X2_TOPIC_* instead.") + return False + + # ---------------- field extraction ---------------- + def _num(self, obj: Any, key: str) -> Optional[float]: + v = _dig(obj, self._map.get(key, "")) + try: + return float(v) if v is not None else None + except (TypeError, ValueError): + return None + + def _ingest_battery(self, msg: Any) -> None: + try: + soc = self._num(msg, "soc") + if soc is None: + return # no SOC = no battery record; downstream expects an int + if self._soc_scale == "fraction" or (self._soc_scale == "auto" and 0.0 < soc <= 1.0): + soc *= 100.0 # ROS BatteryState.percentage is 0..1; vendors often use 0..100 + cur = self._num(msg, "current") + volt = self._num(msg, "voltage") + temp = self._num(msg, "temp") + soh = self._num(msg, "soh") + cyc = self._num(msg, "cycles") + rec = { + "soc": max(0, min(100, int(round(soc)))), + "current_a": round(cur * self._i_scale * self._cur_sign, 2) if cur is not None else 0.0, + "voltage_v": round(volt * self._v_scale, 1) if volt is not None else None, + "temp_c": int(round(temp)) if temp is not None and -40 <= temp <= 150 else None, + "soh": int(round(soh)) if soh is not None else 0, + "cycles": int(round(cyc)) if cyc is not None else 0, + } + with self._lock: + self._bms = rec + self._bms_ts = self._state_ts = time.monotonic() + except Exception: + pass + + def _ingest_joints(self, msg: Any) -> None: + try: + temps: List[float] = [] + for x in _seq(_dig(msg, self._map.get("temps", ""))): + try: + f = float(x) + except (TypeError, ValueError): + continue + if 0 < f <= 200: # 0 = slot not reporting, same rule as g1 + temps.append(f) + max_vel = 0.0 + for x in _seq(_dig(msg, self._map.get("vel", ""))): + try: + max_vel = max(max_vel, abs(float(x))) + except (TypeError, ValueError): + pass + with self._lock: + self._temps = temps + self._max_vel = max_vel + self._state_ts = time.monotonic() + except Exception: + pass + + def _ingest_odom(self, msg: Any) -> None: + try: + x, y = self._num(msg, "x"), self._num(msg, "y") + if x is None or y is None: + return + with self._lock: + self._xy = {"x": round(x, 3), "y": round(y, 3)} + self._state_ts = time.monotonic() + except Exception: + pass + + def _ingest_all(self, d: Any) -> None: + """One combined payload (http backend) feeds all three extractors.""" + self._ingest_battery(d) + self._ingest_joints(d) + self._ingest_odom(d) + try: + fsm = _dig(d, self._map.get("fsm", "")) + fw = _dig(d, self._map.get("fw", "")) + with self._lock: + if fsm is not None: + try: + self._fsm = int(fsm) + except (TypeError, ValueError): + pass + if isinstance(fw, dict): + self._fw.update({str(k): str(v) for k, v in fw.items()}) + except Exception: + pass + + # ---------------- the snapshot contract ---------------- + def snapshot(self) -> Dict[str, Any]: + with self._lock: + now = time.monotonic() + return { + "bms": dict(self._bms) if self._bms else None, + "state_age": (now - self._state_ts) if self._state_ts else None, + "temps": list(self._temps), + "max_vel": self._max_vel, + "xy": dict(self._xy) if self._xy else None, + "fw": dict(self._fw), + } + + def fsm_id(self) -> Optional[int]: + """Only the http backend can carry one (X2_FIELD_FSM); there is no + known X2 loco RPC. Read-only either way — never commands motion.""" + with self._lock: + return self._fsm + + +# AGIBOT X2 publishes no documented loco FSM id scheme. control.mode comes +# from the read-only CONTROL_STATUS_URL when the robot exposes one; fill +# these in once the real ids are known (empty = mode reported as unknown). +_FSM_STATUS: Dict[int, str] = {} +# Control-panel mode labels + the switchable set (fsm_id -> friendly mode). +_CONTROL_MODES: Dict[int, str] = {} +_CONTROL_SWITCHABLE: List[str] = [] + + +_control_cache: Dict[str, Any] = {"ts": 0.0, "data": None} + + +def read_control(cfg: Config) -> Optional[Dict[str, Any]]: + """READ-ONLY control status from the Sanad dashboard's /api/controller/status + (no robot control, no motion). Reports the current loco mode + the switchable set. + Remote SWITCHING is a separate, motion-capable path gated by CONTROL_ENABLE.""" + now = time.monotonic() + if _control_cache["data"] is not None and now - _control_cache["ts"] < 1.5: + return _control_cache["data"] + url = cfg.control_url + if not url: + port = _REMOTE_STAT.get("port") + if not port: + return None + url = f"http://127.0.0.1:{port}/api/controller/status" + try: + r = requests.get(url, timeout=2) + d = r.json() if r.ok else None + except Exception: + d = None + if not isinstance(d, dict): + _control_cache.update(ts=now, data=None) + return None + fid = d.get("fsm_id") + out = { + "fsm_id": fid, + "mode": _CONTROL_MODES.get(fid, "unknown") if fid is not None else None, + "armed": d.get("armed"), + "walk_ready": d.get("walk_ready"), + "teleop_active": d.get("teleop_active"), + "last_velocity": d.get("last_velocity"), + "sdk_available": d.get("sdk_available"), + "switchable_modes": _CONTROL_SWITCHABLE, + "remote_switch_enabled": bool(cfg.control_enable), # remote mode-switch armed? + } + _control_cache.update(ts=now, data=out) + return out + + +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 + + +class Ros2Position: + """Position from a ROS 2 odometry topic, INDEPENDENT of X2_SOURCE. + + The X2's battery comes cleanly from the Control Dash over http, but that + payload carries no odometry — while ROS publishes it on + /aima/mc/leg_odometry (nav_msgs/Odometry). Those are two different + transports, so position is its own source rather than part of the state + backend: X2_SOURCE=http can run with X2_POSITION_SOURCE=ros2 at the + same time. + + Enable with X2_POSITION_SOURCE=ros2 (topic: X2_TOPIC_ODOM, type: + X2_TYPE_ODOM, fields: X2_FIELD_X / X2_FIELD_Y). Requires rclpy — the + systemd unit must source the ROS overlay first. If rclpy is missing or the + topic never publishes, position simply stays null; nothing else is affected.""" + + def __init__(self, cfg: Config): + self.cfg = cfg + self._xy: Optional[Dict[str, float]] = None + self._lock = threading.Lock() + self.ok = False + # read here, NOT as class attributes: the class body runs at import, + # which is before main() calls _load_dotenv(). + self._px = _env("X2_FIELD_X", "") or "pose.pose.position.x" + self._py = _env("X2_FIELD_Y", "") or "pose.pose.position.y" + topic = _env("X2_TOPIC_ODOM", "/odom") + spec = _env("X2_TYPE_ODOM", "nav_msgs/msg/Odometry") + if not topic or not spec: + log.warning("X2_POSITION_SOURCE=ros2 but X2_TOPIC_ODOM/X2_TYPE_ODOM is empty") + return + try: + import rclpy + from rclpy.node import Node + except Exception as e: + log.warning("rclpy unavailable (%s) — position stays null. The unit must " + "source the ROS overlay before starting the agent.", e) + return + try: + cls = _import_msg(spec) + except Exception as e: + log.warning("cannot import %s (%s) — position stays null", spec, e) + return + try: + # AgiBotSource may already have started rclpy when X2_SOURCE=ros2. + if not rclpy.ok(): + rclpy.init(args=None) + node = Node("sanad_api_x2_pos") + node.create_subscription(cls, topic, self._on_odom, _qos()) + self._node = node + threading.Thread(target=lambda: rclpy.spin(node), daemon=True).start() + self.ok = True + log.info("X2 position: ros2 %s (%s)", topic, spec) + except Exception as e: + log.warning("ros2 position init failed (%s) — position stays null", e) + + def _on_odom(self, msg: Any) -> None: + try: + x = _dig(msg, self._px) + y = _dig(msg, self._py) + if x is None or y is None: + return + with self._lock: + self._xy = {"x": round(float(x), 3), "y": round(float(y), 3)} + except Exception: + pass + + 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 # .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: + return f"{self.fmt}:{self.size}:{self.mtime}" + + +def _sha256(path: Path) -> str: + h = hashlib.sha256() + with path.open("rb") as f: + for chunk in iter(lambda: f.read(1024 * 1024), b""): + h.update(chunk) + return h.hexdigest() + + +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"] + roots += list(cfg.extra_map_dirs) # Nav2/Pudu map dirs mounted via EXTRA_MAP_DIRS + 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)) + # Pudu/Nav2 converter emits a plain map + a keepout-BAKED twin (obstacles baked + # in for Foxy, which has no KeepoutFilter). The baked one is the deploy map — drop + # the redundant plain twin so the fleet server gets one canonical map, not two. + baked = {m.stem[: -len("_keepout_baked")] for m in out if m.stem.endswith("_keepout_baked")} + out = [m for m in out if m.stem not in baked] + return out + + +def _map_key(stem: str) -> str: + stem = Path(stem).name + if stem.endswith(".db"): + stem = stem[:-3] + return "".join(c for c in stem if c.isalnum() or c in "_-.") + + +def _read_json(path: Path, default: Any) -> Any: + try: + return json.loads(path.read_text() or "") + except Exception: + return default + + +def _yaw_from_pose(pose: Dict[str, Any]) -> float: + 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)) + return math.atan2(2.0 * (qw * qz + qx * qy), + 1.0 - 2.0 * (qy * qy + qz * qz)) + return float(pose.get("yaw", 0.0)) + + +def _places_files_for(cfg: Config, stem: str) -> List[Path]: + out: List[Path] = [] + key = _map_key(stem) + 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]]: + for pf in _places_files_for(cfg, stem): + data = _read_json(pf, None) if pf.exists() else None + if isinstance(data, dict) and data: + pts: List[Dict[str, Any]] = [] + for name, pose in data.items(): + if not isinstance(pose, dict): + continue + try: + pts.append({ + "name": name, + "type": str(pose.get("type", "waypoint")), + "x": float(pose["x"]), + "y": float(pose["y"]), + "yaw": round(_yaw_from_pose(pose), 4), + }) + except (KeyError, TypeError, ValueError): + continue + return pts + return [] + + +def discover_maps(cfg: Config) -> List[MapArtifact]: + 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 = set() + out: List[MapArtifact] = [] + for root in roots: + if not root.exists(): + continue + for p in sorted(root.glob("*.db")): + rp = str(p.resolve()) + if rp in seen: + continue + 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), + 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]: + if not cfg.web_nav3_url: + return None + try: + r = requests.get(cfg.web_nav3_url + "/api/status", + headers={"X-Robot-Name": cfg.robot}, + timeout=min(cfg.http_timeout, 5)) + r.raise_for_status() + am = (r.json() or {}).get("active_map") + return _map_key(am) if am else None + except requests.RequestException: + return None + + +def select_maps(cfg: Config, maps: List[MapArtifact]) -> List[MapArtifact]: + if not maps: + return [] + if cfg.map_select == "newest": + return maps[:1] + if cfg.map_select == "active": + active = _active_map_name(cfg) + if active: + picked = [m for m in maps if _map_key(m.stem) == active] + if picked: + return picked + return maps[:1] + return maps # "all" + + +def _state_file(cfg: Config) -> Path: + return cfg.state_dir / "uploaded.json" + + +def load_state(cfg: Config) -> Dict[str, str]: + return _read_json(_state_file(cfg), {}) if _state_file(cfg).exists() else {} + + +def save_state(cfg: Config, state: Dict[str, str]) -> None: + try: + 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 map state: %s", e) + + +def build_meta(cfg: Config, m: MapArtifact) -> Dict[str, Any]: + return { + "sn": cfg.sn, + "name": m.stem, + "file": m.name, + "format": m.fmt, + "size_bytes": m.size, + "sha256": m.sha256, + "mtime": m.mtime, + "description": m.description, + "points": m.points, + } + + +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.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(), + timeout=cfg.http_timeout, verify=cfg.verify_tls) + else: # multipart (default) + with m.path.open("rb") as fh: + files = {"db": (m.name, fh, "application/octet-stream")} + data = {"meta": json.dumps(meta)} + resp = session.post(url, files=files, data=data, + 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.name, e) + return False + if not resp.ok: + log.error("map upload %s FAILED: HTTP %s %s", m.name, resp.status_code, resp.text[:300]) + return False + 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 + + +# 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: + _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 = failed = unstable = too_large = current = 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(): + current += 1 + 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) + unstable += 1 + continue + # server rejects bodies over ~8 MB (client_max_body_size) — don't burn + # bandwidth on uploads that will 413. Raster (slam_toolbox) maps are tiny. + if m.fmt != "slam_toolbox" and (m.size / 1048576) > cfg.map_max_upload_mb: + log.warning("map %s is %.0f MB — exceeds server upload cap (~%.0f MB), skipping " + "(export a raster map or raise the server limit)", + m.name, m.size / 1048576, cfg.map_max_upload_mb) + too_large += 1 + continue + 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 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 failed: + _set_map_status(state="failed", uploaded=False, maps_found=len(maps), + last_map=maps[0].stem, error=last_err) + elif uploaded or current: + # at least one map is on the server (just now or previously); note skips + note = None + if too_large: + note = f"{too_large} map(s) skipped: exceed server upload cap (~{cfg.map_max_upload_mb:.0f} MB)" + elif unstable: + note = "newer map still being written (mapping in progress)" + _set_map_status(state="uploaded", uploaded=True, maps_found=len(maps), + last_map=maps[0].stem, error=note) + elif too_large: + _set_map_status(state="failed", uploaded=False, maps_found=len(maps), + last_map=maps[0].stem, + error=f"map exceeds server upload cap (~{cfg.map_max_upload_mb:.0f} MB) — " + "export a raster map or raise the server limit") + elif unstable: + # newest content is still being written (active mapping) — be honest + _set_map_status(state="pending", uploaded=False, maps_found=len(maps), + last_map=maps[0].stem, + error="map still being written (mapping in progress) — " + "will upload when it settles") + else: + _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) + + +# --------------------------------------------------------------------------- # +# logs + alerts (spec: POST /{sn}/logs periodically, POST /{sn}/alert on events) +# --------------------------------------------------------------------------- # +class _RingLogHandler(logging.Handler): + """Buffers the agent's own log lines so they can be shipped to the server.""" + + def __init__(self, maxlen: int = 400): + super().__init__(level=logging.INFO) + from collections import deque + self._buf: Any = deque(maxlen=maxlen) + self._blk = threading.Lock() + + def emit(self, record: logging.LogRecord) -> None: + try: + with self._blk: + self._buf.append(self.format(record)) + except Exception: + pass + + def drain(self) -> List[str]: + with self._blk: + lines = list(self._buf) + self._buf.clear() + return lines + + def requeue(self, lines: List[str]) -> None: + """Put unshipped lines back (front of the ring) so they retry next cycle + instead of being lost — bounded by maxlen, oldest evicted first.""" + with self._blk: + self._buf.extendleft(reversed(lines)) + + +_LOG_RING = _RingLogHandler() + +# shipped-status shown in every telemetry post ("logs" / "alerts" fields) +_LOGS_STAT: Dict[str, Any] = {"last_sent": None, "lines_sent": 0, "ok": None} +_ALERTS_STAT: Dict[str, Any] = {"sent": 0, "last": None, "last_time": None, "ok": None} + +# start times ("started_at" = this run, "last_start" = previous run) +_STARTED: Dict[str, Any] = {"now": None, "prev": None, "mono": time.monotonic()} + + +def _init_start_times(cfg: Config) -> None: + """Record this agent start; remember the previous one (persisted in STATE_DIR).""" + f = cfg.state_dir / "agent_state.json" + prev = (_read_json(f, {}) or {}).get("started_at") + now_s = _now_str() + try: + cfg.state_dir.mkdir(parents=True, exist_ok=True) + f.write_text(json.dumps({"started_at": now_s})) + except Exception as e: + log.debug("could not persist start time: %s", e) + _STARTED.update(now=now_s, prev=prev, mono=time.monotonic()) + + +class ProjectLogTail: + """Tails the robot's main PROJECT logs (e.g. the sanadr1 / sanad-p4 app) + and feeds them into the shipped log lines, labeled "[-logs] …". + + Sources, in priority order: + PROJECT_LOG_PATH explicit log file (or dir -> newest *.log) via /host + PROJECT_LOG_CONTAINER a docker container name; "auto" (default) scans the + host's docker metadata (/host/var/lib/docker) for a + RUNNING Sanad project (sanadr1, sanad-p4, sanad*) + Reads the container's json-log through the read-only /:/host mount — no + docker socket needed, read-only, cannot disturb the project.""" + + KNOWN = ("sanadr1", "sanad-p4", "sanadv3", "sanad") + + def __init__(self, cfg: Config): + self.label: Optional[str] = None + self._cur: Optional[Path] = None + self._pos = 0 + self._backfill = max(0, cfg.project_log_backfill) + self._exclude = None + if cfg.project_log_exclude: + try: + import re + self._exclude = re.compile(cfg.project_log_exclude) + except Exception: + self._exclude = None + self.active = False + try: + self._resolve(cfg) + except Exception as e: + log.debug("project-log resolve failed: %s", e) + if self.active: + log.info("project logs: sharing '%s' (%s)", self.label, self._cur) + else: + log.info("project logs: none found (project_logs=null)") + + def _resolve(self, cfg: Config) -> None: + # explicit file/dir + if cfg.project_log_path: + p = Path(cfg.project_log_path) + if p.is_dir(): + logs = sorted(p.glob("*.log"), key=lambda f: f.stat().st_mtime, reverse=True) + p = logs[0] if logs else None + if p and p.exists(): + self._start(p, cfg.project_log_label or f"{p.stem}-logs") + return + # docker container json-log via /host + base = Path("/host/var/lib/docker/containers") + if not base.exists(): + return + want = cfg.project_log_container + candidates: List[Any] = [] + for cf in base.glob("*/config.v2.json"): + try: + d = json.loads(cf.read_text()) + except Exception: + continue + name = (d.get("Name") or "").lstrip("/") + running = bool((d.get("State") or {}).get("Running")) + lp = d.get("LogPath") or "" + if not name or not lp: + continue + if want != "auto": + if name == want: + candidates.append((0, name, lp, running)) + elif running and "sanad" in name.lower() and not name.startswith("sanad-api"): + # rank known Sanad projects first + rank = self.KNOWN.index(name) if name in self.KNOWN else len(self.KNOWN) + candidates.append((rank, name, lp, running)) + if not candidates: + return + candidates.sort(key=lambda c: c[0]) + _, name, lp, _ = candidates[0] + p = Path("/host" + lp) if not lp.startswith("/host") else Path(lp) + if p.exists(): + self._start(p, cfg.project_log_label or f"{name}-logs") + + def _start(self, p: Path, label: str) -> None: + self._cur = p + self.label = label + size = p.stat().st_size + self._pos = size # new lines ship from here on + self._pending: List[str] = [] + # backfill: the last N RELEVANT lines (post-filter) from the last ~1 MB, + # so the noise (access-log spam) doesn't eat the history window + if self._backfill and size: + try: + take = min(size, 8 * 1024 * 1024) + with p.open("rb") as f: + f.seek(size - take) + tail = f.read(take) + raw = tail.decode("utf-8", "replace").splitlines() + if take < size and raw: + raw = raw[1:] # first line is a partial record — drop it + for ln in raw: + ln = ln.strip() + if ln.startswith("{"): + try: + ln = (json.loads(ln).get("log") or "").rstrip() + except Exception: + pass + if not ln: + continue + if self._exclude is not None and self._exclude.search(ln): + continue + self._pending.append(f"[{label}] {ln}") + self._pending = self._pending[-self._backfill:] + except Exception: + self._pending = [] + self.active = True + + def poll(self) -> List[str]: + """New lines since last poll (docker json-log unwrapped), labeled. + The startup backfill (last N relevant lines) is returned on first call.""" + if not self.active or self._cur is None: + return [] + pend, self._pending = self._pending, [] + out: List[str] = [] + try: + st = self._cur.stat() + if st.st_size < self._pos: # log rotated + self._pos = 0 + if st.st_size > self._pos: + with self._cur.open("rb") as f: + f.seek(self._pos) + chunk = f.read(min(st.st_size - self._pos, 256 * 1024)) + self._pos = f.tell() + for ln in chunk.decode("utf-8", "replace").splitlines(): + ln = ln.strip() + if ln.startswith("{"): + try: + ln = (json.loads(ln).get("log") or "").rstrip() + except Exception: + pass + if not ln: + continue + if self._exclude is not None and self._exclude.search(ln): + continue # access-log noise — not in the project's log panel + out.append(f"[{self.label}] {ln}") + except Exception as e: + log.debug("project-log poll failed: %s", e) + return pend + out[-100:] # backfill first, then ≤100 new/cycle + + +_PROJECT_TAIL: Optional[ProjectLogTail] = None + + +def ship_logs(cfg: Config, session: requests.Session) -> None: + """POST buffered agent log lines (+ project logs) to /{sn}/logs. + Best-effort: failures are logged at DEBUG only (below the ring's level -> + no feedback loop).""" + lines = _LOG_RING.drain() + if _PROJECT_TAIL is not None and _PROJECT_TAIL.active: + lines.extend(_PROJECT_TAIL.poll()) + if not lines: + return + try: + r = session.post(cfg.logs_url(), + json={"sn": cfg.sn, "name": cfg.name, + "lines": lines, "ts": int(time.time())}, + headers=cfg.auth_headers(), + timeout=cfg.http_timeout, verify=cfg.verify_tls) + if r.ok: + _LOGS_STAT.update(last_sent=_now_str(), ok=True) + _LOGS_STAT["lines_sent"] += len(lines) + log.debug("logs shipped: %d lines -> HTTP %s", len(lines), r.status_code) + else: + _LOGS_STAT["ok"] = False + _LOG_RING.requeue(lines) # retry next cycle (server keeps 500ing) + log.debug("logs ship failed: HTTP %s (%d lines requeued)", r.status_code, len(lines)) + except requests.RequestException as e: + _LOGS_STAT["ok"] = False + _LOG_RING.requeue(lines) + log.debug("logs ship failed (%d lines requeued): %s", len(lines), e) + + +def logs_loop(cfg: Config, session: requests.Session) -> None: + while True: + time.sleep(cfg.logs_interval) + try: + ship_logs(cfg, session) + except Exception: + pass + + +# --------------------------------------------------------------------------- # +# remote dashboard — discover the Sanad web UI and register its URL for the +# fleet dashboard to embed ("full dashboard through the fleet API"). No changes +# to the Sanad app: we only probe its port and POST the URL to /{sn}/remote. +# --------------------------------------------------------------------------- # +_REMOTE_STAT: Dict[str, Any] = {"url": None, "port": None, "kind": None, "ok": None, + "ssh": None, "ssh_ok": None} + + +def _primary_ip() -> str: + import socket + try: + s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + s.connect(("1.1.1.1", 80)) + ip = s.getsockname()[0] + s.close() + return ip + except Exception: + return "127.0.0.1" + + +def discover_dashboard(cfg: Config) -> Optional[Dict[str, Any]]: + """Find the Sanad dashboard: probe each candidate port on localhost; the one + that returns 200 with a 'Sanad'/dashboard page wins. Returns {url, port}.""" + if cfg.remote_url: # explicit URL (e.g. a tunnel) overrides + return {"url": cfg.remote_url, "port": None} + host = cfg.remote_host or _primary_ip() # the LAN IP the fleet can reach + for tok in cfg.remote_ports.split(","): + tok = tok.strip() + if not tok.isdigit(): + continue + port = int(tok) + try: + r = requests.get(f"http://127.0.0.1:{port}/", timeout=2) + except requests.RequestException: + continue + if r.status_code == 200 and ("sanad" in r.text.lower() or "dashboard" in r.text.lower()): + return {"url": f"http://{host}:{port}", "port": port} + return None + + +def _post_remote(cfg: Config, session: requests.Session, body: Dict[str, Any]) -> bool: + try: + r = session.post(cfg.remote_url_ep(), json=body, headers=cfg.auth_headers(), + timeout=cfg.http_timeout, verify=cfg.verify_tls) + if not r.ok: + log.warning("remote(%s) register failed: HTTP %s %s", + body.get("kind"), r.status_code, r.text[:120]) + return bool(r.ok) + except requests.RequestException as e: + log.debug("remote(%s) register failed: %s", body.get("kind"), e) + return False + + +def register_remote(cfg: Config, session: requests.Session) -> None: + if not cfg.remote_enable: + return + host = cfg.remote_host or _primary_ip() + # 1) the Sanad dashboard (kind=web) — the exact UI, for the fleet to embed + d = discover_dashboard(cfg) + if d: + ok = _post_remote(cfg, session, { + "sn": cfg.sn, "name": cfg.name, "kind": cfg.remote_kind, + "url": d["url"], "label": f"{cfg.name} — Sanad Dashboard", + "port": d["port"], "ts": int(time.time())}) + _REMOTE_STAT.update(url=d["url"], port=d["port"], kind=cfg.remote_kind, ok=ok) + if ok: + log.info("remote dashboard registered: %s -> HTTP 200", d["url"]) + else: + _REMOTE_STAT.update(url=None, port=None, ok=None) + # 2) SSH access (kind=ssh) — "ssh @". Skipped when the login user + # isn't known: registering a guessed username sends the fleet UI a command + # that silently fails for whoever tries it. + if cfg.ssh_enable and cfg.ssh_user: + cmd = f"ssh {cfg.ssh_user}@{host}" + if cfg.ssh_port != 22: + cmd += f" -p {cfg.ssh_port}" + ok = _post_remote(cfg, session, { + "sn": cfg.sn, "name": cfg.name, "kind": "ssh", + "url": f"ssh://{cfg.ssh_user}@{host}:{cfg.ssh_port}", # URL-valid form + "command": cmd, # "ssh @" + "host": host, "port": cfg.ssh_port, "user": cfg.ssh_user, + "label": f"{cfg.name} — SSH", "ts": int(time.time())}) + _REMOTE_STAT.update(ssh=cmd, ssh_ok=ok) + if ok: + log.info("remote SSH registered: %s -> HTTP 200", cmd) + + +def remote_loop(cfg: Config, session: requests.Session) -> None: + while True: + try: + register_remote(cfg, session) + except Exception as e: + log.debug("remote loop: %s", e) + time.sleep(cfg.remote_interval) + + +_ALERT_SEEN: Dict[str, float] = {} # fault CODE -> monotonic ts of last alert + + +def _post_alert(cfg: Config, session: requests.Session, text: str) -> bool: + """POST a single alert string to /{sn}/alert; updates the alert status.""" + try: + r = session.post(cfg.alert_url(), + json={"sn": cfg.sn, "name": cfg.name, + "alert": text, "message": text, "ts": int(time.time())}, + headers=cfg.auth_headers(), + timeout=cfg.http_timeout, verify=cfg.verify_tls) + _ALERTS_STAT.update(last=text, last_time=_now_str(), ok=bool(r.ok)) + if r.ok: + _ALERTS_STAT["sent"] += 1 + log.info("alert sent: %s -> HTTP %s", text[:120], r.status_code) + return bool(r.ok) + except requests.RequestException as e: + _ALERTS_STAT.update(last=text, last_time=_now_str(), ok=False) + log.debug("alert send failed (%s): %s", text[:60], e) + return False + + +def _fault_code(f: str) -> str: + """Dedup key for a fault string: the CODE before the first ':'.""" + return (f.split(":", 1)[0].strip() or f) + + +def send_alerts(cfg: Config, session: requests.Session, faults: List[str]) -> None: + """POST each NEW fault (rising edge) to /{sn}/alert. Faults are strings. + Includes LOW_BATTERY (<= LOW_SOC, default 50%), MOTOR_OVERTEMP, COMMS_STALE. + + Deduped on the fault CODE, never on the whole string: every fault embeds a + LIVE number ("battery 49%", "motor temp 87C", "no robot state for 12s") that + changes almost every tick, so string-dedup re-fired the same fault every + POLL_INTERVAL for as long as it lasted — COMMS_STALE counts up by 1 s + forever, which meant an alert POST every 2 s until the state source came back. + + A code alerts on its rising edge, then at most once per ALERT_LOG_COOLDOWN + while it persists (so a long outage still re-asserts, but doesn't flood). + Clearing the fault drops the code, so the next occurrence alerts again.""" + now = time.monotonic() + current: Dict[str, str] = {} + for f in faults: + current.setdefault(_fault_code(f), f) + for code in [c for c in _ALERT_SEEN if c not in current]: + del _ALERT_SEEN[code] # cleared — next occurrence is a rising edge again + for code, text in sorted(current.items()): + last = _ALERT_SEEN.get(code) + if last is None or (now - last) > cfg.alert_log_cooldown: + _ALERT_SEEN[code] = now + _post_alert(cfg, session, text) + + +class LogAlertScanner: + """Scans the robot's project logs (sanadr1) for error/billing patterns and + fires an alert on each NEW signature (deduped with a cooldown). Catches the + Gemini 'credits depleted' billing error and any ERROR/Traceback the app logs.""" + + def __init__(self, cfg: Config): + import re + self._path: Optional[Path] = None + self._pos = 0 + self._cooldown = cfg.alert_log_cooldown + self._seen: Dict[str, float] = {} + self._patterns: List[Any] = [] + self._pending: List[Any] = [] + for entry in cfg.alert_log_patterns.split(";;"): + if "=" in entry: + code, rx = entry.split("=", 1) + try: + # case-SENSITIVE (uppercase log levels); use (?i) inline for text + self._patterns.append((code.strip(), re.compile(rx))) + except re.error as e: + log.warning("bad alert pattern %s: %s", code, e) + if _PROJECT_TAIL is not None and _PROJECT_TAIL.active and _PROJECT_TAIL._cur: + self._path = _PROJECT_TAIL._cur + try: + size = self._path.stat().st_size + self._pos = size # ongoing reads start at EOF + # STARTUP BACKFILL: scan a large tail once (access-logs bury real + # errors) so a currently-active error (e.g. Gemini billing) alerts + # right away. Only the matching lines are kept — cheap. + back = min(size, cfg.alert_backfill_bytes) + with self._path.open("rb") as f: + f.seek(size - back) + raw = f.read(back).decode("utf-8", "replace").splitlines() + if back < size and raw: + raw = raw[1:] # first line is partial + matches = [] + for ln in raw: + ln = ln.strip() + if ln.startswith("{"): + try: + ln = (json.loads(ln).get("log") or "").rstrip() + except Exception: + pass + if not ln: + continue + for code, rx in self._patterns: + if rx.search(ln): + matches.append((code, ln)) + break + # keep the most-recent unique signature per (code+line), cap 20 + seen = set() + uniq = [] + for code, ln in reversed(matches): + sig = code + ":" + re.sub(r"\d+", "#", ln)[:120] + if sig in seen: + continue + seen.add(sig) + uniq.append((code, ln)) + self._pending = list(reversed(uniq))[-20:] + except Exception: + self._pos = 0 + if self._path and self._patterns: + log.info("log-alert scan on %s (%d patterns, %d backfilled)", + self._path.name, len(self._patterns), len(self._pending)) + + def scan(self, cfg: Config, session: requests.Session) -> None: + if not self._path or not self._patterns: + return + import re + now0 = time.monotonic() + if self._pending: # flush startup backfill first + pend, self._pending = self._pending, [] + for code, ln in pend: + sig = code + ":" + re.sub(r"\d+", "#", ln)[:120] + self._seen[sig] = now0 + _post_alert(cfg, session, f"{code}: {ln[:220]}") + try: + st = self._path.stat() + if st.st_size < self._pos: # rotated + self._pos = 0 + if st.st_size <= self._pos: + return + with self._path.open("rb") as f: + f.seek(self._pos) + chunk = f.read(min(st.st_size - self._pos, 512 * 1024)) + self._pos = f.tell() + except Exception as e: + log.debug("alert scan read failed: %s", e) + return + now = time.monotonic() + for ln in chunk.decode("utf-8", "replace").splitlines(): + ln = ln.strip() + if ln.startswith("{"): + try: + ln = (json.loads(ln).get("log") or "").rstrip() + except Exception: + pass + if not ln: + continue + for code, rx in self._patterns: + if rx.search(ln): + sig = code + ":" + re.sub(r"\d+", "#", ln)[:120] # dedup key + if now - self._seen.get(sig, -1e9) > self._cooldown: + self._seen[sig] = now + _post_alert(cfg, session, f"{code}: {ln[:220]}") + break + + +_LOG_ALERTS: Optional[LogAlertScanner] = None + + +def alert_scan_loop(cfg: Config, session: requests.Session) -> None: + while True: + time.sleep(cfg.alert_scan_interval) + try: + if _LOG_ALERTS is not None: + _LOG_ALERTS.scan(cfg, session) + except Exception as e: + log.debug("alert scan loop: %s", e) + + +# --------------------------------------------------------------------------- # +# telemetry assembly +# --------------------------------------------------------------------------- # +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(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("state_age") is not None and snap["state_age"] > 3.0: + faults.append(f"COMMS_STALE: no robot state for {snap['state_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("state_age") is not None and snap["state_age"] <= 3.0 + if not alive and bms is None: + return "offline" + if charging: + return "charging" + if snap.get("max_vel", 0.0) > 0.15: + return "moving" + return base or "idle" + + +def build_telemetry(cfg: Config, mac: str, reader: Optional[AgiBotSource], + 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}, + "state_age": 0.1, "temps": [sim.get("temp", 45)], "max_vel": sim.get("max_vel", 0.0), + "xy": sim.get("position")} + fsm = sim.get("fsm") + else: + snap = reader.snapshot() if reader else {"bms": None, "state_age": None, "temps": [], "max_vel": 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. x2_10) + "mac": mac, + "brand": cfg.brand, + "type": cfg.robot_type, # humanoid + "model": cfg.model, # g1 + "software": read_software(cfg), # ros/os/kernel/arch/python/agent + "firmware": {**read_firmware_static(), **(snap.get("fw") or {})}, + # board/l4t/kernel + live robot/bms fw versions + "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 + "control": read_control(cfg), # loco mode (zero_torque/damp/lock/running) + switchable set + "faults": faults, + "map": get_map_status(), # SHOWS whether the saved map made it to the server + "logs": dict(_LOGS_STAT), # log-shipping status (last_sent, lines_sent, ok) + "project_logs": (_PROJECT_TAIL.label + if _PROJECT_TAIL is not None and _PROJECT_TAIL.active + else None), # e.g. "sanadr1-logs"; null = no project found + "remote": (dict(_REMOTE_STAT) if _REMOTE_STAT.get("url") else None), + # Sanad dashboard URL registered for the fleet UI + "alerts": dict(_ALERTS_STAT), # alert status (sent, last, last_time, ok) + "time": _now_str(), # full date+time of this post + "started_at": _STARTED["now"], # when this agent run started + "last_start": _STARTED["prev"], # previous agent start (null on first ever) + "uptime_s": int(time.monotonic() - _STARTED["mono"]), + "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_vel": 0.4 if moving else 0.0, "fsm": None, + "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) + 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) + 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="AGIBOT X2 fleet agent: telemetry + map sync") + ap.add_argument("--simulate", action="store_true", help="synthetic robot state (map scan stays real)") + ap.add_argument("--once", action="store_true", help="one map pass + one telemetry post, then exit") + ap.add_argument("--map-only", action="store_true", help="upload discovered maps once (no state source, no telemetry), 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("--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") + # buffer our own log lines for shipping to /{sn}/logs + _LOG_RING.setFormatter(logging.Formatter("%(asctime)s %(levelname)s %(name)s: %(message)s")) + logging.getLogger().addHandler(_LOG_RING) + _load_dotenv() + cfg = Config.from_env() + if args.interval is not None: + cfg.poll_interval = args.interval + + if args.list: + cmd_list(cfg) + return 0 + + _init_start_times(cfg) + global _PROJECT_TAIL, _LOG_ALERTS + _PROJECT_TAIL = ProjectLogTail(cfg) + _LOG_ALERTS = LogAlertScanner(cfg) # error/billing alerts from the project logs + mac = read_mac(cfg.mac_interface) + log.info("sanad_api_x2 — 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.net_interface, + cfg.position_source, cfg.maps_dir, cfg.map_poll_interval, + " [SIMULATE]" if args.simulate else "") + + session = requests.Session() + + if args.map_only: + # map upload only — search all map roots (incl. EXTRA_MAP_DIRS) and ship, + # without opening a state source or posting telemetry (won't disturb a live feed). + map_sync_once(cfg, session, force=args.force, dry_run=args.dry_run) + return 0 + + reader = None + pos = None + if not args.simulate: + reader = AgiBotSource(cfg) + if cfg.position_source == "rosbridge": + pos = RosbridgePosition(cfg) + elif cfg.position_source == "ros2": + # independent of X2_SOURCE: the Control Dash has no odometry, ROS does + pos = Ros2Position(cfg) + time.sleep(1.0) + + 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) + send_alerts(cfg, session, payload.get("faults") or []) + tick += 1 + + if args.once or args.dry_run: + # 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 cfg.remote_enable and not args.dry_run: + register_remote(cfg, session) + elif cfg.remote_enable: + d = discover_dashboard(cfg) + if d: + _REMOTE_STAT.update(url=d["url"], port=d["port"], kind=cfg.remote_kind, ok=None) + if args.once: + one_telemetry() + ship_logs(cfg, session) + return 0 + for _ in range(3): + one_telemetry() + time.sleep(min(cfg.poll_interval, 1.0)) + return 0 + + # loop mode: map sync + log shipping + remote registration in bg threads + threading.Thread(target=map_loop, args=(cfg, session), daemon=True).start() + threading.Thread(target=logs_loop, args=(cfg, session), daemon=True).start() + threading.Thread(target=alert_scan_loop, args=(cfg, session), daemon=True).start() + if cfg.remote_enable: + threading.Thread(target=remote_loop, args=(cfg, session), daemon=True).start() + log.info("telemetry every %.1fs; map check every %.0fs; logs every %.0fs (Ctrl-C to stop)", + cfg.poll_interval, cfg.map_poll_interval, cfg.logs_interval) + while True: + try: + one_telemetry() + except Exception as e: + log.exception("telemetry 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/docs/X2_INTERFACE.md b/docs/X2_INTERFACE.md new file mode 100644 index 0000000..5da0daa --- /dev/null +++ b/docs/X2_INTERFACE.md @@ -0,0 +1,190 @@ +# AGIBOT X2 — what the robot actually exposes + +Everything here was read off the live robot (`10.255.254.84`) with +`tools/probe_x2.sh` and `ros2` introspection. It is the reference for *why* the +`.env` mapping looks the way it does, so nobody has to rediscover it. + +--- + +## 1. Platform + +| | | +|---|---| +| Board | NVIDIA Jetson Orin NX Engineering Reference Developer Kit Super V2Board | +| Arch / OS | aarch64 · Ubuntu 22.04.5 LTS · kernel 5.15.148-tegra | +| L4T | R36.4.3 | +| Python | 3.10.12 (`/usr/bin/python3`) | +| ROS | Humble, `ROS_DOMAIN_ID=0` | +| Docker | 27.5.1 installed, daemon **inactive + masked** (no passwordless sudo) | +| systemd user bus | available · `linger=no` (polkit denies enabling it unprivileged) | + +### Network interfaces + +| iface | address | note | +|---|---|---| +| `wifi0` | `10.255.254.84` | fleet-reachable → `MAC_INTERFACE=wifi0` | +| `develop0` | `10.0.1.41` | | +| `sensor0` | `10.11.1.1` | | +| `ssh0` | `10.0.200.41` | | + +### Serial number + +`X230028C5Z0058` — the AgiBot device serial, found in `/home/agi/agibot_report.py`. + +Prefer it over `/proc/device-tree/serial-number` (`1421126035442`) and over +`/etc/machine-id`, which are board identifiers rather than the robot's identity. + +--- + +## 2. The Control Dash — the practical state source + +A dashboard listens on **`:8770`** and serves a single JSON object at +**`/api/state`**. This is the cleanest source on the robot: battery is already +percent / volts / amps, no protobuf decoding, no ROS dependency. + +```json +{ + "mode": "STAND_DEFAULT", "mode_desc": "Stable stand", "mode_status": "Running", + "battery_pct": 68.0, "battery_voltage": 51.996, "battery_current": 3.544, + "battery_temp": 41.6, "battery_cycles": 14, "charging": true, + "pmu_temp": 40.17, "fan_rpm": 6960.0, "fan_pct": 68, + "rails": { "bus_48v": {...}, "output_48v": {...}, "output_12v": {...}, + "head_power": {...}, "orin": {...}, "rk3588": {...} }, + "imu": { "chest": { "roll":…, "pitch":…, "yaw":…, "accel_x":…, "gyro_x":… }, + "torso": { … } }, + "joints": { "head": [2], "waist": [3], "arm": [14], "leg": [12] }, + "hand_type": "None", "hand_state": { "left": [], "right": [] }, + "connection": { "online": true, "transport": "agent", "host": "127.0.0.1:8781", + "ros_domain_id": 0, "uptime_s": … } +} +``` + +Each `joints.[]` element is +`{name, position, velocity, effort, error}` — **an array of objects, not +numbers**. That is why the velocity mapping needs the wildcard form +`joints.leg[*].velocity`; a plain path would resolve to a list of dicts and +silently yield nothing. + +**Units confirmed on hardware:** `battery_pct` is already 0–100 (→ +`X2_SOC_SCALE=percent`), voltage is volts and current is amps (→ both scales +`1`), and a **positive** `battery_current` accompanies `charging: true` (→ +`X2_CURRENT_SIGN=1`). + +Other listening ports: `8781` (agent transport, no HTTP), `50080`, `21274/21275`, +`39101`, `11511` — none served a usable state document. + +--- + +## 3. ROS 2 topics + +The X2 runs AgiBot's **`aima`** stack. `ros2 topic list` shows ~70 topics; the +telemetry-relevant ones: + +| topic | type | use | +|---|---|---| +| `/aima/mc/leg_odometry` | `nav_msgs/msg/Odometry` | **position — in use** | +| `/slam/localization/odometry` | `nav_msgs/msg/Odometry` | not publishing | +| `/pnc/estimate_odom` | `nav_msgs/msg/Odometry` | not publishing | +| `/aima/hal/pmu/state` | `aimdk_msgs/msg/PmuState` | voltages + currents, **no SOC** | +| `/aima/battery_state/pb_3Aaimdk_2Eprotocol_2EBmsState` | `ros2_plugin_proto/msg/RosMsgWrapper` | protobuf-wrapped BMS — opaque | +| `/aima/hal/joint/{leg,arm,head,waist}/state` | `aimdk_msgs/msg/JointStateArray` | joint state | +| `/aima/hal/imu/{chest,torso}/state` | `sensor_msgs/msg/Imu` | | +| `/aima/sm/system_state` | `aimdk_msgs/msg/SmSystemState` | | + +### ⚠ QoS — the trap that costs an afternoon + +`/aima/mc/leg_odometry` publishes with: + +``` +Reliability: BEST_EFFORT +Durability: TRANSIENT_LOCAL +``` + +rclpy defaults subscribers to **`RELIABLE`**, which is **incompatible** with a +`BEST_EFFORT` publisher. The failure mode is silent: the subscription is +created, no exception is raised, no warning is logged — messages simply never +arrive and the field stays `null` forever. Meanwhile `ros2 topic echo` works +fine, so the topic looks healthy. + +A `BEST_EFFORT` subscriber can read from **either** publisher type, so the agent +defaults to it (`X2_ROS_QOS=best_effort`). Only set `reliable` if a specific +topic demands it. + +### `aimdk_msgs` is not in base ROS + +It lives in an overlay. To introspect these types: + +```bash +source /opt/ros/humble/setup.bash +source /home/agi/aimdk/install/local_setup.bash # ← required for aimdk_msgs +ros2 interface show aimdk_msgs/msg/PmuState +``` + +Note `ros2 interface show` fails for a few types (`JointNoRealTimeStateArray`, +`ros2_plugin_proto/msg/RosMsgWrapper`) — their `.idl` isn't installed. Read the +`.msg` files under +`/home/agi/aimdk/install/aimdk_msgs/share/aimdk_msgs/msg/` instead. + +Sourcing `setup.bash` under `set -u` **aborts** — it reads unbound variables +(`AMENT_TRACE_SETUP_FILES`, `COLCON_TRACE`). Always `set +u` first. This is why +the systemd unit and the probe script both do so. + +--- + +## 4. Why `motor_temp` is permanently `null` + +Searching every message definition in the overlay: + +```bash +grep -rl -i 'temp' /home/agi/aimdk/install/aimdk_msgs/share/aimdk_msgs/msg/*.msg +# -> PmuState.msg (only this one) +``` + +`PmuState` carries `battery_voltage`, six rail currents, and an over-temperature +*status bit* — it is the **power unit**, not the motors. The per-joint message +is: + +``` +aimdk_msgs/msg/JointState: + string name + float64 position + float64 velocity + float64 effort + uint16 error_code +``` + +No temperature field anywhere. The X2 does not publish per-motor temperatures, +so `motor_temp` reports `null`. Putting `pmu_temp` in that field would be +fabricating data. + +--- + +## 5. Alternative: full ROS 2 backend + +If the Control Dash is ever unavailable, the agent can read state over ROS 2 +instead (`X2_SOURCE=ros2`). Caveats: + +- Battery **SOC is not available over ROS** in plain form — `PmuState` has + voltage/current only, and the real BMS arrives protobuf-wrapped inside + `ros2_plugin_proto/msg/RosMsgWrapper`, which needs AgiBot's protobuf + definitions to decode. +- Custom types must be importable inside whatever runs the agent, i.e. the + `aimdk` overlay must be sourced. +- The same `X2_ROS_QOS=best_effort` rule applies to every subscription. + +For those reasons **http remains the recommended state source**, with ROS 2 used +only for position. + +--- + +## 6. AimRT + +`aimrt_py` **is** installed +(`/usr/local/lib/python3.10/dist-packages/aimrt_py/`), and AgiBot's open-source +X1 stack is built on AimRT — so it is very likely the underlying middleware here. + +The agent deliberately does **not** bind AimRT channels natively: the X2's +channel names and message definitions are not public, and guessing them yields +an agent that imports cleanly, builds, and heartbeats `battery: null` forever — +a failure only visible on the robot. AimRT ships a ROS 2 plugin, and the bridged +topics above are what the agent uses instead. diff --git a/tools/probe_x2.sh b/tools/probe_x2.sh new file mode 100644 index 0000000..8fcd3cb --- /dev/null +++ b/tools/probe_x2.sh @@ -0,0 +1,159 @@ +#!/usr/bin/env bash +# probe_x2.sh — read-only discovery for an AGIBOT X2 (or any unknown robot). +# +# tools/probe_x2.sh +# ./fleet_install.sh probe x2 --user # same, via the installer +# +# WHY THIS EXISTS +# The X2 exposes no fixed topic contract, so sanad_api_x2.py reads every topic +# name and field path from .env instead of hard-coding them. This script finds +# the real values to put there. +# +# It ONLY READS: no install, no write, no package manager, no motion. Safe to run +# on a live robot. Everything it prints maps to a variable in agents/x2/.env.example. +set -uo pipefail + +# No default login user is assumed for the X2 — its login account is a property +# of the robot, not something to guess. Pass it explicitly. +IP="${1:-}"; USER_="${2:-}" +[ -n "$IP" ] && [ -n "$USER_" ] \ + || { echo "usage: $0 " >&2 + echo " the X2's login user is not assumed — pass the account you SSH in as" >&2 + exit 1; } + +SSH_OPTS=(-o BatchMode=yes -o ConnectTimeout=8 -o StrictHostKeyChecking=accept-new) +[ -n "${FLEET_SSH_KEY:-}" ] && SSH_OPTS+=(-i "$FLEET_SSH_KEY") +ssh -n "${SSH_OPTS[@]}" "$USER_@$IP" true 2>/dev/null \ + || { echo "ERROR: cannot SSH to $USER_@$IP (key auth?)" >&2; exit 1; } + +echo "==============================================================" +echo " AGIBOT X2 probe — $USER_@$IP" +echo "==============================================================" + +# NOTE: no -n here. The probe body is piped in on stdin, and -n would redirect +# stdin from /dev/null, so `bash -s` would read an empty script and print nothing. +ssh "${SSH_OPTS[@]}" "$USER_@$IP" 'bash -s' <<'REMOTE' +set -uo pipefail +sec(){ echo; echo "── $* ──────────────────────────────────────────"; } +have(){ command -v "$1" >/dev/null 2>&1; } + +sec "1. PLATFORM -> software.* / firmware.* in telemetry" +. /etc/os-release 2>/dev/null && echo "os: ${PRETTY_NAME:-?} (VERSION_ID=${VERSION_ID:-?})" +echo "kernel: $(uname -r)" +echo "arch: $(uname -m) # arm64 => build natively on the robot" +for f in /sys/firmware/devicetree/base/model /proc/device-tree/model; do + [ -r "$f" ] && { echo "board: $(tr -d '\0' < "$f")"; break; } +done +[ -r /etc/nv_tegra_release ] && echo "l4t: $(head -1 /etc/nv_tegra_release)" +echo "python3: $(python3 -V 2>&1)" + +sec "2. DEPLOY PREREQS -> can fleet_install.sh work here?" +if have docker; then + echo "docker: $(docker --version 2>&1)" + docker ps >/dev/null 2>&1 \ + && echo " OK '$USER' can talk to dockerd (no sudo needed)" \ + || echo " !! '$USER' CANNOT use docker — add to the docker group" +else + echo "docker: !! NOT INSTALLED — required" +fi +systemctl --user show-environment >/dev/null 2>&1 \ + && echo "systemd: OK user bus available (user-level service, no sudo)" \ + || echo "systemd: !! no user bus — auto-start needs another mechanism" +echo "linger: $(loginctl show-user "$USER" 2>/dev/null | grep -i linger || echo '?')" + +sec "3. NETWORK -> MAC_INTERFACE, REMOTE_HOST" +ip -o -4 addr show 2>/dev/null | awk '{printf " %-10s %s\n", $2, $4}' +echo " MACs:" +for n in /sys/class/net/*; do + i="$(basename "$n")"; [ "$i" = lo ] && continue + echo " $(printf '%-10s' "$i") $(cat "$n/address" 2>/dev/null)" +done + +sec "4. SERIAL NUMBER -> SN (keys the robot on the fleet server)" +for f in /sys/class/dmi/id/product_serial /proc/device-tree/serial-number \ + /etc/machine-id /etc/robot_sn /etc/agibot/sn; do + [ -r "$f" ] && echo " $f = $(tr -d '\0' < "$f" 2>/dev/null | head -1)" +done +echo " (prefer the serial printed on the robot / in AgiBot's tooling over machine-id)" + +sec "5. ROS 2 -> X2_SOURCE=ros2, X2_TOPIC_*, ROS_DOMAIN_ID" +if [ -d /opt/ros ]; then + echo "distros: $(ls /opt/ros 2>/dev/null | tr '\n' ' ')" + D="${ROS_DISTRO:-$(ls /opt/ros 2>/dev/null | head -1)}" + echo "using: $D ROS_DOMAIN_ID=${ROS_DOMAIN_ID:-0}" + if [ -f "/opt/ros/$D/setup.bash" ]; then + # ROS setup.bash reads unbound vars (AMENT_TRACE_SETUP_FILES, COLCON_TRACE, + # …). Under `set -u` that is FATAL and kills this whole probe silently, so + # relax nounset across the source and restore it after. + set +u + # shellcheck disable=SC1090 + . "/opt/ros/$D/setup.bash" 2>/dev/null || true + echo "--- ros2 topic list (5s) ---" + timeout 5 ros2 topic list 2>/dev/null | sed 's/^/ /' || echo " (none / timed out)" + echo "--- types for likely telemetry topics ---" + for t in $(timeout 5 ros2 topic list 2>/dev/null | grep -Ei 'batt|power|bms|joint|motor|odom|imu|state|fsm|mode'); do + printf " %-42s %s\n" "$t" "$(timeout 3 ros2 topic info "$t" 2>/dev/null | awk -F': ' '/Type/{print $2; exit}')" + done + set -u + fi +else + echo " no /opt/ros — the ros2 backend is unavailable unless AgiBot ships one" +fi + +sec "6. AIMRT / VENDOR SDK -> X2_SOURCE=aimrt (or the ros2 bridge)" +python3 -c 'import aimrt_py, sys; print(" aimrt_py importable:", aimrt_py.__file__)' 2>/dev/null \ + || echo " aimrt_py not importable for this python3" +ls -d /opt/aimrt /opt/agibot /usr/local/aimrt ~/aimrt* ~/agibot* 2>/dev/null | sed 's/^/ path: /' +python3 -c 'import agibot, sys; print(" agibot module:", agibot.__file__)' 2>/dev/null || true + +sec "7. LISTENING PORTS -> X2_STATE_URL (http backend), REMOTE_PORTS" +(ss -lntp 2>/dev/null || netstat -lntp 2>/dev/null) | sed 's/^/ /' | head -30 +echo " probe a candidate: curl -s http://127.0.0.1:/ | head -c 400" + +sec "8. RUNNING WORKLOAD -> PROJECT_LOG_CONTAINER, what publishes state" +have docker && docker ps --format ' {{.Names}} {{.Status}} ({{.Image}})' 2>/dev/null +echo " top processes:" +ps -eo comm,pcpu --sort=-pcpu 2>/dev/null | head -12 | sed 's/^/ /' + +sec "9. MAPS -> MAPS_DIR / EXTRA_MAP_DIRS" +find "$HOME" -maxdepth 5 \( -name '*.pgm' -o -name 'map*.yaml' -o -name '*.db' -o -name '*.posegraph' \) \ + 2>/dev/null | head -20 | sed 's/^/ /' +echo " (the agent wants the DIRECTORY holding a matching .pgm + .yaml pair, or a .db)" + +sec "10. DISK -> storage.* in telemetry" +df -h / 2>/dev/null | sed 's/^/ /' +REMOTE + +cat <<'NEXT' + +============================================================== + NEXT STEPS +============================================================== + 1. Pick a backend from what appeared above: + section 5 found battery/joint/odom topics -> X2_SOURCE=ros2 + section 7 found a vendor JSON API -> X2_SOURCE=http + neither -> X2_SOURCE=none + (identity + maps + logs + still work; battery null) + + 2. Install with the real serial from section 4: + + ./fleet_install.sh install x2 --sn \ + --token --server-url https://eco.yslootahrobotics.com + + ros2 backend? build against a ROS base matching the robot's distro: + X2_BASE_IMAGE=ros:humble-ros-base ./fleet_install.sh install x2 ... + + 3. Put the discovered names in the robot's .env and restart: + + ssh @ 'nano ~/sanad_api_x2/.env' # X2_TOPIC_* / X2_FIELD_* + ssh @ 'systemctl --user restart sanad-api-x2' + + 4. Confirm what it is actually sending: + + ./fleet_install.sh data x2 + + 5. VERIFY THE TWO UNIT CONVENTIONS on real hardware: + - battery % looks right -> else set X2_SOC_SCALE=fraction|percent + - "charging" true while charging -> else set X2_CURRENT_SIGN=-1 +NEXT