engineai_fleet/agent/AGENT_README.md

sanad_api_eng.py — internals

One process, five loops, one payload schema. Read docs/PM01_INTERFACE.md first for what the robot exposes; this file is about how the agent is put together.

---

Loops

loop	cadence	what
telemetry	POLL_INTERVAL (2 s)	build payload → POST /ingest/telemetry → rising-edge alerts
map	MAP_POLL_INTERVAL (30 s)	scan MAPS_DIR → upload new content once per server
logs	LOGS_INTERVAL (60 s)	drain the agent's log ring + project-log tail → POST /{sn}/logs
alert scan	ALERT_SCAN_INTERVAL (10 s)	regex the project log → POST /{sn}/alert
remote	REMOTE_INTERVAL (60 s)	discover the Sanad dashboard → POST /{sn}/remote

All four background loops are daemon threads; the telemetry loop is the main thread. Any loop raising is caught and logged — one failing subsystem never stops telemetry.

---

The seam: EngineAiSource.snapshot()

The only robot-specific class. Whatever the backend, it fills one dict, and that dict is the entire interface between "this robot" and the shared telemetry / fault / status pipeline:

{ "bms": {soc, current_a, voltage_v, temp_c, soh, cycles, current_limit_a} | None,
  "state_age": float | None,        # seconds since ANY topic last updated
  "temps": [float],                 # motor_temperature[]
  "mos_temps": [float],             # mos_temperature[]
  "motor_faults": ["joint7=0x2"],   # error_code[] != 0
  "motor_offline": [int],           # offline[] != 0
  "power_err": int | None,          # PowerInfo.error_code
  "power_enabled": bool | None,     # PowerInfo.enable
  "max_vel": float,                 # max |JointState.velocity|
  "xy": {x, y} | None,
  "motion": str | None,             # MotionState.current_motion_task
  "transitions": [str],             # available_transition_motions
  "fw": {} }

Porting to another EngineAI model means changing .env topic names, not this class. Porting to a different vendor means writing one new class with the same snapshot() contract.

Every ingest method is wrapped in try/except: pass. A wrong field mapping degrades that one field to null; it never kills the subscription or the loop.

---

_dig(obj, "a.b[0].c")

Walks a dotted path over both ROS message objects (getattr) and plain dicts (.get) — so the same ENG_FIELD_* mapping works for the ros2 and http backends. Supports list indices (cell_temp[0]) and a wildcard (joints[*].temp) that collects a field from every element and flattens.

Returns None for any missing link. That is the whole reason a mistyped .env path shows up as one null field on the dashboard instead of a crash loop.

---

Multi-server fan-out

Config.endpoints is a list of Endpoint(name, url, token, enabled). Every ingest call goes through _post_each(), which POSTs to each enabled endpoint and returns {server_name: {ok, code, error}}.

Design points that are load-bearing:

• Per-endpoint tokens. eco and eco-dev maintain independent token stores; a token minted on one 401s on the other. A single shared DEVICE_TOKEN would silently fail on whichever server did not issue it.
• Independent failure. One server being down, slow or unauthorised cannot block or fail the others — each POST is its own try/except.
• Bytes, not file handles, for multipart. A file handle streams once; with two servers the second upload would send an empty body. Map blobs are read into memory (capped by MAP_MAX_UPLOAD_MB) and posted to each.
• Per-server map state. uploaded.json is {server: {path: fingerprint}}. With one shared key, enabling a second server later would find every map already "uploaded" and that server would never receive them. The old flat layout is migrated on read.
• Throttled error reporting. A persistently failing server would emit an ERROR every 2 s forever, burying the log and filling the ring that gets shipped to /{sn}/logs. Each (what, server, status) signature reports in full at most once per ERROR_LOG_COOLDOWN (60 s); suppressed occurrences are counted and shown on the next line that prints ([+3 more since last report]), and the counter resets on success, so nothing is hidden — only de-duplicated.

---

Status and faults

derive_status() returns the same vocabulary the X2 reports — charging | moving | idle | offline — so one dashboard renders every robot:

