chore(dummy): add comments + better visualization

lerobot/common/robots/dummy/configuration_dummy.py

@@ -0,0 +1,31 @@
+# Copyright 2024 The HuggingFace Inc. team. All rights reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from dataclasses import dataclass, field
+
+from lerobot.common.cameras.configs import CameraConfig
+from lerobot.common.cameras.opencv.configuration_opencv import OpenCVCameraConfig
+from lerobot.common.robots.config import RobotConfig
+
+
+@RobotConfig.register_subclass("dummy")
+@dataclass
+class DummyConfig(RobotConfig):
+    id = "dummy"
+
+    cameras: dict[str, CameraConfig] = field(
+        default_factory=lambda: {
+            "cam": OpenCVCameraConfig(camera_index=0, fps=30, width=1280, height=720),
+        }
+    )

lerobot/common/robots/dummy/dummy.py

@@ -0,0 +1,101 @@
+# Copyright 2024 The HuggingFace Inc. team. All rights reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+
+import logging
+
+import numpy as np
+
+from lerobot.common.cameras.utils import make_cameras_from_configs
+from lerobot.common.constants import OBS_STATE
+from lerobot.common.errors import DeviceAlreadyConnectedError, DeviceNotConnectedError
+
+from ..robot import Robot
+from .configuration_dummy import DummyConfig
+
+
+class Dummy(Robot):
+    config_class = DummyConfig
+    name = "dummy"
+
+    def __init__(self, config: DummyConfig):
+        super().__init__(config)
+        self.cameras = make_cameras_from_configs(config.cameras)
+        self.is_connected = False
+
+    @property
+    def state_feature(self) -> dict:
+        logging.warning("Dummy has nothing to send.")
+
+    @property
+    def action_feature(self) -> dict:
+        logging.warning("Dummy has nothing to send.")
+
+    @property
+    def camera_features(self) -> dict[str, dict]:
+        cam_ft = {
+            "cam": {
+                "shape": (480, 640, 3),
+                "names": ["height", "width", "channels"],
+                "info": None,
+            },
+        }
+        return cam_ft
+
+    def connect(self) -> None:
+        if self.is_connected:
+            raise DeviceAlreadyConnectedError(
+                "Dummy is already connected. Do not run `robot.connect()` twice."
+            )
+
+        logging.info("Connecting cameras.")
+        for cam in self.cameras.values():
+            cam.connect()
+
+        self.is_connected = True
+
+    def calibrate(self) -> None:
+        logging.warning("Dummy has nothing to calibrate.")
+        return
+
+    def get_observation(self) -> dict[str, np.ndarray]:
+        if not self.is_connected:
+            raise DeviceNotConnectedError("Dummy is not connected. You need to run `robot.connect()`.")
+
+        obs_dict = {}
+
+        for cam_key, cam in self.cameras.items():
+            frame = cam.async_read()
+            obs_dict[f"{OBS_STATE}.{cam_key}"] = frame
+
+        return obs_dict
+
+    def send_action(self, action: np.ndarray) -> np.ndarray:
+        logging.warning("Dummy has nothing to send.")
+
+    def print_logs(self):
+        pass
+
+    def disconnect(self):
+        if not self.is_connected:
+            raise DeviceNotConnectedError(
+                "Dummy is not connected. You need to run `robot.connect()` before disconnecting."
+            )
+        for cam in self.cameras.values():
+            cam.disconnect()
+        self.is_connected = False
+
+    def __del__(self):
+        if getattr(self, "is_connected", False):
+            self.disconnect()

lerobot/common/robots/dummy/dummy_app.py

@@ -0,0 +1,71 @@
+# Copyright 2024 The HuggingFace Inc. team. All rights reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import logging
+import time
+
+import rerun as rr
+
+from lerobot.common.robots.dummy.configuration_dummy import DummyConfig
+from lerobot.common.robots.dummy.dummy import Dummy
+from lerobot.common.teleoperators.so100 import SO100Teleop, SO100TeleopConfig
+
+
+# IMO, it's better to use rerun in the application code instead of the library code
+def main():
+    logging.info("Configuring Devices")
+    leader_arm_config = SO100TeleopConfig(port="/dev/tty.usbmodem58760434171")
+    leader_arm = SO100Teleop(leader_arm_config)
+
+    robot_config = DummyConfig()
+    robot = Dummy(robot_config)
+
+    logging.info("Connecting SO100 Devices")
+    leader_arm.connect()
+
+    logging.info("Connecting Dummy")
+    robot.connect()
+
+    rr.init("rerun_dummy_data")
+    # If data source and visualizer are in different host, use .connect() instead to establish a tcp connection
+    # We can define a custom blueprint configuration for the visualizer panels
+    # Memory limit will make sure no more than 5% of the memory is used by the visualizer
+    rr.spawn(memory_limit="5%")
+
+    logging.info("Starting...")
+    i = 0
+    while i < 10000:
+        # An alternative would be do rerun log inside of these methods. But then that means embedding rerun into the library
+        arm_action = leader_arm.get_action()
+        observation = robot.get_observation()
+
+        for j in range(arm_action.size):
+            # If you want to disable batching we can do it: export RERUN_FLUSH_NUM_BYTES=256
+            current_time = time.time()
+            rr.set_time_seconds(f"arm_action_{j}", current_time)
+            rr.log(f"arm_action_{j}", rr.Scalar(arm_action[j]))
+
+        for k, v in observation.items():
+            # This discards all previous image frames
+            rr.set_time_seconds(k, 0.0)
+            rr.log(k, rr.Image(v))
+
+        i += 1
+
+    robot.disconnect()
+    leader_arm.disconnect()
+
+
+if __name__ == "__main__":
+    main()