feat(devices): add lazy loading for 3rd party robots cameras and teleoperators (#2123)

* feat(devices): add lazy loading for 3rd party robots cameras and teleoperators

Co-authored-by: Darko Lukić <lukicdarkoo@gmail.com>

* feat(devices): load device class based on assumptions in naming

* docs(devices): instructions for using 3rd party devices

* docs: address review feedback

* chore(docs): add example for 3rd party devices

---------

Co-authored-by: Darko Lukić <lukicdarkoo@gmail.com>
This commit is contained in:
Steven Palma
2025-10-07 17:46:22 +02:00
committed by GitHub
parent 9f32e00f90
commit bf3c8746b7
9 changed files with 255 additions and 5 deletions
+94
View File
@@ -15,6 +15,10 @@
# limitations under the License.
import importlib
import logging
import pkgutil
from typing import Any
from draccus.choice_types import ChoiceRegistry
def is_package_available(pkg_name: str, return_version: bool = False) -> tuple[bool, str] | bool:
@@ -58,3 +62,93 @@ def is_package_available(pkg_name: str, return_version: bool = False) -> tuple[b
_transformers_available = is_package_available("transformers")
def make_device_from_device_class(config: ChoiceRegistry) -> Any:
"""
Dynamically instantiates an object from its `ChoiceRegistry` configuration.
This factory uses the module path and class name from the `config` object's
type to locate and instantiate the corresponding device class (not the config).
It derives the device class name by removing a trailing 'Config' from the config
class name and tries a few candidate modules where the device implementation is
commonly located.
"""
if not isinstance(config, ChoiceRegistry):
raise ValueError(f"Config should be an instance of `ChoiceRegistry`, got {type(config)}")
config_cls = config.__class__
module_path = config_cls.__module__ # typical: lerobot_teleop_mydevice.config_mydevice
config_name = config_cls.__name__ # typical: MyDeviceConfig
# Derive device class name (strip "Config")
if not config_name.endswith("Config"):
raise ValueError(f"Config class name '{config_name}' does not end with 'Config'")
device_class_name = config_name[:-6] # typical: MyDeviceConfig -> MyDevice
# Build candidate modules to search for the device class
parts = module_path.split(".")
parent_module = ".".join(parts[:-1]) if len(parts) > 1 else module_path
candidates = [
parent_module, # typical: lerobot_teleop_mydevice
parent_module + "." + device_class_name.lower(), # typical: lerobot_teleop_mydevice.mydevice
]
# handle modules named like "config_xxx" -> try replacing that piece with "xxx"
last = parts[-1] if parts else ""
if last.startswith("config_"):
candidates.append(".".join(parts[:-1] + [last.replace("config_", "")]))
# de-duplicate while preserving order
seen: set[str] = set()
candidates = [c for c in candidates if not (c in seen or seen.add(c))]
tried: list[str] = []
for candidate in candidates:
tried.append(candidate)
try:
module = importlib.import_module(candidate)
except ImportError:
continue
if hasattr(module, device_class_name):
cls = getattr(module, device_class_name)
if callable(cls):
try:
return cls(config)
except TypeError as e:
raise TypeError(
f"Failed to instantiate '{device_class_name}' from module '{candidate}': {e}"
) from e
raise ImportError(
f"Could not locate device class '{device_class_name}' for config '{config_name}'. "
f"Tried modules: {tried}. Ensure your device class name is the config class name without "
f"'Config' and that it's importable from one of those modules."
)
def register_third_party_devices() -> None:
"""
Discover and import third-party lerobot_* plugins so they can register themselves.
Scans top-level modules on sys.path for packages starting with
'lerobot_robot_', 'lerobot_camera_' or 'lerobot_teleoperator_' and imports them.
"""
prefixes = ("lerobot_robot_", "lerobot_camera_", "lerobot_teleoperator_")
imported: list[str] = []
failed: list[str] = []
for module_info in pkgutil.iter_modules():
name = module_info.name
if name.startswith(prefixes):
try:
importlib.import_module(name)
imported.append(name)
logging.info("Imported third-party plugin: %s", name)
except Exception:
logging.exception("Could not import third-party plugin: %s", name)
failed.append(name)
logging.debug("Third-party plugin import summary: imported=%s failed=%s", imported, failed)