add: docstrings

Author: T-K-233

actuator_control/api.py

@@ -26,7 +26,31 @@ class ActuatorState:
 
 
 class BusBase:
-    """Shared high-level wrapper behavior for Python bus objects."""
+    """Shared high-level wrapper behavior for Python bus objects.
+
+    Args:
+        channel: CAN interface name, e.g. `can0`.
+        actuators: Actuators keyed by actuator name.
+        calibration: Optional per-actuator calibration overrides.
+        bitrate: CAN bitrate in bits per second.
+
+    We follow the LeRobot style actuator calibration logic, where the `calibration` is a
+    dictionary of actuators with their rotation direction and homing offset.
+
+    An example would be:
+    ```python
+    calibration = {
+        "actuator_1": {
+            "direction": 1.0,
+            "homing_offset": 3.14,
+        },
+        "actuator_2": {
+            "direction": -1.0,
+            "homing_offset": -0.1,
+        },
+    }
+    ```
+    """
 
     def __init__(
         self,
@@ -35,34 +59,52 @@ def __init__(
         calibration: dict[str, dict[str, Any]] | None,
         bitrate: int,
     ) -> None:
+        """Store Python-side bus configuration before the backend is created."""
         self.channel = channel
         self.actuators = dict(actuators)
         self.calibration = calibration.copy() if calibration else {}
         self.bitrate = bitrate
         self._is_connected = False
 
     def __len__(self) -> int:
+        """Return the number of configured actuators."""
         return len(self.actuators)
 
     @property
     def is_connected(self) -> bool:
+        """Whether the Python wrapper currently considers the backend connected."""
         return self._is_connected
 
     def connect(self) -> None:
+        """Open the CAN backend and start its receiver thread."""
         self._core.connect()
         self._is_connected = True
 
     def disconnect(self, disable_torque: bool = True) -> None:
+        """Disconnect the backend and optionally disable torque first.
+
+        Args:
+            disable_torque: Whether to issue per-actuator disable commands before closing.
+        """
         self._core.disconnect(disable_torque=disable_torque)
         self._is_connected = False
 
     def enable(self, actuator: str) -> None:
+        """Enable the specified actuator."""
         self._core.enable(actuator)
 
     def disable(self, actuator: str) -> None:
+        """Disable the specified actuator."""
         self._core.disable(actuator)
 
     def write_mit_kp_kd(self, actuator: str, kp: float, kd: float) -> None:
+        """Update MIT-mode proportional and derivative gains.
+
+        Args:
+            actuator: Actuator name.
+            kp: Proportional gain in backend-specific MIT units.
+            kd: Derivative gain in backend-specific MIT units.
+        """
         self._core.write_mit_kp_kd(actuator, kp, kd)
 
     def write_mit_control(
@@ -72,20 +114,39 @@ def write_mit_control(
         velocity: float = 0.0,
         torque: float = 0.0,
     ) -> None:
+        """Send one MIT control command.
+
+        Args:
+            actuator: Actuator name.
+            position: Target output position in radians.
+            velocity: Target output velocity in radians per second.
+            torque: Feedforward output torque in newton-meters.
+        """
         self._core.write_mit_control(actuator, position, velocity, torque)
 
     def request_state(self, actuator: str) -> None:
+        """Trigger an asynchronous state refresh for the specified actuator."""
         self._core.request_state(actuator)
 
     @property
     def tx_counter(self) -> int:
+        """Return the number of CAN frames transmitted by the backend."""
         return self._core.read_tx_counter()
 
     @property
     def rx_counter(self) -> int:
+        """Return the number of CAN frames received by the backend."""
         return self._core.read_rx_counter()
 
     def get_state(self, actuator: str) -> ActuatorState | None:
+        """Return the last cached actuator state.
+
+        Args:
+            actuator: Actuator name.
+
+        Returns:
+            The cached state, or ``None`` if no feedback has been received yet.
+        """
         state = self._core.read_state(actuator)
         if state is None:
             return None
@@ -100,30 +161,71 @@ def get_state(self, actuator: str) -> ActuatorState | None:
         )
 
     def get_fault_status(self, actuator: str) -> tuple[str, ...] | None:
+        """Return the cached fault labels for the specified actuator."""
         state = self.get_state(actuator)
         if state is None:
             return None
         return state.faults
 
     def get_mit_state(self, actuator: str) -> tuple[float, float, float] | None:
+        """Return cached MIT feedback as `(position, velocity, torque)`.
+
+        Args:
+            actuator: Actuator name.
+
+        Returns:
+            The cached state, or `None` if no feedback has been received yet.
+        """
         state = self.get_state(actuator)
         if state is None:
             return None
         return state.position, state.velocity, state.torque
 
     def get_position(self, actuator: str) -> float | None:
+        """Return the cached output position in radians.
+
+        Args:
+            actuator: Actuator name.
+
+        Returns:
+            The cached position, or `None` if no feedback has been received yet.
+        """
         state = self.get_state(actuator)
         return state.position if state is not None else None
 
     def get_velocity(self, actuator: str) -> float | None:
+        """Return the cached output velocity in radians per second.
+
+        Args:
+            actuator: Actuator name.
+
+        Returns:
+            The cached velocity, or `None` if no feedback has been received yet.
+        """
         state = self.get_state(actuator)
         return state.velocity if state is not None else None
 
     def get_torque(self, actuator: str) -> float | None:
+        """Return the cached output torque in newton-meters.
+
+        Args:
+            actuator: Actuator name.
+
+        Returns:
+            The cached torque, or `None` if no feedback has been received yet.
+        """
         state = self.get_state(actuator)
         return state.torque if state is not None else None
 
     def get_temperature(self, actuator: str) -> float | None:
+        """Return the cached actuator temperature in degrees Celsius.
+
+        Args:
+            actuator: Actuator name.
+
+        Returns:
+            The cached temperature, or `None` if no feedback has been received yet.
+        """
         state = self.get_state(actuator)
         return state.temperature if state is not None else None
 
@@ -134,6 +236,16 @@ def ping_by_id(
         device_id: int,
         timeout: float = 0.1,
     ) -> tuple[int, bytes] | None:
+        """Probe one device ID on one channel.
+
+        Args:
+            channel: CAN interface name.
+            device_id: Device ID to probe.
+            timeout: Receive timeout in seconds.
+
+        Returns:
+            Backend-specific response metadata and payload, or `None` if no device replied.
+        """
         raise NotImplementedError(f"{cls.__name__}.ping_by_id() is not implemented yet.")
 
     @classmethod
@@ -144,7 +256,17 @@ def scan_channel(
         end_id: int = 255,
         timeout: float = 0.1,
     ) -> dict[int, tuple[int, bytes]]:
-        """Probe one channel and return scan responses keyed by device ID."""
+        """Probe a channel and collect responses keyed by device ID.
+
+        Args:
+            channel: CAN interface name.
+            start_id: First device ID to probe, inclusive.
+            end_id: Last device ID to probe, inclusive.
+            timeout: Receive timeout per probe in seconds.
+
+        Returns:
+            Mapping from detected device ID to the raw response tuple returned by `ping_by_id`.
+        """
         _validate_scan_range(start_id, end_id)
 
         device_ids: dict[int, tuple[int, bytes]] = {}

actuator_control/erob/bus.py

@@ -10,6 +10,8 @@
 
 
 class ERobBus(BusBase):
+    """High-level Python wrapper for the eRob CAN backend."""
+
     protocol = EROB_PROTOCOL
 
     def __init__(
@@ -19,6 +21,14 @@ def __init__(
         calibration: dict[str, dict[str, Any]] | None = None,
         bitrate: int = 1_000_000,
     ) -> None:
+        """Create an eRob bus wrapper.
+
+        Args:
+            channel: CAN interface name, e.g. `can0`.
+            actuators: Actuators keyed by actuator name.
+            calibration: Optional per-actuator calibration overrides.
+            bitrate: CAN bitrate in bits per second.
+        """
         super().__init__(channel, actuators, calibration, bitrate)
         self._core = _ErobBus(
             channel=channel,
@@ -28,7 +38,9 @@ def __init__(
         )
 
     def read(self, actuator: str, parameter: int) -> int:
+        """Read one eRob parameter as an integer value."""
         return self._core.read(actuator, int(parameter))
 
     def write(self, actuator: str, parameter: int, value: int) -> None:
+        """Write one integer eRob parameter."""
         self._core.write(actuator, int(parameter), int(value))

actuator_control/erob/protocol.py

@@ -6,12 +6,16 @@
 
 
 class ErobCommandType:
+    """eRob command byte constants."""
+
     START_MOTION = 0x83
     STOP_MOTION = 0x84
     SAVE_PARAMS = 0xE8
 
 
 class ErobParameterType:
+    """eRob parameter ID constants used by `ERobBus.read` and `write`."""
+
     ACTUAL_POSITION = 0x0002
     ACTUAL_SPEED = 0x0005
     MOTOR_CURRENT = 0x0008
@@ -34,6 +38,8 @@ class ErobParameterType:
 
 
 class ErobMode:
+    """eRob control mode constants."""
+
     TORQUE = 1
     SPEED = 2
     POSITION = 3

actuator_control/protocol.py

@@ -8,7 +8,12 @@
 
 @dataclass(frozen=True, slots=True)
 class ActuatorProtocol:
-    """Static protocol metadata for one actuator family."""
+    """Static protocol metadata for one actuator family.
+
+    Attributes:
+        name: Stable protocol name used in the Python API.
+        parameter_data_types: Mapping from parameter ID to ``"int"`` or ``"float"``.
+    """
 
     name: str
     parameter_data_types: dict[int, str]
@@ -59,7 +64,15 @@ def get(self, key: str, default: ActuatorProtocol | None = None) -> ActuatorProt
 
 
 def get_parameter_data_type(actuator_type: str, parameter: int) -> str | None:
-    """Return the declared Python-side type for a protocol parameter."""
+    """Return the declared Python-side type for a protocol parameter.
+
+    Args:
+        actuator_type: Protocol name such as ``"robstride"``.
+        parameter: Backend parameter ID.
+
+    Returns:
+        ``"int"`` or ``"float"`` when the parameter is known, otherwise ``None``.
+    """
 
     protocol = ACTUATOR_PROTOCOLS.get(actuator_type)
     if protocol is None: