2679 lines
111 KiB
Python
2679 lines
111 KiB
Python
#!/usr/bin/env python3
|
|
"""sanad_api_eng — EngineAI PM01 fleet agent: TELEMETRY + MAP sync in ONE service.
|
|
|
|
Same concept, same payload schema and same endpoints as the AGIBOT X2 agent
|
|
(agi_fleet/agent/sanad_api_x2.py) — retargeted to the EngineAI PM01 ("T800")
|
|
and extended to post the SAME data to MORE THAN ONE fleet server.
|
|
|
|
1. TELEMETRY — every ~2 s POSTs the robot's live status to EVERY enabled
|
|
fleet server:
|
|
POST {SERVER_URL}/api/v1/fleet/ingest/telemetry
|
|
{ sn, name, mac, brand, type, model, software, firmware, battery,
|
|
charging, battery_detail, motor_temp, storage, status, position,
|
|
control, faults, map, logs, project_logs, remote, alerts, time,
|
|
started_at, last_start, uptime_s, ts }
|
|
|
|
2. MAP — a background loop (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 PER SERVER (re-upload only if its content changes):
|
|
POST {SERVER_URL}/api/v1/fleet/ingest/{sn}/map
|
|
|
|
The map result is SHOWN inside every telemetry post as the "map" field, so
|
|
the server always sees whether the map made it — and why not.
|
|
|
|
3. LOGS / ALERTS / REMOTE — /{sn}/logs, /{sn}/alert, /{sn}/remote, all
|
|
fanned out to every enabled server exactly like telemetry.
|
|
|
|
MULTI-SERVER (the one structural difference from the X2 agent)
|
|
---------------------------------------------------------------
|
|
The X2 agent posts to a single SERVER_URL. This one keeps a LIST of endpoints,
|
|
each with its OWN token, and sends the identical payload to each:
|
|
|
|
SERVER_URL / DEVICE_TOKEN (primary)
|
|
SERVER_2_URL / SERVER_2_TOKEN / SERVER_2_ENABLE
|
|
SERVER_3_URL / SERVER_3_TOKEN / SERVER_3_ENABLE
|
|
|
|
A server that is unreachable, unauthorised or slow NEVER blocks the others:
|
|
each POST is independent and its result is reported per-server in the log line
|
|
and in the telemetry status blocks. Turning a second server on is an .env edit
|
|
plus a restart — never a code change.
|
|
|
|
DATA SOURCES (EngineAI PM01 — pluggable, configured in .env)
|
|
-------------------------------------------------------------
|
|
backend : ENG_SOURCE = auto | ros2 | http | none
|
|
battery / charging : /hardware/power_info interface_protocol/msg/PowerInfo
|
|
motor temperatures : /hardware/motor_debug interface_protocol/msg/MotorDebug
|
|
locomotion (moving): /hardware/joint_state interface_protocol/msg/JointState
|
|
motion mode : /motion/motion_state interface_protocol/msg/MotionState
|
|
position {x,y} : ENG_POSITION_SOURCE (none|ros2|rosbridge|http) — the PM01
|
|
publishes NO odometry, so this is null until a nav stack
|
|
is running; see docs/PM01_INTERFACE.md
|
|
storage / OS / MAC : read directly from the host
|
|
maps : MAPS_DIR/**.pgm+yaml and *.db, places under
|
|
DATA_DIR/<robot>/places/<map>.json
|
|
|
|
Nothing above is hard-coded: run tools/probe_eng.sh on the robot to discover the
|
|
real topics/fields, then set them in .env.
|
|
|
|
Read-only toward the robot: it subscribes to four topics and reads an HTTP
|
|
status page. It NEVER publishes, and never commands motion.
|
|
|
|
CLI: --simulate | --once | --dry-run | --force (map re-upload) | --list | -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_eng")
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# env helpers
|
|
# --------------------------------------------------------------------------- #
|
|
def _load_dotenv(path: str = "") -> None:
|
|
"""Load KEY=VALUE lines from .env (next to this file unless overridden)."""
|
|
p = Path(path) if path else Path(__file__).resolve().parent / ".env"
|
|
if not p.exists():
|
|
p = Path(".env")
|
|
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 _env_float(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)
|
|
|
|
|
|
def _now_str() -> str:
|
|
"""Full local date+time with UTC offset (e.g. 2026-08-27 19:20:33+04:00).
|
|
TZ_OFFSET_HOURS (default +4, Dubai) keeps clocks honest without tzdata."""
|
|
off = _env_float("TZ_OFFSET_HOURS", "4")
|
|
tz = _dt.timezone(_dt.timedelta(hours=off))
|
|
return _dt.datetime.now(tz).isoformat(sep=" ", timespec="seconds")
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# fleet endpoints — the SAME payload goes to every enabled server
|
|
# --------------------------------------------------------------------------- #
|
|
def _host_label(url: str) -> str:
|
|
"""'https://eco-dev.yslootahrobotics.com' -> 'eco-dev' (for log lines)."""
|
|
host = url.split("://", 1)[-1].split("/", 1)[0]
|
|
return host.split(".", 1)[0] or host
|
|
|
|
|
|
@dataclass
|
|
class Endpoint:
|
|
"""One fleet server: its base URL and its OWN device token.
|
|
|
|
Tokens are per-server on purpose. eco and eco-dev maintain independent
|
|
token stores, so a token minted on one is rejected (401) by the other —
|
|
sharing a single DEVICE_TOKEN across both silently 401s forever on
|
|
whichever server did not issue it."""
|
|
name: str
|
|
url: str
|
|
token: str
|
|
enabled: bool = True
|
|
|
|
def headers(self) -> Dict[str, str]:
|
|
return {"Authorization": f"Bearer {self.token}"}
|
|
|
|
def at(self, tmpl: str, sn: str = "") -> str:
|
|
return self.url + (tmpl.format(sn=sn) if "{sn}" in tmpl else tmpl)
|
|
|
|
|
|
def _endpoints_from_env() -> List[Endpoint]:
|
|
"""SERVER_URL/DEVICE_TOKEN plus SERVER_<N>_URL/SERVER_<N>_TOKEN for N=2..5.
|
|
|
|
A slot with no URL is skipped entirely. A slot with a URL but no token is
|
|
kept but DISABLED with a loud warning — that is the "server configured,
|
|
token not issued yet" state, and it must be visible rather than silent."""
|
|
out: List[Endpoint] = []
|
|
primary_url = _env("SERVER_URL").rstrip("/")
|
|
primary_tok = _env("DEVICE_TOKEN")
|
|
if primary_url:
|
|
out.append(Endpoint(name=_env("SERVER_NAME", "") or _host_label(primary_url),
|
|
url=primary_url, token=primary_tok,
|
|
enabled=bool(primary_tok) and _env_bool("SERVER_ENABLE", True)))
|
|
for n in range(2, 6):
|
|
url = _env(f"SERVER_{n}_URL").rstrip("/")
|
|
if not url:
|
|
continue
|
|
tok = _env(f"SERVER_{n}_TOKEN")
|
|
if not tok and _env_bool(f"SERVER_{n}_SHARE_TOKEN", False):
|
|
tok = primary_tok
|
|
out.append(Endpoint(name=_env(f"SERVER_{n}_NAME", "") or _host_label(url),
|
|
url=url, token=tok,
|
|
enabled=bool(tok) and _env_bool(f"SERVER_{n}_ENABLE", True)))
|
|
return out
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# config
|
|
# --------------------------------------------------------------------------- #
|
|
@dataclass
|
|
class Config:
|
|
endpoints: List[Endpoint]
|
|
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
|
|
position_source: str
|
|
rosbridge_url: str
|
|
low_soc: int
|
|
motor_temp_max: float
|
|
mos_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]
|
|
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
|
|
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 (the robot's Sanad app container)
|
|
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":
|
|
eps = _endpoints_from_env()
|
|
if not eps:
|
|
raise SystemExit("[config] missing required env: SERVER_URL (and DEVICE_TOKEN)")
|
|
for e in eps:
|
|
if not e.token:
|
|
log.warning("server %r (%s) has NO token — DISABLED. Set its "
|
|
"SERVER_*_TOKEN in .env and restart to enable it.", e.name, e.url)
|
|
elif not e.enabled:
|
|
log.warning("server %r (%s) is configured but switched OFF "
|
|
"(SERVER_*_ENABLE=0)", e.name, e.url)
|
|
if not any(e.enabled for e in eps):
|
|
raise SystemExit("[config] no fleet server is enabled — set DEVICE_TOKEN "
|
|
"(and/or SERVER_<N>_TOKEN) in .env")
|
|
iface = _env("ENG_INTERFACE", "eth0")
|
|
data_dir = _env("DATA_DIR")
|
|
legacy = _env("LEGACY_PLACES")
|
|
return cls(
|
|
endpoints=eps,
|
|
sn=_env("SN", "pm01_0000"),
|
|
name=_env("ROBOT_NAME", "") or _env("SN", "pm01_0000"),
|
|
brand=_env("ROBOT_BRAND", "engineai"),
|
|
robot_type=_env("ROBOT_TYPE", "humanoid"),
|
|
model=_env("ROBOT_MODEL", "pm01"),
|
|
storage_path=_env("STORAGE_PATH", ""),
|
|
data_path=_env("STORAGE_DATA_PATH", ""),
|
|
net_interface=iface,
|
|
domain_id=int(_env("ROS_DOMAIN_ID", "69") or "69"),
|
|
mac_interface=_env("MAC_INTERFACE", iface),
|
|
position_source=_env("ENG_POSITION_SOURCE", "none").lower(),
|
|
rosbridge_url=_env("ROSBRIDGE_URL", "ws://127.0.0.1:9090"),
|
|
low_soc=int(_env("LOW_SOC", "50") or "50"),
|
|
motor_temp_max=_env_float("MOTOR_TEMP_MAX", "85"),
|
|
mos_temp_max=_env_float("MOS_TEMP_MAX", "100"),
|
|
# log-driven alerts: "CODE=regex" entries separated by ";;" (regex may
|
|
# contain '|'). NOTE: matching is CASE-SENSITIVE (log levels are
|
|
# uppercase); use an inline (?i) prefix for case-insensitive text.
|
|
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=_env_float("ALERT_LOG_COOLDOWN", "300"),
|
|
alert_scan_interval=_env_float("ALERT_SCAN_INTERVAL", "10"),
|
|
alert_backfill_bytes=int(_env("ALERT_BACKFILL_BYTES", str(8 * 1024 * 1024))),
|
|
poll_interval=_env_float("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")),
|
|
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=_env_float("MAP_POLL_INTERVAL", "30"),
|
|
map_max_upload_mb=_env_float("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=_env_float("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", "8014,8001,9002,9003,8000,8080"),
|
|
remote_url=_env("REMOTE_URL", ""),
|
|
remote_interval=_env_float("REMOTE_INTERVAL", "60"),
|
|
ssh_enable=_env_bool("SSH_REGISTER", True),
|
|
ssh_user=_env("SSH_USER", ""),
|
|
ssh_port=int(_env("SSH_PORT", "22") or "22"),
|
|
control_url=_env("CONTROL_STATUS_URL", ""),
|
|
control_enable=_env_bool("CONTROL_ENABLE", False),
|
|
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") or "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=_env_float("HTTP_TIMEOUT", "30"),
|
|
)
|
|
|
|
def live(self) -> List[Endpoint]:
|
|
return [e for e in self.endpoints if e.enabled]
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# multi-server POST fan-out
|
|
#
|
|
# Every ingest call goes to EVERY enabled server. One server failing (down,
|
|
# 401, slow) must never stop the others or raise: each POST is independent and
|
|
# its outcome is recorded per server name.
|
|
# --------------------------------------------------------------------------- #
|
|
def _brief(text: str, limit: int = 160) -> str:
|
|
"""One-line summary of a server error body.
|
|
|
|
Error pages are HTML: dumping them raw breaks the log into a dozen lines and
|
|
those lines are then shipped to /{sn}/logs, so a single 401 becomes a dozen
|
|
junk entries per report. Collapse to one line and strip tags."""
|
|
import re
|
|
t = re.sub(r"<[^>]+>", " ", text or "")
|
|
t = re.sub(r"\s+", " ", t).strip()
|
|
return t[:limit]
|
|
|
|
|
|
# Repeated-failure throttle. A server that is down or holding a bad token would
|
|
# otherwise emit one ERROR every POLL_INTERVAL (every 2 s) forever, burying the
|
|
# rest of the log and filling the shipped-log ring. Each (what, server, code)
|
|
# signature is reported in full at most once per ERROR_LOG_COOLDOWN; the
|
|
# suppressed occurrences are COUNTED and reported on the next line that does
|
|
# print, so nothing is hidden — only de-duplicated.
|
|
_ERR_SEEN: Dict[str, List[float]] = {} # sig -> [last_logged_ts, suppressed]
|
|
_ERR_LOCK = threading.Lock()
|
|
|
|
|
|
def _should_log_error(sig: str, cooldown: float) -> Optional[int]:
|
|
"""None = suppress. Otherwise the number of occurrences suppressed since the
|
|
last time this signature was logged (0 on a first/rising-edge report)."""
|
|
now = time.monotonic()
|
|
with _ERR_LOCK:
|
|
ent = _ERR_SEEN.get(sig)
|
|
if ent is None or (now - ent[0]) > cooldown:
|
|
suppressed = int(ent[1]) if ent else 0
|
|
_ERR_SEEN[sig] = [now, 0.0]
|
|
return suppressed
|
|
ent[1] += 1
|
|
return None
|
|
|
|
|
|
def _clear_error(sig: str) -> None:
|
|
"""Forget a signature once it succeeds, so its next failure logs immediately."""
|
|
with _ERR_LOCK:
|
|
_ERR_SEEN.pop(sig, None)
|
|
|
|
|
|
def _post_each(cfg: Config, session: requests.Session, tmpl: str, *,
|
|
json_body: Optional[Dict[str, Any]] = None,
|
|
body_bytes: Optional[bytes] = None,
|
|
filename: str = "", form: Optional[Dict[str, str]] = None,
|
|
what: str = "post", quiet: bool = False) -> Dict[str, Any]:
|
|
"""POST to every enabled endpoint. Returns {server_name: {ok, code, error}}.
|
|
|
|
body_bytes+filename selects multipart upload (map .db); json_body selects a
|
|
JSON body. The bytes are held in memory rather than a file handle because a
|
|
handle can only be streamed ONCE — with several servers the second upload
|
|
would send an empty body.
|
|
|
|
quiet=True downgrades failures to DEBUG (used by the log-shipping path,
|
|
whose own ERROR lines would otherwise feed back into the ring it ships).
|
|
Every other caller reports failures, throttled per signature.
|
|
"""
|
|
results: Dict[str, Any] = {}
|
|
cooldown = _env_float("ERROR_LOG_COOLDOWN", "60")
|
|
for e in cfg.live():
|
|
url = e.at(tmpl, cfg.sn)
|
|
try:
|
|
if body_bytes is not None:
|
|
resp = session.post(url,
|
|
files={"db": (filename, body_bytes, "application/octet-stream")},
|
|
data=form or {}, headers=e.headers(),
|
|
timeout=cfg.http_timeout, verify=cfg.verify_tls)
|
|
else:
|
|
resp = session.post(url, json=json_body, headers=e.headers(),
|
|
timeout=cfg.http_timeout, verify=cfg.verify_tls)
|
|
except requests.RequestException as ex:
|
|
results[e.name] = {"ok": False, "code": None, "error": _brief(str(ex), 200)}
|
|
sig = f"{what}|{e.name}|transport"
|
|
if quiet:
|
|
log.debug("%s -> %s FAILED (transport): %s", what, e.name, ex)
|
|
else:
|
|
n = _should_log_error(sig, cooldown)
|
|
if n is not None:
|
|
log.error("%s -> %s FAILED (transport): %s%s", what, e.name, ex,
|
|
f" [+{n} more since last report]" if n else "")
|
|
continue
|
|
ok = bool(resp.ok)
|
|
results[e.name] = {"ok": ok, "code": resp.status_code,
|
|
"error": None if ok else _brief(resp.text, 200)}
|
|
sig = f"{what}|{e.name}|{resp.status_code}"
|
|
if ok:
|
|
_clear_error(sig)
|
|
continue
|
|
if quiet:
|
|
log.debug("%s -> %s FAILED: HTTP %s", what, e.name, resp.status_code)
|
|
continue
|
|
n = _should_log_error(sig, cooldown)
|
|
if n is not None:
|
|
hint = ""
|
|
if resp.status_code in (401, 403):
|
|
hint = (f" <- token rejected by {e.name}; each fleet server issues its "
|
|
f"OWN token, so a token minted elsewhere will never work here "
|
|
f"(set the token for {e.url} in .env)")
|
|
elif resp.status_code == 413:
|
|
hint = " <- body too large for the server's upload cap"
|
|
log.error("%s -> %s FAILED: HTTP %s %s%s%s", what, e.name,
|
|
resp.status_code, _brief(resp.text), hint,
|
|
f" [+{n} more since last report]" if n else "")
|
|
return results
|
|
|
|
|
|
def _fmt_results(results: Dict[str, Any]) -> str:
|
|
"""'eco-dev=200 eco=401' — one compact token per server for the log line."""
|
|
return " ".join(f"{n}={(r.get('code') if r.get('code') is not None else 'ERR')}"
|
|
for n, r in results.items()) or "(no server enabled)"
|
|
|
|
|
|
def _any_ok(results: Dict[str, Any]) -> bool:
|
|
return any(r.get("ok") for r in results.values())
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# mac + storage + software/firmware cards
|
|
# --------------------------------------------------------------------------- #
|
|
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.08.27"
|
|
|
|
_software_cache: Optional[Dict[str, Any]] = None
|
|
|
|
|
|
def read_software(cfg: Config) -> Dict[str, Any]:
|
|
"""Robot software/OS card: ROS distro, host OS, kernel, arch, python."""
|
|
global _software_cache
|
|
if _software_cache is not None:
|
|
return _software_cache
|
|
sw: Dict[str, Any] = {}
|
|
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
|
|
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
|
|
sw["os_version"] = os_ver
|
|
# 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
|
|
sw["arch"] = u.machine
|
|
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): compute board model + Jetson L4T/BSP
|
|
release + kernel + the EngineAI product tag. The state source adds live fw."""
|
|
global _firmware_cache
|
|
if _firmware_cache is not None:
|
|
return dict(_firmware_cache)
|
|
fw: Dict[str, Any] = {}
|
|
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: "# R36 (release), REVISION: 4.3, ..." -> "R36.4.3"
|
|
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
|
|
# board serial (Jetson module) — stable hardware id, useful next to SN
|
|
for p in ("/proc/device-tree/serial-number", "/host/proc/device-tree/serial-number"):
|
|
try:
|
|
s = Path(p).read_bytes().decode().strip("\x00 \n")
|
|
if s:
|
|
fw["board_serial"] = s
|
|
break
|
|
except Exception:
|
|
continue
|
|
# EngineAI product tag, e.g. "export PRODUCT=t800"
|
|
for p in ("/app/applications/install/bringup/product.env",
|
|
"/host/app/applications/install/bringup/product.env"):
|
|
try:
|
|
for line in Path(p).read_text().splitlines():
|
|
if "PRODUCT" in line and "=" in line:
|
|
fw["product"] = line.split("=", 1)[1].strip().strip('"')
|
|
break
|
|
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 app data-dir size."""
|
|
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 ROS 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
|
|
|
|
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:
|
|
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):
|
|
""""interface_protocol/msg/PowerInfo" -> 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 subscription is created, no error is raised, and the field
|
|
just stays null forever.
|
|
|
|
Override with ENG_ROS_QOS=reliable if a topic ever requires it."""
|
|
from rclpy.qos import HistoryPolicy, QoSProfile, ReliabilityPolicy
|
|
want = _env("ENG_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 _ros_shutdown() -> None:
|
|
"""Stop rclpy before the interpreter tears down.
|
|
|
|
Without this the CycloneDDS C++ worker threads are still running when Python
|
|
finalises, and the process dies with SIGABRT ("terminate called without an
|
|
active exception") AFTER doing all its work correctly. systemd would record
|
|
that as a failed exit and it would mask a genuine crash, so the exit path is
|
|
made explicit rather than left to garbage collection."""
|
|
try:
|
|
import rclpy
|
|
if rclpy.ok():
|
|
rclpy.shutdown()
|
|
time.sleep(0.2) # let spin() unblock and the DDS threads wind down
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
def _exit(code: int) -> int:
|
|
"""Flush logs, stop ROS, and leave with a clean status.
|
|
|
|
os._exit skips the C++ static destructors that abort the process even after
|
|
a successful rclpy.shutdown(); the flush above it means nothing is lost."""
|
|
_ros_shutdown()
|
|
try:
|
|
logging.shutdown()
|
|
sys.stdout.flush()
|
|
sys.stderr.flush()
|
|
except Exception:
|
|
pass
|
|
os._exit(code)
|
|
|
|
|
|
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]
|
|
|
|
|
|
def _floats(v: Any, lo: float, hi: float) -> List[float]:
|
|
"""Numeric members of v inside (lo, hi]. Out-of-range slots (0 = 'not
|
|
reporting' on this hardware) are dropped, not counted as real readings."""
|
|
out: List[float] = []
|
|
for x in _seq(v):
|
|
try:
|
|
f = float(x)
|
|
except (TypeError, ValueError):
|
|
continue
|
|
if lo < f <= hi:
|
|
out.append(f)
|
|
return out
|
|
|
|
|
|
class EngineAiSource:
|
|
"""EngineAI PM01 state source — the ONE robot-specific class in this agent.
|
|
|
|
Backends:
|
|
ENG_SOURCE=auto (default) ros2 if rclpy imports, else http if
|
|
ENG_STATE_URL is set, else none (heartbeat mode)
|
|
ENG_SOURCE=ros2 subscribe ENG_TOPIC_* (types from ENG_TYPE_*)
|
|
ENG_SOURCE=http poll ENG_STATE_URL (one JSON object); same ENG_FIELD_*
|
|
ENG_SOURCE=none never read; always heartbeat
|
|
|
|
Whatever the backend, it fills one snapshot —
|
|
{bms, state_age, temps, mos_temps, max_vel, xy, motion, transitions,
|
|
motor_faults, power_err, power_enabled, fw} — which is the entire seam
|
|
between this robot and the shared telemetry / fault / status pipeline.
|
|
|
|
Every read is wrapped: a wrong mapping yields null fields and heartbeat
|
|
mode, never a crashed loop.
|
|
|
|
THROTTLING matters here. /hardware/joint_state publishes at 500 Hz and
|
|
/hardware/motor_debug at 100 Hz. Running a Python callback on every message
|
|
would burn CPU on the robot's own compute for data we resample every 2 s, so
|
|
each stream is decimated to ENG_*_MIN_PERIOD. The agent must not tax the
|
|
machine it is monitoring."""
|
|
|
|
# Defaults match the PM01's REAL interface_protocol messages.
|
|
_DEFAULTS = {
|
|
"soc": "percentage", # PowerInfo.percentage (0..100)
|
|
"voltage": "voltage", # PowerInfo.voltage (V)
|
|
"current": "current", # PowerInfo.current (A)
|
|
"current_limit": "current_limit",
|
|
"power_err": "error_code", # PowerInfo.error_code
|
|
"power_enable": "enable", # PowerInfo.enable
|
|
"temp": "", # PowerInfo carries NO pack temp
|
|
"soh": "", # not published
|
|
"cycles": "", # not published
|
|
"temps": "motor_temperature", # MotorDebug.motor_temperature[]
|
|
"mos_temps": "mos_temperature", # MotorDebug.mos_temperature[]
|
|
"motor_err": "error_code", # MotorDebug.error_code[]
|
|
"motor_offline": "offline", # MotorDebug.offline[]
|
|
"vel": "velocity", # JointState.velocity[]
|
|
"motion": "current_motion_task", # MotionState.current_motion_task
|
|
"transitions": "available_transition_motions",
|
|
"x": "pose.pose.position.x", # nav_msgs/Odometry (if one appears)
|
|
"y": "pose.pose.position.y",
|
|
"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._mos_temps: List[float] = []
|
|
self._motor_faults: List[str] = []
|
|
self._motor_offline: List[int] = []
|
|
self._power_err: Optional[int] = None
|
|
self._power_enabled: Optional[bool] = None
|
|
self._max_vel = 0.0
|
|
self._xy: Optional[Dict[str, float]] = None
|
|
self._motion: Optional[str] = None
|
|
self._transitions: List[str] = []
|
|
self._fw: Dict[str, Any] = {}
|
|
self._stop = False
|
|
self.backend = "none"
|
|
self.ok = False
|
|
self._map = {k: _env("ENG_FIELD_" + k.upper(), d) for k, d in self._DEFAULTS.items()}
|
|
|
|
# +1 = positive current means CHARGING (the ROS BatteryState convention).
|
|
# The PM01 reports the OPPOSITE: measured on the robot, PowerInfo.current
|
|
# stays positive (~2 A) while the pack drains (27% -> 26%, 54.74 -> 54.58 V),
|
|
# so the shipped default is -1 and "charging" is not inverted.
|
|
self._cur_sign = _env_float("ENG_CURRENT_SIGN", "-1")
|
|
# The telemetry schema is VOLTS and AMPS; PowerInfo already uses both.
|
|
self._v_scale = _env_float("ENG_VOLTAGE_SCALE", "1")
|
|
self._i_scale = _env_float("ENG_CURRENT_SCALE", "1")
|
|
self._soc_scale = _env("ENG_SOC_SCALE", "percent").lower()
|
|
# decimation (see class docstring) — seconds between processed messages
|
|
self._joint_period = _env_float("ENG_JOINT_MIN_PERIOD", "0.05") # 500Hz -> 20Hz
|
|
self._motor_period = _env_float("ENG_MOTOR_MIN_PERIOD", "0.2") # 100Hz -> 5Hz
|
|
self._last_joint = 0.0
|
|
self._last_motor = 0.0
|
|
self._start()
|
|
|
|
# ---------------- backend selection ----------------
|
|
def _start(self) -> None:
|
|
want = (_env("ENG_SOURCE", "auto").lower() or "auto")
|
|
order = ["ros2", "http"] if want == "auto" else [want]
|
|
for b in order:
|
|
try:
|
|
if b == "none":
|
|
break
|
|
if b == "ros2" and self._start_ros2():
|
|
self.backend = "ros2"
|
|
break
|
|
if b == "http" and self._start_http():
|
|
self.backend = "http"
|
|
break
|
|
except Exception as e:
|
|
log.warning("EngineAI source %r failed to start (%s)", b, e)
|
|
if self.backend == "none":
|
|
log.warning("no EngineAI state source active (ENG_SOURCE=%s) — telemetry runs "
|
|
"in heartbeat mode. Check that the unit sources ros_env.sh "
|
|
"(ROS_DOMAIN_ID=%s + CYCLONEDDS_URI) and run tools/probe_eng.sh "
|
|
"on the robot to confirm the topic names.", want, self.cfg.domain_id)
|
|
else:
|
|
self.ok = True
|
|
log.info("EngineAI source up: backend=%s", self.backend)
|
|
|
|
def _start_http(self) -> bool:
|
|
url = _env("ENG_STATE_URL")
|
|
if not url:
|
|
return False
|
|
self._http_url = url
|
|
self._http_period = max(0.2, _env_float("ENG_HTTP_INTERVAL", "1"))
|
|
threading.Thread(target=self._http_loop, daemon=True).start()
|
|
log.info("EngineAI 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("EngineAI http state: HTTP %s", r.status_code)
|
|
except Exception as e:
|
|
log.debug("EngineAI 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
|
|
the vendor's interface_protocol messages work as long as the overlay is
|
|
sourced (the systemd unit sources ros_env.sh, which does exactly that)."""
|
|
try:
|
|
import rclpy
|
|
from rclpy.node import Node
|
|
except Exception as e:
|
|
log.debug("rclpy unavailable (%s)", e)
|
|
return False
|
|
# (topic, type, callback, high_rate_key) — high-rate topics are
|
|
# subscribed raw so the gate can drop messages before deserialization.
|
|
wanted = [
|
|
(_env("ENG_TOPIC_POWER", "/hardware/power_info"),
|
|
_env("ENG_TYPE_POWER", "interface_protocol/msg/PowerInfo"),
|
|
self._ingest_battery, ""),
|
|
(_env("ENG_TOPIC_MOTORS", "/hardware/motor_debug"),
|
|
_env("ENG_TYPE_MOTORS", "interface_protocol/msg/MotorDebug"),
|
|
self._ingest_motors, "motor"),
|
|
(_env("ENG_TOPIC_JOINTS", "/hardware/joint_state"),
|
|
_env("ENG_TYPE_JOINTS", "interface_protocol/msg/JointState"),
|
|
self._ingest_joints, "joint"),
|
|
(_env("ENG_TOPIC_MOTION", "/motion/motion_state"),
|
|
_env("ENG_TYPE_MOTION", "interface_protocol/msg/MotionState"),
|
|
self._ingest_motion, ""),
|
|
(_env("ENG_TOPIC_ODOM", ""),
|
|
_env("ENG_TYPE_ODOM", "nav_msgs/msg/Odometry"), self._ingest_odom, ""),
|
|
]
|
|
if not any(t for t, _, _, _ in wanted):
|
|
return False
|
|
use_raw = _env_bool("ENG_RAW_SUBSCRIBE", True)
|
|
if not rclpy.ok():
|
|
rclpy.init(args=None)
|
|
node = Node("sanad_api_eng")
|
|
n = 0
|
|
for topic, spec, cb, hot in wanted:
|
|
if not topic or not spec:
|
|
continue
|
|
try:
|
|
cls = _import_msg(spec)
|
|
except Exception as e:
|
|
log.warning("EngineAI ros2: cannot import %s for %s (%s) — skipped",
|
|
spec, topic, e)
|
|
continue
|
|
raw = bool(hot) and use_raw
|
|
if raw:
|
|
try:
|
|
node.create_subscription(cls, topic, self._raw_cb(cls, hot),
|
|
_qos(), raw=True)
|
|
except TypeError:
|
|
# rclpy build without raw= support: fall back to a normal
|
|
# subscription (correct, just more CPU). Never a reason to
|
|
# lose the topic entirely.
|
|
log.info("EngineAI ros2: raw subscriptions unsupported here — "
|
|
"normal subscription for %s", topic)
|
|
raw = False
|
|
node.create_subscription(cls, topic, cb, _qos())
|
|
else:
|
|
node.create_subscription(cls, topic, cb, _qos())
|
|
note = ""
|
|
if raw:
|
|
per = self._joint_period if hot == "joint" else self._motor_period
|
|
note = " [raw, gated to ~%.0f Hz]" % (1.0 / per if per > 0 else 0)
|
|
log.info("EngineAI ros2: subscribed %s (%s)%s", topic, spec, note)
|
|
n += 1
|
|
if not n:
|
|
return False
|
|
self._node = node
|
|
threading.Thread(target=lambda: rclpy.spin(node), daemon=True).start()
|
|
return True
|
|
|
|
# ---------------- 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 _gate(self, which: str) -> bool:
|
|
"""Rate gate for the high-frequency streams. True = process this message.
|
|
|
|
Cheap and lock-free: it is the first thing every high-rate callback does,
|
|
including the raw ones, so a dropped message costs one clock read."""
|
|
now = time.monotonic()
|
|
if which == "joint":
|
|
if now - self._last_joint < self._joint_period:
|
|
return False
|
|
self._last_joint = now
|
|
return True
|
|
if now - self._last_motor < self._motor_period:
|
|
return False
|
|
self._last_motor = now
|
|
return True
|
|
|
|
def _raw_cb(self, cls: Any, which: str):
|
|
"""Wrap a high-rate subscription so the rate gate runs BEFORE the message
|
|
is deserialized.
|
|
|
|
Measured on this robot, per subscription, as % of ONE core (12 available):
|
|
|
|
all four topics 26.5 <- as first written
|
|
without /hardware/joint_state 6.5 => joint_state alone ~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
|
|
(max |velocity|) that telemetry samples every 2 s.
|
|
|
|
raw=True hands the callback the serialized bytes, so a message the gate
|
|
drops is never deserialized. That measured 27.2% -> 23.8%: a real win,
|
|
but smaller than expected, which says most of the cost is CycloneDDS
|
|
delivering 500 msg/s and waking the executor, NOT building the Python
|
|
object. Deserialization is only the part we can avoid without changing
|
|
what DDS does.
|
|
|
|
The remaining ~20% could be removed by creating and destroying the
|
|
subscription around each sample, but 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, so the subscription is left stable.
|
|
|
|
If the CPU ever matters more than the "moving" status, ENG_TOPIC_JOINTS=
|
|
(empty) drops this subscription entirely and the agent costs ~6%."""
|
|
from rclpy.serialization import deserialize_message
|
|
sink = self._joints_from if which == "joint" else self._motors_from
|
|
|
|
def cb(data: Any) -> None:
|
|
if not self._gate(which):
|
|
return
|
|
try:
|
|
sink(deserialize_message(data, cls))
|
|
except Exception:
|
|
pass
|
|
return cb
|
|
|
|
def _ingest_battery(self, msg: Any) -> None:
|
|
"""PowerInfo -> the battery record the telemetry schema expects."""
|
|
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
|
|
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")
|
|
lim = self._num(msg, "current_limit")
|
|
err = _dig(msg, self._map.get("power_err", ""))
|
|
en = _dig(msg, self._map.get("power_enable", ""))
|
|
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,
|
|
"current_limit_a": round(lim, 1) if lim is not None else None,
|
|
}
|
|
with self._lock:
|
|
self._bms = rec
|
|
if err is not None:
|
|
try:
|
|
self._power_err = int(err)
|
|
except (TypeError, ValueError):
|
|
pass
|
|
if en is not None:
|
|
self._power_enabled = bool(en)
|
|
self._bms_ts = self._state_ts = time.monotonic()
|
|
except Exception:
|
|
pass
|
|
|
|
def _ingest_motors(self, msg: Any) -> None:
|
|
"""MotorDebug -> per-motor temperatures + per-motor fault/offline flags.
|
|
|
|
This is the field the X2 could never fill: that robot publishes no
|
|
per-motor temperature at all, so its motor_temp was permanently null.
|
|
The PM01 publishes 25 motor temps AND 25 MOSFET temps, so both are
|
|
real readings here — reported, not invented."""
|
|
if not self._gate("motor"):
|
|
return
|
|
self._motors_from(msg)
|
|
|
|
def _motors_from(self, msg: Any) -> None:
|
|
now = time.monotonic()
|
|
try:
|
|
temps = _floats(_dig(msg, self._map.get("temps", "")), 0, 200)
|
|
mos = _floats(_dig(msg, self._map.get("mos_temps", "")), 0, 200)
|
|
faults: List[str] = []
|
|
offline: List[int] = []
|
|
for i, code in enumerate(_seq(_dig(msg, self._map.get("motor_err", "")))):
|
|
try:
|
|
c = int(code)
|
|
except (TypeError, ValueError):
|
|
continue
|
|
if c != 0:
|
|
faults.append(f"joint{i}=0x{c:x}")
|
|
for i, off in enumerate(_seq(_dig(msg, self._map.get("motor_offline", "")))):
|
|
try:
|
|
if int(off):
|
|
offline.append(i)
|
|
except (TypeError, ValueError):
|
|
continue
|
|
with self._lock:
|
|
self._temps = temps
|
|
self._mos_temps = mos
|
|
self._motor_faults = faults
|
|
self._motor_offline = offline
|
|
self._state_ts = now
|
|
except Exception:
|
|
pass
|
|
|
|
def _ingest_joints(self, msg: Any) -> None:
|
|
"""JointState -> max |velocity| across joints, which drives 'moving'."""
|
|
if not self._gate("joint"):
|
|
return
|
|
self._joints_from(msg)
|
|
|
|
def _joints_from(self, msg: Any) -> None:
|
|
now = time.monotonic()
|
|
try:
|
|
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._max_vel = max_vel
|
|
self._state_ts = now
|
|
except Exception:
|
|
pass
|
|
|
|
def _ingest_motion(self, msg: Any) -> None:
|
|
"""MotionState -> the live motion task + the transitions it allows.
|
|
|
|
Read-only: the transitions list is REPORTED so the dashboard can show
|
|
what the robot would accept. The agent never requests one."""
|
|
try:
|
|
task = _dig(msg, self._map.get("motion", ""))
|
|
trans = _seq(_dig(msg, self._map.get("transitions", "")))
|
|
with self._lock:
|
|
if task is not None:
|
|
self._motion = str(task)
|
|
self._transitions = [str(t) for t in trans]
|
|
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 every extractor."""
|
|
self._ingest_battery(d)
|
|
self._ingest_motors(d)
|
|
self._ingest_joints(d)
|
|
self._ingest_motion(d)
|
|
self._ingest_odom(d)
|
|
try:
|
|
fw = _dig(d, self._map.get("fw", ""))
|
|
if isinstance(fw, dict):
|
|
with self._lock:
|
|
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),
|
|
"mos_temps": list(self._mos_temps),
|
|
"motor_faults": list(self._motor_faults),
|
|
"motor_offline": list(self._motor_offline),
|
|
"power_err": self._power_err,
|
|
"power_enabled": self._power_enabled,
|
|
"max_vel": self._max_vel,
|
|
"xy": dict(self._xy) if self._xy else None,
|
|
"motion": self._motion,
|
|
"transitions": list(self._transitions),
|
|
"fw": dict(self._fw),
|
|
}
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# control card — the PM01 reports its own motion mode natively
|
|
# --------------------------------------------------------------------------- #
|
|
_control_cache: Dict[str, Any] = {"ts": 0.0, "data": None}
|
|
|
|
|
|
def read_control(cfg: Config, snap: Dict[str, Any]) -> Optional[Dict[str, Any]]:
|
|
"""READ-ONLY control status (no robot control, no motion).
|
|
|
|
Unlike the X2 — where the mode had to be guessed from an undocumented FSM id
|
|
and reported "unknown" — the PM01 publishes /motion/motion_state, which
|
|
carries the mode BY NAME plus the exact set of transitions it will accept.
|
|
So control.mode and control.switchable_modes are the robot's own words.
|
|
|
|
The Sanad dashboard's /api/controller/status is merged in when reachable
|
|
(arm/teleop/velocity detail), but it is never required."""
|
|
out: Dict[str, Any] = {
|
|
"mode": snap.get("motion"), # e.g. "pd_sitdown"
|
|
"switchable_modes": snap.get("transitions") or [],
|
|
"remote_switch_enabled": bool(cfg.control_enable),
|
|
"source": "ros2:/motion/motion_state" if snap.get("motion") else None,
|
|
}
|
|
now = time.monotonic()
|
|
if _control_cache["data"] is not None and now - _control_cache["ts"] < 1.5:
|
|
d = _control_cache["data"]
|
|
else:
|
|
url = cfg.control_url
|
|
if not url:
|
|
port = _REMOTE_STAT.get("port")
|
|
url = f"http://127.0.0.1:{port}/api/controller/status" if port else ""
|
|
d = None
|
|
if url:
|
|
try:
|
|
r = requests.get(url, timeout=2)
|
|
d = r.json() if r.ok else None
|
|
except Exception:
|
|
d = None
|
|
_control_cache.update(ts=now, data=d)
|
|
if isinstance(d, dict):
|
|
out.update({
|
|
"fsm_id": d.get("fsm_id"),
|
|
"fsm_mode": d.get("fsm_mode"),
|
|
"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"),
|
|
})
|
|
if out["mode"] is None and not isinstance(d, dict):
|
|
return None
|
|
return out
|
|
|
|
|
|
class RosbridgePosition:
|
|
"""Position over a rosbridge websocket (ENG_POSITION_SOURCE=rosbridge)."""
|
|
|
|
def __init__(self, cfg: Config):
|
|
self.cfg = cfg
|
|
self._xy: Optional[Dict[str, float]] = None
|
|
self._lock = threading.Lock()
|
|
self._stop = False
|
|
self._topic = _env("ENG_TOPIC_ODOM", "") or "/odom"
|
|
self._type = _env("ENG_TYPE_ODOM", "nav_msgs/Odometry").replace("/msg/", "/")
|
|
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": self._topic,
|
|
"type": self._type, "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 ENG_SOURCE.
|
|
|
|
The PM01 as shipped publishes NO odometry topic at all (see
|
|
docs/PM01_INTERFACE.md) — the motion stack is a whole-body controller, not a
|
|
navigation stack, so there is nothing to localise against until a nav
|
|
bringup is started. Position therefore reports null rather than a fabricated
|
|
origin. This class exists so that the moment an odometry topic does appear,
|
|
it is one .env line (ENG_TOPIC_ODOM=...) to start reporting it."""
|
|
|
|
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("ENG_FIELD_X", "") or "pose.pose.position.x"
|
|
self._py = _env("ENG_FIELD_Y", "") or "pose.pose.position.y"
|
|
topic = _env("ENG_TOPIC_ODOM", "")
|
|
spec = _env("ENG_TYPE_ODOM", "nav_msgs/msg/Odometry")
|
|
if not topic or not spec:
|
|
log.warning("ENG_POSITION_SOURCE=ros2 but ENG_TOPIC_ODOM is empty — the PM01 "
|
|
"publishes no odometry; position stays null until a nav stack runs")
|
|
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:
|
|
# EngineAiSource may already have started rclpy when ENG_SOURCE=ros2.
|
|
if not rclpy.ok():
|
|
rclpy.init(args=None)
|
|
node = Node("sanad_api_eng_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("EngineAI 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
|
|
|
|
|
|
class HttpPosition:
|
|
"""Position from the Sanad dashboard's nav API (ENG_POSITION_SOURCE=http).
|
|
|
|
The PM01's Sanad app exposes /api/nav/status; when its nav bringup is
|
|
running that payload carries the robot pose. While bringup is down the
|
|
endpoint answers reachable:false and this reports null — never a stale or
|
|
invented coordinate."""
|
|
|
|
def __init__(self, cfg: Config):
|
|
self.cfg = cfg
|
|
self._xy: Optional[Dict[str, float]] = None
|
|
self._lock = threading.Lock()
|
|
self._stop = False
|
|
self._url = _env("ENG_POSITION_URL", "") or "http://127.0.0.1:8014/api/nav/status"
|
|
self._px = _env("ENG_POSITION_FIELD_X", "") or "pose.x"
|
|
self._py = _env("ENG_POSITION_FIELD_Y", "") or "pose.y"
|
|
self._period = max(0.5, _env_float("ENG_POSITION_INTERVAL", "2"))
|
|
self.ok = True
|
|
threading.Thread(target=self._run, daemon=True).start()
|
|
log.info("EngineAI position: http %s (%s / %s)", self._url, self._px, self._py)
|
|
|
|
def _run(self) -> None:
|
|
while not self._stop:
|
|
try:
|
|
r = requests.get(self._url, timeout=3)
|
|
if r.ok:
|
|
d = r.json()
|
|
x, y = _dig(d, self._px), _dig(d, self._py)
|
|
if x is not None and y is not None:
|
|
with self._lock:
|
|
self._xy = {"x": round(float(x), 3), "y": round(float(y), 3)}
|
|
else:
|
|
with self._lock:
|
|
self._xy = None
|
|
except Exception as e:
|
|
log.debug("http position poll failed: %s", e)
|
|
time.sleep(self._period)
|
|
|
|
def get(self) -> Optional[Dict[str, float]]:
|
|
with self._lock:
|
|
return dict(self._xy) if self._xy else None
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# map sync (saved maps → EVERY fleet server, uploaded ONCE per content per server)
|
|
# --------------------------------------------------------------------------- #
|
|
@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)
|
|
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()
|
|
|
|
|
|
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
|
|
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: <stem>.yaml + <stem>.pgm."""
|
|
roots = [cfg.maps_dir, cfg.maps_dir / cfg.robot, cfg.maps_dir / "maps_slam"]
|
|
roots += list(cfg.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))
|
|
# a keepout-BAKED twin 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")
|
|
out.append(cfg.web_data_dir / "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", ""),
|
|
))
|
|
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, Dict[str, str]]:
|
|
"""Uploaded-map state, keyed PER SERVER: {server_name: {map_path: fingerprint}}.
|
|
|
|
Per-server on purpose. With one shared key, enabling a second fleet server
|
|
later would find every map already 'uploaded' and that server would never
|
|
receive them."""
|
|
raw = _read_json(_state_file(cfg), {}) if _state_file(cfg).exists() else {}
|
|
if not isinstance(raw, dict):
|
|
return {}
|
|
# migrate the flat {path: fingerprint} layout written by the X2 agent
|
|
if raw and all(isinstance(v, str) for v in raw.values()):
|
|
primary = cfg.endpoints[0].name if cfg.endpoints else "default"
|
|
return {primary: dict(raw)}
|
|
return {k: dict(v) for k, v in raw.items() if isinstance(v, dict)}
|
|
|
|
|
|
def save_state(cfg: Config, state: Dict[str, 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_map(cfg: Config, m: MapArtifact, session: requests.Session,
|
|
targets: List[Endpoint]) -> Dict[str, Any]:
|
|
"""Upload ONE map to the given servers. Returns {server: {ok, code, error}}.
|
|
|
|
The body is built once and posted to each target, so every server receives
|
|
byte-identical map content."""
|
|
if m.fmt == "slam_toolbox":
|
|
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 {e.name: {"ok": False, "code": None, "error": "unparseable pgm"} for e in targets}
|
|
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),
|
|
})
|
|
results = _post_each(_scoped(cfg, targets), session, cfg.map_endpoint_tmpl,
|
|
json_body=body, what=f"map[{m.stem}]")
|
|
if _any_ok(results):
|
|
log.info("map uploaded: %s (slam_toolbox %dx%d @ %sm, %d points) -> %s",
|
|
m.stem, pgm["width"], pgm["height"], ymeta.get("resolution"),
|
|
len(m.points), _fmt_results(results))
|
|
return results
|
|
|
|
meta = build_meta(cfg, m)
|
|
if cfg.map_upload_mode == "base64json":
|
|
body = dict(meta)
|
|
body["db_base64"] = base64.b64encode(m.path.read_bytes()).decode("ascii")
|
|
results = _post_each(_scoped(cfg, targets), session, cfg.map_endpoint_tmpl,
|
|
json_body=body, what=f"map[{m.name}]")
|
|
else: # multipart (default) — read once, post to each server
|
|
blob = m.path.read_bytes()
|
|
results = _post_each(_scoped(cfg, targets), session, cfg.map_endpoint_tmpl,
|
|
body_bytes=blob, filename=m.name,
|
|
form={"meta": json.dumps(meta)}, what=f"map[{m.name}]")
|
|
if _any_ok(results):
|
|
log.info("map uploaded: %s (%.2f MB, %d points) -> %s",
|
|
m.name, m.size / 1024 / 1024, len(m.points), _fmt_results(results))
|
|
return results
|
|
|
|
|
|
def _scoped(cfg: Config, targets: List[Endpoint]) -> Config:
|
|
"""A shallow Config view limited to `targets` (for per-server map uploads)."""
|
|
import copy
|
|
c = copy.copy(cfg)
|
|
c.endpoints = list(targets)
|
|
return c
|
|
|
|
|
|
# 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, "servers": {},
|
|
}
|
|
|
|
|
|
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 saved maps and upload anything new, PER SERVER.
|
|
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,
|
|
servers={}, error=f"no saved map found "
|
|
f"(maps_dir={cfg.maps_dir}, robot={cfg.robot})")
|
|
return 0
|
|
|
|
state = load_state(cfg)
|
|
uploaded = failed = unstable = too_large = current = 0
|
|
per_server: Dict[str, Any] = {}
|
|
last_err: Optional[str] = None
|
|
now = time.time()
|
|
for m in maps:
|
|
key = str(m.path.resolve())
|
|
# which servers still need THIS content?
|
|
targets = [e for e in cfg.live()
|
|
if force or state.get(e.name, {}).get(key) != m.fingerprint()]
|
|
if not targets:
|
|
current += 1
|
|
continue # every server already has this exact content — one-time rule
|
|
# stability guard: a map 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 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) to %s",
|
|
m.name, m.size / 1024 / 1024, len(m.points),
|
|
",".join(e.name for e in targets))
|
|
continue
|
|
results = upload_map(cfg, m, session, targets)
|
|
per_server.update({k: v.get("code") for k, v in results.items()})
|
|
got_one = False
|
|
for e in targets:
|
|
r = results.get(e.name) or {}
|
|
if r.get("ok"):
|
|
state.setdefault(e.name, {})[key] = m.fingerprint()
|
|
got_one = True
|
|
else:
|
|
last_err = f"{m.name} -> {e.name}: {r.get('error') or 'failed'}"
|
|
if got_one:
|
|
save_state(cfg, state)
|
|
uploaded += 1
|
|
if any(not (results.get(e.name) or {}).get("ok") for e in targets):
|
|
failed += 1
|
|
|
|
if failed and not uploaded:
|
|
_set_map_status(state="failed", uploaded=False, maps_found=len(maps),
|
|
last_map=maps[0].stem, servers=per_server, error=last_err)
|
|
elif uploaded or current:
|
|
note = last_err
|
|
if not note and too_large:
|
|
note = f"{too_large} map(s) skipped: exceed server upload cap (~{cfg.map_max_upload_mb:.0f} MB)"
|
|
elif not note and 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, servers=per_server, error=note)
|
|
elif too_large:
|
|
_set_map_status(state="failed", uploaded=False, maps_found=len(maps),
|
|
last_map=maps[0].stem, servers=per_server,
|
|
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:
|
|
_set_map_status(state="pending", uploaded=False, maps_found=len(maps),
|
|
last_map=maps[0].stem, servers=per_server,
|
|
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, servers=per_server, 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 (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 servers."""
|
|
|
|
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, "servers": {}}
|
|
_ALERTS_STAT: Dict[str, Any] = {"sent": 0, "last": None, "last_time": None,
|
|
"ok": None, "servers": {}}
|
|
|
|
# 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 (the Sanad app) and feeds them into
|
|
the shipped log lines, labeled "[<project>-logs] …".
|
|
|
|
Sources, in priority order:
|
|
PROJECT_LOG_PATH explicit log file (or dir -> newest *.log)
|
|
PROJECT_LOG_CONTAINER a docker container name; "auto" (default) scans the
|
|
host's docker metadata for a RUNNING Sanad project
|
|
(sanad-t8, sanadr1, sanad-p4, sanad*)
|
|
|
|
Reads the container's json-log directly from the docker data dir — no docker
|
|
socket needed, read-only, cannot disturb the project. On this robot that
|
|
path is root-owned, which is one reason the agent runs as a system service."""
|
|
|
|
KNOWN = ("sanad-t8", "sanadr1", "sanad-p4", "sanadv3", "sanad")
|
|
# /host/... first (containerised agent with the read-only / mount), then the
|
|
# native path (this agent runs natively on the PM01).
|
|
DOCKER_ROOTS = ("/host/var/lib/docker/containers", "/var/lib/docker/containers")
|
|
|
|
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
|
|
self._pending: List[str] = []
|
|
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
|
|
base = None
|
|
for root in self.DOCKER_ROOTS:
|
|
if Path(root).is_dir():
|
|
base = Path(root)
|
|
break
|
|
if base is None:
|
|
return
|
|
want = cfg.project_log_container
|
|
candidates: List[Any] = []
|
|
try:
|
|
config_files = list(base.glob("*/config.v2.json"))
|
|
except PermissionError:
|
|
log.warning("cannot read %s (permission denied) — project logs disabled. "
|
|
"Run the agent as a system service (root) to ship them.", base)
|
|
return
|
|
for cf in config_files:
|
|
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 = 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(lp)
|
|
if not p.exists() and Path("/host" + lp).exists():
|
|
p = Path("/host" + 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
|
|
# backfill: the last N RELEVANT lines (post-filter) from the tail, so
|
|
# access-log noise 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 is returned on the 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 on EVERY
|
|
server. Best-effort: failures are logged at DEBUG only (below the ring's
|
|
level -> no feedback loop). Lines are requeued unless at least one server
|
|
accepted them."""
|
|
lines = _LOG_RING.drain()
|
|
if _PROJECT_TAIL is not None and _PROJECT_TAIL.active:
|
|
lines.extend(_PROJECT_TAIL.poll())
|
|
if not lines:
|
|
return
|
|
body = {"sn": cfg.sn, "name": cfg.name, "lines": lines, "ts": int(time.time())}
|
|
results = _post_each(cfg, session, cfg.logs_endpoint, json_body=body,
|
|
what="logs", quiet=True)
|
|
_LOGS_STAT["servers"] = {n: r.get("code") for n, r in results.items()}
|
|
if _any_ok(results):
|
|
_LOGS_STAT.update(last_sent=_now_str(), ok=True)
|
|
_LOGS_STAT["lines_sent"] += len(lines)
|
|
log.debug("logs shipped: %d lines -> %s", len(lines), _fmt_results(results))
|
|
else:
|
|
_LOGS_STAT["ok"] = False
|
|
_LOG_RING.requeue(lines) # retry next cycle
|
|
log.debug("logs ship failed: %s (%d lines requeued)", _fmt_results(results), len(lines))
|
|
|
|
|
|
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. No changes to the Sanad app: we only probe its port
|
|
# and POST the URL to /{sn}/remote on every server.
|
|
# --------------------------------------------------------------------------- #
|
|
_REMOTE_STAT: Dict[str, Any] = {"url": None, "port": None, "kind": None, "ok": None,
|
|
"ssh": None, "ssh_ok": None, "servers": {}}
|
|
|
|
|
|
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 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:
|
|
results = _post_each(cfg, session, cfg.remote_endpoint, json_body={
|
|
"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())}, what="remote(web)", quiet=True)
|
|
ok = _any_ok(results)
|
|
_REMOTE_STAT.update(url=d["url"], port=d["port"], kind=cfg.remote_kind, ok=ok,
|
|
servers={n: r.get("code") for n, r in results.items()})
|
|
if ok:
|
|
log.info("remote dashboard registered: %s -> %s", d["url"], _fmt_results(results))
|
|
else:
|
|
_REMOTE_STAT.update(url=None, port=None, ok=None)
|
|
# 2) SSH access (kind=ssh) — "ssh <user>@<ip>". 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}"
|
|
results = _post_each(cfg, session, cfg.remote_endpoint, json_body={
|
|
"sn": cfg.sn, "name": cfg.name, "kind": "ssh",
|
|
"url": f"ssh://{cfg.ssh_user}@{host}:{cfg.ssh_port}", # URL-valid form
|
|
"command": cmd, # "ssh <user>@<ip>"
|
|
"host": host, "port": cfg.ssh_port, "user": cfg.ssh_user,
|
|
"label": f"{cfg.name} — SSH", "ts": int(time.time())},
|
|
what="remote(ssh)", quiet=True)
|
|
ok = _any_ok(results)
|
|
_REMOTE_STAT.update(ssh=cmd, ssh_ok=ok)
|
|
if ok:
|
|
log.info("remote SSH registered: %s -> %s", cmd, _fmt_results(results))
|
|
|
|
|
|
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 on every server."""
|
|
body = {"sn": cfg.sn, "name": cfg.name, "alert": text,
|
|
"message": text, "ts": int(time.time())}
|
|
results = _post_each(cfg, session, cfg.alert_endpoint, json_body=body,
|
|
what="alert", quiet=True)
|
|
ok = _any_ok(results)
|
|
_ALERTS_STAT.update(last=text, last_time=_now_str(), ok=ok,
|
|
servers={n: r.get("code") for n, r in results.items()})
|
|
if ok:
|
|
_ALERTS_STAT["sent"] += 1
|
|
log.info("alert sent: %s -> %s", text[:120], _fmt_results(results))
|
|
return ok
|
|
|
|
|
|
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.
|
|
|
|
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 would re-fire the same fault every
|
|
POLL_INTERVAL for as long as it lasted.
|
|
|
|
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 for error/billing patterns and fires an
|
|
alert on each NEW signature (deduped with a cooldown)."""
|
|
|
|
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 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]:
|
|
"""Fault STRINGS, not objects — the fleet ingest 500s on fault objects.
|
|
|
|
The PM01 exposes real hardware fault channels the X2 did not: PowerInfo
|
|
carries enable+error_code, and MotorDebug carries a per-motor error_code and
|
|
offline flag. Those become first-class faults rather than being inferred."""
|
|
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)")
|
|
mos = snap.get("mos_temps") or []
|
|
if mos and max(mos) >= cfg.mos_temp_max:
|
|
faults.append(f"MOS_OVERTEMP: driver MOSFET temp {max(mos):.0f}C (warning)")
|
|
perr = snap.get("power_err")
|
|
if perr:
|
|
faults.append(f"POWER_FAULT: power_info error_code 0x{int(perr):x} (critical)")
|
|
if snap.get("power_enabled") is False:
|
|
faults.append("POWER_DISABLED: battery output reports enable=false (critical)")
|
|
mf = snap.get("motor_faults") or []
|
|
if mf:
|
|
faults.append(f"MOTOR_FAULT: {len(mf)} motor(s) reporting an error code "
|
|
f"[{', '.join(mf[:6])}] (critical)")
|
|
off = snap.get("motor_offline") or []
|
|
if off:
|
|
faults.append(f"MOTOR_OFFLINE: {len(off)} motor(s) offline "
|
|
f"[{', '.join(str(i) for i in off[:8])}] (critical)")
|
|
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
|
|
|
|
|
|
# Motion tasks that mean "powered down / not actively controlled". Everything
|
|
# else is a live controller, and then velocity decides moving vs idle.
|
|
_PASSIVE_MOTIONS = {"passive", "idle", "none", ""}
|
|
|
|
|
|
def derive_status(cfg: Config, snap: Dict[str, Any]) -> str:
|
|
"""charging | moving | idle | offline — the same vocabulary the X2 reports."""
|
|
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) > _env_float("MOVING_VEL", "0.15"):
|
|
return "moving"
|
|
return "idle"
|
|
|
|
|
|
def build_telemetry(cfg: Config, mac: str, reader: Optional[EngineAiSource],
|
|
pos: Optional[Any],
|
|
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": 54.8, "temp_c": None, "soh": 0, "cycles": 0,
|
|
"current_limit_a": 90.0},
|
|
"state_age": 0.1, "temps": [sim.get("temp", 45)],
|
|
"mos_temps": [sim.get("temp", 45) + 2], "motor_faults": [],
|
|
"motor_offline": [], "power_err": 0, "power_enabled": True,
|
|
"max_vel": sim.get("max_vel", 0.0), "xy": sim.get("position"),
|
|
"motion": sim.get("motion", "passive"), "transitions": [], "fw": {}}
|
|
else:
|
|
snap = (reader.snapshot() if reader else
|
|
{"bms": None, "state_age": None, "temps": [], "mos_temps": [],
|
|
"motor_faults": [], "motor_offline": [], "power_err": None,
|
|
"power_enabled": None, "max_vel": 0.0, "xy": None,
|
|
"motion": None, "transitions": [], "fw": {}})
|
|
|
|
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)
|
|
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"),
|
|
"current_limit_a": bms.get("current_limit_a")}
|
|
|
|
temps = snap.get("temps") or []
|
|
mos = snap.get("mos_temps") or []
|
|
motor_temp = None
|
|
if temps:
|
|
motor_temp = {"max": round(max(temps), 1), "avg": round(sum(temps) / len(temps), 1),
|
|
"min": round(min(temps), 1), "count": len(temps)}
|
|
if mos:
|
|
motor_temp["mos_max"] = round(max(mos), 1)
|
|
motor_temp["mos_avg"] = round(sum(mos) / len(mos), 1)
|
|
|
|
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. pm01_150)
|
|
"mac": mac,
|
|
"brand": cfg.brand, # engineai
|
|
"type": cfg.robot_type, # humanoid
|
|
"model": cfg.model, # pm01
|
|
"software": read_software(cfg), # ros/os/kernel/arch/python/agent
|
|
"firmware": {**read_firmware_static(), **(snap.get("fw") or {})},
|
|
# board/l4t/product/board_serial + live fw
|
|
"battery": battery, # null = couldn't read (heartbeat)
|
|
"charging": charging,
|
|
"battery_detail": battery_detail,
|
|
"motor_temp": motor_temp, # REAL on the PM01 (25 motors) — null = not receiving
|
|
"storage": read_storage(cfg),
|
|
"status": status,
|
|
"position": position, # null when no odom/localization source
|
|
"control": read_control(cfg, snap), # live motion task + allowed transitions
|
|
"faults": faults,
|
|
"map": get_map_status(), # SHOWS whether the saved map made it to each 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. "sanad-t8-logs"; null = no project found
|
|
"remote": (dict(_REMOTE_STAT) if _REMOTE_STAT.get("url") else None),
|
|
"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:
|
|
"""POST the identical payload to every enabled fleet server."""
|
|
results = _post_each(cfg, session, cfg.telemetry_endpoint,
|
|
json_body=payload, what="telemetry")
|
|
mp = payload.get("map") or {}
|
|
mt = payload.get("motor_temp") or {}
|
|
log.info("telemetry ok: battery=%s charging=%s status=%s mode=%s pos=%s "
|
|
"motor_max=%s faults=%d map=%s -> %s",
|
|
payload["battery"], payload["charging"], payload["status"],
|
|
(payload.get("control") or {}).get("mode"),
|
|
payload["position"], mt.get("max"), len(payload["faults"]),
|
|
mp.get("state"), _fmt_results(results))
|
|
return _any_ok(results)
|
|
|
|
|
|
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,
|
|
"motion": "rl_basic" if moving else "pd_sitdown",
|
|
"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 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="EngineAI PM01 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, 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)
|
|
|
|
# systemd stops the unit with SIGINT (KillSignal=SIGINT) and falls back to
|
|
# SIGTERM. Handle both so a restart tears ROS down cleanly instead of
|
|
# aborting — otherwise every `systemctl restart` is recorded as a failure.
|
|
import signal
|
|
|
|
def _on_signal(sig: int, _frame: Any) -> None:
|
|
log.info("received %s — stopping", signal.Signals(sig).name)
|
|
_exit(0)
|
|
|
|
for _s in (signal.SIGINT, signal.SIGTERM):
|
|
try:
|
|
signal.signal(_s, _on_signal)
|
|
except Exception:
|
|
pass
|
|
|
|
_load_dotenv()
|
|
cfg = Config.from_env()
|
|
if args.interval is not None:
|
|
cfg.poll_interval = args.interval
|
|
|
|
if args.list:
|
|
cmd_list(cfg)
|
|
return _exit(0)
|
|
|
|
_init_start_times(cfg)
|
|
global _PROJECT_TAIL, _LOG_ALERTS
|
|
_PROJECT_TAIL = ProjectLogTail(cfg)
|
|
_LOG_ALERTS = LogAlertScanner(cfg)
|
|
mac = read_mac(cfg.mac_interface)
|
|
log.info("sanad_api_eng — sn=%s name=%s mac=%s iface=%s pos=%s map_dir=%s "
|
|
"map_every=%.0fs%s",
|
|
cfg.sn, cfg.name, mac, cfg.mac_interface, cfg.position_source,
|
|
cfg.maps_dir, cfg.map_poll_interval,
|
|
" [SIMULATE]" if args.simulate else "")
|
|
for e in cfg.endpoints:
|
|
log.info(" fleet server %-10s %s %s", e.name, e.url,
|
|
"ENABLED" if e.enabled else "disabled (no token / SERVER_*_ENABLE=0)")
|
|
|
|
session = requests.Session()
|
|
|
|
if args.map_only:
|
|
map_sync_once(cfg, session, force=args.force, dry_run=args.dry_run)
|
|
return _exit(0)
|
|
|
|
reader = None
|
|
pos = None
|
|
if not args.simulate:
|
|
reader = EngineAiSource(cfg)
|
|
if cfg.position_source == "rosbridge":
|
|
pos = RosbridgePosition(cfg)
|
|
elif cfg.position_source == "ros2":
|
|
pos = Ros2Position(cfg)
|
|
elif cfg.position_source == "http":
|
|
pos = HttpPosition(cfg)
|
|
time.sleep(1.5) # let the first ROS messages land before the first post
|
|
|
|
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, indent=2))
|
|
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 _exit(0)
|
|
for _ in range(3):
|
|
one_telemetry()
|
|
time.sleep(min(cfg.poll_interval, 1.0))
|
|
return _exit(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 to %d server(s); map check every %.0fs; logs every %.0fs",
|
|
cfg.poll_interval, len(cfg.live()), 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 _exit(0)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|