no state at all + no battery   → offline
current_a > 0.05               → charging
max |joint velocity| > 0.15    → moving          (MOVING_VEL)
otherwise                      → idle

The real motion mode (pd_sitdown, rl_basic, …) goes in control.mode rather than being squeezed into status, because it is a much larger vocabulary than the dashboard's four states.

derive_faults() emits strings, not objects — the fleet ingest 500s on fault objects. Codes: LOW_BATTERY, MOTOR_OVERTEMP, MOS_OVERTEMP, POWER_FAULT, POWER_DISABLED, MOTOR_FAULT, MOTOR_OFFLINE, COMMS_STALE.

Alerts dedup on the code before the first :, never the whole string: every fault embeds a live number (battery 21%, no robot state for 12s) that changes almost every tick, so string-dedup re-fires the same alert every POLL_INTERVAL. A code alerts on its rising edge, then at most once per ALERT_LOG_COOLDOWN while it persists; clearing it makes the next occurrence a rising edge again.

---

Rate decimation — measured, not assumed

/hardware/joint_state publishes at 500 Hz and /hardware/motor_debug at 100 Hz. Telemetry resamples every 2 s, so nearly all of that is thrown away.

Measured on the robot, as % of one core (the Jetson has 12), by starting the agent with subscriptions removed:

configuration	CPU	implies
all four topics	26.5%	—
without joint_state	6.5%	joint_state ≈ 20%
without joint_state + motor_debug	2.2%	motor_debug ≈ 4.3%
		everything else ≈ 2.2%

So the 500 Hz stream was ~75% of the agent's cost, for one number.

Two mitigations, in order of how much they actually helped:

1. ENG_JOINT_MIN_PERIOD / ENG_MOTOR_MIN_PERIOD gate the callbacks to 20 Hz / 5 Hz. The gate is the first statement in every high-rate callback.
2. ENG_RAW_SUBSCRIBE=1 (default) subscribes those two topics with raw=True, so the gate runs before deserialization and a dropped message is never turned into a Python object. Measured 27.2% → 23.8%.

That second number is the interesting one: it is a real win but far smaller than the hypothesis predicted, which means most of the remaining cost is CycloneDDS delivering 500 msg/s and waking the executor — not building the message object. The docstring records this so nobody re-derives it.

Why it is not optimised further. The remaining ~20% would disappear if the subscription were created and destroyed around each sample (a ~12% duty cycle). That would emit DDS endpoint-discovery traffic every 2 s onto eth1 — the network this robot's motion controller lives on (peer 192.168.0.163). Trading ~2% of an application core for periodic discovery churn on a realtime control network is the wrong trade. The subscription stays stable.

If CPU ever matters more than the moving status, ENG_TOPIC_JOINTS= (empty) drops it entirely and the agent costs ~6% of one core. status then reports idle instead of moving, and control.mode still shows the real motion task.

---

Temperature filtering

_floats(v, 0, 200) keeps only readings in 0 < t <= 200 °C. A 0.0 in motor_temperature[] means "slot not reporting", not "0 degrees": averaging it in would drag motor_temp.avg down and mask a genuinely hot joint. count in the payload is the number of valid readings, so a shrinking count is itself a signal.

---

Exit path

rclpy's CycloneDDS C++ threads are still running when Python finalises, so the process dies with SIGABRT — after completing its work correctly. systemd records that as a failed exit, which would mask a genuine crash.

_exit(code) therefore: shuts rclpy down, sleeps 200 ms so spin() unblocks, flushes the log handlers, then os._exit() to skip the static destructors that abort. SIGINT/SIGTERM handlers route into it, so systemctl restart records Deactivated successfully instead of Failed with result 'signal'.

---

CLI

--dry-run    build every payload and print it; never POST   (safe on a live feed)
--once       one map pass + one real telemetry post, exit
--map-only   upload discovered maps once; no state source, no telemetry
--simulate   synthetic robot state (the map scan stays real)
--force      re-upload maps even if unchanged
--list       list discovered maps and exit
--interval   override POLL_INTERVAL
-v           debug logging (includes per-request urllib3 lines)

--dry-run is the right first move after any .env change: it exercises the full read path and prints the exact JSON without touching the server.