Monitorear robots Recon

Dos tipos de lecturas cubren una flota de robots: las operaciones de estado dicen qué hace cada robot ahora, y la operación de eventos dice qué hizo. Los ejemplos usan la función auxiliar robot_api de la descripción general.

Necesitas una clave de API con acceso de solo lectura a las ubicaciones de los robots.

Estado de la flota

getMinimalStateList devuelve una entrada por robot:

for robot in robot_api("getMinimalStateList", {})["minimalStates"]:
    progress = ""
    if robot.get("activity") == "ROUTE_RUN" and robot.get("waypointTotal"):
        progress = f"stop {robot['waypointIndex'] + 1} of {robot['waypointTotal']}"
    print(f"{robot['name']}: {robot['connectionStatus']}, {robot.get('batteryPercent')}%, {robot.get('activity')} {progress}")

Para un solo robot, getFullState devuelve los mismos campos más las capacidades del robot. Incluye force (false sirve); sin él, la llamada se rechaza:

state = robot_api("getFullState", {"deviceUuid": ROBOT_UUID, "force": False})["fullState"]
capability = state.get("robotPlatformCapability") or {}
print(state["activity"], state.get("phase"), capability.get("supportedActions"))

Usa activity y phase para saber si un robot está en su base, cargando o trabajando; los campos mode y charging no se informan.

Actividades

Se pueden agregar valores nuevos. Trata como ocupado cualquier valor que no reconozcas.

Historial de eventos

findRobotEvents devuelve lo que hizo un robot entre startMs (incluido) y endMs (excluido). La respuesta es JSON delimitado por saltos de línea (application/x-ndjson), con un evento por línea, del más reciente al más antiguo:

import json
import time

def robot_events(robot_uuid, start_ms, end_ms):
    response = requests.post(
        f"{API}/findRobotEvents",
        headers=HEADERS,
        json={"deviceUuid": robot_uuid, "startMs": start_ms, "endMs": end_ms},
        timeout=60,
    )
    response.raise_for_status()
    events = []
    for line in response.text.splitlines():
        if not line.strip():
            continue
        event = json.loads(line)
        if "_error" in event:
            raise RuntimeError(f"findRobotEvents stopped early: {event.get('_message')}")
        events.append(event)
    return events

now = int(time.time() * 1000)
for event in robot_events(ROBOT_UUID, now - 24 * 3600 * 1000, now):
    print(event["timestampMs"], event.get("activity"), event.get("event"), event.get("phase"), event.get("failureReason"))

• No hay paginación. Pide una ventana que puedas manejar, por ejemplo un día, y retrocede ventana por ventana para periodos más largos.
• Una última línea con _error significa que el flujo se detuvo antes del final, aunque el estado sea 200. Vuelve a pedir esa ventana.
• Los eventos pueden llegar tarde. Un robot que estuvo desconectado sube lo que registró cuando se reconecta, con el timestampMs original. Si recopilas eventos de forma periódica, superpón las ventanas y omite los eventos cuyo uuid ya guardaste.

Todas las actividades siguen el mismo ciclo de vida: STARTED, luego eventos PROGRESS a medida que avanza por sus fases (llegar a una parada es un evento PROGRESS cuyo phaseEvent es COMPLETED), posiblemente SUSPENDED y RESUMED cuando algo de mayor prioridad la interrumpe, y exactamente uno de COMPLETED, FAILED o CANCELED al final. Una actividad que el robot rechazó es un único evento REJECTED, con preemptedBy indicando lo que la bloqueó.

Genera alertas sobre lo que importa

Consultar la lista de estado cada uno o dos minutos queda muy por debajo de tu límite de tasa. Los eventos de robots no llegan por el WebSocket de eventos ni por los webhooks de organización; para recibir actualizaciones en tiempo real, conéctate al socket en vivo del robot.