Add telemetry client

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'
 ```
-

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,13 +1366,15 @@ 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,
             calibration_path=calibration_path,
             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"
 

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()