From 2bacf592c7643bef2cc7fa27aa71a939a50e8ac7 Mon Sep 17 00:00:00 2001 From: Patrick Date: Sun, 7 Jun 2026 00:25:59 +0200 Subject: [PATCH] Add telemetry --- README.md | 3 +- orca_core/hardware_hand.py | 79 +++++++++++++++++- orca_core/telemetry.py | 159 +++++++++++++++++++++++++++++++++++++ 3 files changed, 238 insertions(+), 3 deletions(-) create mode 100644 orca_core/telemetry.py diff --git a/README.md b/README.md index 92623d3..76c1f51 100644 --- a/README.md +++ b/README.md @@ -10,6 +10,8 @@ Orca Core is the core control package of the ORCA Hand. It's used to abstract hardware, provide scripts for calibration, tensioning and to control the hand with simple high-level control methods in joint space. +Orca Core sends minimal usage telemetry by default; opt out with `python -m orca_core.telemetry disable`. + ## Get Started To get started with Orca Core, follow these steps: @@ -95,4 +97,3 @@ port: /dev/ttyACM0 # or /'dev/cu.usbmodemXXXX' on macOS baudrate: 1000000 # 1M for v2; 3M for v1 motor_type: dynamixel # or 'feetech' ``` - diff --git a/orca_core/hardware_hand.py b/orca_core/hardware_hand.py index bb1877f..a1efbe9 100644 --- a/orca_core/hardware_hand.py +++ b/orca_core/hardware_hand.py @@ -26,6 +26,7 @@ from .hardware.motor_client import MotorClient from .hardware.sensing.types import ResultantReading, TactileReading, TaxelReading from .hardware.tactile_client import TactileClient +from .telemetry import Telemetry from .utils.utils import ( auto_detect_port, find_single_usb_serial_port, @@ -89,6 +90,7 @@ def __init__( model_version: str | None = None, model_name: str | None = None, config: OrcaHandConfig | None = None, + telemetry_enabled: bool = True, ): super().__init__( config_path=config_path, @@ -98,6 +100,17 @@ def __init__( model_name=model_name, ) + self.telemetry = Telemetry(enabled=telemetry_enabled) + self.telemetry.emit( + "session_start", + hand_class=type(self).__name__, + hand_type=self.config.type, + model_version=model_version, + model_name=model_name, + motor_count=len(self.config.motor_ids), + joint_count=len(self.config.joint_ids), + ) + self._wrap_offsets_dict: Dict[int, float] = None self._motor_client: MotorClient = None self._motor_lock: RLock = RLock() @@ -243,10 +256,22 @@ def connect(self) -> tuple[bool, str]: self._motor_client.connect() self._persist_resolved_driver(motor_type, port, baudrate) + self.telemetry.emit( + "connect", + success=True, + motor_type=motor_type, + baudrate=baudrate, + ) return True, f"Connection successful ({motor_type} @ {port}, {baudrate} baud)" except Exception as e: self._motor_client = None + self.telemetry.emit( + "connect", + success=False, + reason="exception", + error=type(e).__name__, + ) return False, f"Connection failed on {port}: {str(e)}" def _persist_resolved_driver(self, motor_type: str, port: str, baudrate: int) -> None: @@ -289,8 +314,15 @@ def disconnect(self) -> tuple[bool, str]: self.disable_torque() time.sleep(0.1) self._motor_client.disconnect() + self.telemetry.emit("disconnect", success=True) return True, "Disconnected successfully" except Exception as e: + self.telemetry.emit( + "disconnect", + success=False, + reason="exception", + error=type(e).__name__, + ) return False, f"Disconnection failed: {str(e)}" def is_connected(self) -> bool: @@ -480,6 +512,7 @@ def _get_joint_positions(self) -> OrcaJointPositions: def _set_joint_positions(self, joint_pos: OrcaJointPositions) -> bool: motor_pos = self._joint_to_motor_pos(joint_pos.as_dict()) self._set_motor_pos(motor_pos) + self.telemetry.sample("joint_positions", positions=joint_pos.as_dict()) return True def init_joints(self, force_calibrate: bool = False, move_to_neutral: bool = True): @@ -584,11 +617,34 @@ def calibrate( :class:`~orca_core.CalibrationResult`. Partial progress is written to disk after every step so an interrupted run is never fully lost. """ + self.telemetry.emit( + "calibration_start", + blocking=blocking, + force_wrist=force_wrist, + joint_count=( + len(joints) if joints is not None else len(self.config.joint_ids) + ), + ) if blocking: self._task_stop_event.clear() - result = self._calibrate(force_wrist=force_wrist, joints=joints) + try: + result = self._calibrate(force_wrist=force_wrist, joints=joints) + except Exception as e: + self.telemetry.emit( + "calibration_end", + success=False, + reason="exception", + error=type(e).__name__, + ) + raise if result is not None: self.calibration = result + self.telemetry.emit( + "calibration_end", + success=result is not None, + calibrated=result.calibrated if result is not None else False, + wrist_calibrated=result.wrist_calibrated if result is not None else False, + ) else: self._start_task(self._calibrate_and_apply, force_wrist=force_wrist, joints=joints) @@ -1198,6 +1254,7 @@ def _jitter( def _tension(self, move_motors: bool = True): # TODO(fracapuano): Move this to a standard stateless function + self.telemetry.emit("tension_start", move_motors=move_motors) control_mode = self.config.control_mode self.set_control_mode(CURRENT_BASED_POSITION) if move_motors: @@ -1258,6 +1315,11 @@ def _tension(self, move_motors: bool = True): finally: self.set_control_mode(control_mode) self.disable_torque() + self.telemetry.emit( + "tension_end", + move_motors=move_motors, + stopped=self._task_stop_event.is_set(), + ) def _run_task(self, task_fn, *args, **kwargs): with self._lock: @@ -1304,6 +1366,7 @@ def __init__( model_version: str | None = None, model_name: str | None = None, config: OrcaHandTouchConfig | None = None, + telemetry_enabled: bool = True, ): super().__init__( config_path=config_path, @@ -1311,6 +1374,7 @@ def __init__( model_version=model_version, model_name=model_name, config=config, + telemetry_enabled=telemetry_enabled, ) self._tactile_client = None @@ -1377,6 +1441,7 @@ def connect(self) -> tuple[bool, str]: return success, msg sensor_ok, sensor_msg = self._connect_sensor_with_fallback() + self.telemetry.emit("sensor_connect", success=sensor_ok) if not sensor_ok: return False, f"{msg} | {sensor_msg}" return True, f"{msg} | {sensor_msg}" @@ -1388,7 +1453,9 @@ def connect_sensors_only(self) -> tuple[bool, str]: powered. After this call, tactile methods work; motor-control methods will fail because the motor client is not initialised. """ - return self._connect_sensor_with_fallback() + success, msg = self._connect_sensor_with_fallback() + self.telemetry.emit("sensor_connect", success=success, sensors_only=True) + return success, msg def disconnect(self) -> None: if self._tactile_client is not None and self._tactile_client.is_connected: @@ -1464,6 +1531,10 @@ class MockOrcaHand(OrcaHand): port is opened and motor state is simulated in memory. """ + def __init__(self, *args, **kwargs): + kwargs["telemetry_enabled"] = False + super().__init__(*args, **kwargs) + def _resolve_port(self) -> str: return self.config.port or "/dev/null" @@ -1487,6 +1558,10 @@ def _create_motor_client( class MockOrcaHandTouch(OrcaHandTouch): """Drop-in :class:`OrcaHandTouch` with in-memory mock motor + sensor clients (no serial I/O).""" + def __init__(self, *args, **kwargs): + kwargs["telemetry_enabled"] = False + super().__init__(*args, **kwargs) + def _resolve_port(self) -> str: return self.config.port or "/dev/null" diff --git a/orca_core/telemetry.py b/orca_core/telemetry.py new file mode 100644 index 0000000..0600447 --- /dev/null +++ b/orca_core/telemetry.py @@ -0,0 +1,159 @@ +from __future__ import annotations + +import argparse +import json +import threading +import time +import urllib.request +import uuid +from datetime import datetime, timezone +from pathlib import Path +from queue import Empty, Queue +from typing import Any + + +state_dir = Path.home() / ".config" / "orca_core" + + +class Telemetry: + """Small event logger owned by one hand instance.""" + + endpoint = "https://orcahand.com/api/telemetry" + # Send at most this many events in one HTTP request. + batch_size = 100 + # Keep at most this many unsent events locally; drop new events after that. + queue_size = 2000 + # Send queued events after this delay, unless a batch fills sooner. + post_period_s = 5 + + def __init__( + self, + *, + enabled: bool = True, + sample_hz: float = 5, + ) -> None: + self.enabled = enabled and opted_in() + if not self.enabled: + return + + self.session_id = uuid.uuid4().hex + self.install_id = install_id() + # Limit high-frequency sampled events, such as joint telemetry, to 5 Hz by default. + self.sample_interval_s = 1 / sample_hz + self.last_sample: dict[str, float] = {} + self.queue: Queue[dict[str, Any]] = Queue(maxsize=self.queue_size) + self.thread = threading.Thread( + target=self.send_loop, + name="orca-core-telemetry", + daemon=True, + ) + self.thread.start() + + def emit(self, event_name: str, **payload: Any) -> None: + """Queue an event. Never blocks and never raises.""" + if not self.enabled: + return + + try: + self.queue.put_nowait( + { + "install_id": self.install_id, + "session_id": self.session_id, + "event_name": str(event_name), + "occurred_at": datetime.now(timezone.utc).isoformat(), + "payload": payload, + } + ) + except Exception: + pass + + def sample(self, event_name: str, **payload: Any) -> None: + """Like emit(), but rate-limited to 5 Hz per event name.""" + if not self.enabled: + return + + now = time.monotonic() + if now - self.last_sample.get(event_name, 0.0) < self.sample_interval_s: + return + self.last_sample[event_name] = now + self.emit(event_name, **payload) + + def send_loop(self) -> None: + while True: + rows = self.drain() + try: + http_post(rows) + except Exception: + pass + + def drain(self) -> list[dict[str, Any]]: + rows = [self.queue.get()] + deadline = time.monotonic() + self.post_period_s + while len(rows) < self.batch_size: + timeout = max(0.0, deadline - time.monotonic()) + if timeout == 0.0: + break + try: + rows.append(self.queue.get(timeout=timeout)) + except Empty: + break + return rows + + +def opted_in() -> bool: + return not (state_dir / "telemetry_disabled").exists() + + +def disable() -> None: + state_dir.mkdir(parents=True, exist_ok=True) + (state_dir / "telemetry_disabled").touch() + + +def enable() -> None: + (state_dir / "telemetry_disabled").unlink(missing_ok=True) + + +def install_id() -> str: + try: + value = (state_dir / "install_id").read_text(encoding="utf-8").strip() + if value: + return value + except OSError: + pass + + value = uuid.uuid4().hex + try: + state_dir.mkdir(parents=True, exist_ok=True) + (state_dir / "install_id").write_text(value, encoding="utf-8") + except OSError: + pass + return value + + +def http_post(rows: list[dict[str, Any]]) -> None: + req = urllib.request.Request( + Telemetry.endpoint, + data=json.dumps({"events": rows}, separators=(",", ":")).encode("utf-8"), + headers={"Content-Type": "application/json", "User-Agent": "orca_core"}, + method="POST", + ) + urllib.request.urlopen(req, timeout=5).read() + + +def main() -> None: + parser = argparse.ArgumentParser(prog="python -m orca_core.telemetry") + parser.add_argument("command", choices=["enable", "disable", "status"]) + args = parser.parse_args() + + if args.command == "enable": + enable() + print("Telemetry enabled.") + elif args.command == "disable": + disable() + print(f"Telemetry disabled. Marker: {state_dir / 'telemetry_disabled'}") + else: + print("enabled" if opted_in() else "disabled") + + +if __name__ == "__main__": + main()