clean-up

env.py

@@ -24,10 +24,8 @@ import logging
 import os
 import sys
 from pathlib import Path
-from typing import Any
 
 import gymnasium as gym
-import yaml
 
 from lerobot.envs.configs import EnvConfig
 
@@ -39,39 +37,38 @@ DEFAULT_CONFIG_FILE = "configs/config.yaml"
 def _download_and_import(filename: str):
     """Download file from Hub and import as module."""
     from huggingface_hub import hf_hub_download
+
     local_path = hf_hub_download(repo_id=HUB_REPO_ID, filename=filename)
     module_dir = os.path.dirname(local_path)
-    
+
     # Add directory to sys.path so modules can import each other
     if module_dir not in sys.path:
         sys.path.insert(0, module_dir)
-    
+
     module_name = filename.replace(".py", "")
-    
+
     # Read file content and replace relative imports with absolute imports
-    with open(local_path, "r") as f:
+    with open(local_path) as f:
         content = f.read()
-    
+
     # Replace relative imports (e.g., "from .errors" -> "from errors")
     content = content.replace("from .errors import", "from errors import")
     content = content.replace("from .isaaclab_env_wrapper import", "from isaaclab_env_wrapper import")
-    
+
     # Create module and execute modified content
-    module = importlib.util.module_from_spec(
-        importlib.util.spec_from_file_location(module_name, local_path)
-    )
+    module = importlib.util.module_from_spec(importlib.util.spec_from_file_location(module_name, local_path))
     sys.modules[module_name] = module
-    
+
     # Compile and execute the modified code
     code = compile(content, local_path, "exec")
-    exec(code, module.__dict__)
-    
+    exec(code, module.__dict__)  # noqa: S102
+
     return module
 
 
 try:
-    from .isaaclab_env_wrapper import IsaacLabEnvWrapper, cleanup_isaaclab
     from .errors import IsaacLabArenaCameraKeyError, IsaacLabArenaStateKeyError
+    from .isaaclab_env_wrapper import IsaacLabEnvWrapper, cleanup_isaaclab
 except ImportError:
     _errors = _download_and_import("errors.py")
     _isaaclab_wrapper = _download_and_import("isaaclab_env_wrapper.py")
@@ -151,7 +148,7 @@ ENVIRONMENT_ALIASES: dict[str, str] = {
     "kitchen_pnp": (
         f"{ISAACLAB_ARENA_ENV_MODULE}.kitchen_pick_and_place_environment.KitchenPickAndPlaceEnvironment"
     ),
-    "press_button": (f"{ISAACLAB_ARENA_ENV_MODULE}.press_button_environment.PressButtonEnvironment"),
+    "press_button": f"{ISAACLAB_ARENA_ENV_MODULE}.press_button_environment.PressButtonEnvironment",
 }
 
 
@@ -175,8 +172,6 @@ def _find_config_file(config_path: str | Path | None = None) -> Path | None:
     Returns:
         Path to config file, or None if not found (uses defaults).
     """
-    module_dir = Path(__file__).parent
-
     search_paths = []
 
     # Explicit path from argument
@@ -204,6 +199,7 @@ def _find_config_file(config_path: str | Path | None = None) -> Path | None:
     # This is needed when env.py is loaded from HF cache (only env.py is downloaded)
     try:
         from huggingface_hub import hf_hub_download
+
         logging.info(f"Downloading config from Hub: {HUB_REPO_ID}/{DEFAULT_CONFIG_FILE}")
         hub_config_path = hf_hub_download(
             repo_id=HUB_REPO_ID,
@@ -215,6 +211,7 @@ def _find_config_file(config_path: str | Path | None = None) -> Path | None:
 
     raise FileNotFoundError("config.yaml not found and no configuration provided.")
 
+
 def _create_isaaclab_env(config: dict, n_envs: int) -> dict[str, dict[int, gym.vector.VectorEnv]]:
     """Create IsaacLab Arena environment from configuration.
 
@@ -270,9 +267,11 @@ def _create_isaaclab_env(config: dict, n_envs: int) -> dict[str, dict[int, gym.v
     camera_keys = tuple(k.strip() for k in (config.get("camera_keys") or "").split(",") if k.strip())
     try:
         validate_config(
-            raw_env, state_keys, camera_keys,
+            raw_env,
+            state_keys,
+            camera_keys,
             config.get("state_dim", 54),
-            config.get("action_dim", 36)
+            config.get("action_dim", 36),
         )
     except (IsaacLabArenaCameraKeyError, IsaacLabArenaStateKeyError, ValueError) as e:
         logging.error(f"Validation failed: {e}")
@@ -303,9 +302,8 @@ def _create_isaaclab_env(config: dict, n_envs: int) -> dict[str, dict[int, gym.v
 
 def make_env(
     n_envs: int = 1,
-    use_async_envs: bool = False,
+    use_async_envs: bool = False,  # noqa: ARG001
     cfg: EnvConfig | None = None,
-    # **kwargs: Any,
 ) -> dict[str, dict[int, gym.vector.VectorEnv]]:
     """Create IsaacLab Arena environments (EnvHub-compatible API).
 
@@ -316,7 +314,7 @@ def make_env(
     Args:
         n_envs: Number of parallel environments (default: 1).
         use_async_envs: Ignored for IsaacLab (GPU-based batched execution).
-        **kwargs: Configuration parameters passed directly. Required keys:
+        cfg: Configuration object with environment parameters. Required keys:
             - environment: Environment name or alias (e.g., "gr1_microwave")
             - headless: Whether to run headless (default: True)
             - enable_cameras: Enable camera rendering (default: False)
@@ -338,7 +336,7 @@ def make_env(
     if n_envs < 1:
         raise ValueError("`n_envs` must be at least 1")
 
-    if not hasattr(cfg, 'environment') or cfg.environment is None:
+    if not hasattr(cfg, "environment") or cfg.environment is None:
         raise ValueError(
             "No 'environment' specified. Pass it via kwargs or create "
             "configs/config.yaml with environment settings."
@@ -370,25 +368,8 @@ def make_env(
         "camera_keys": cfg.camera_keys or "",  # Pass empty string for no cameras
         "task": cfg.task,
     }
-    # else:
-    # # Fallback to YAML for local development
-    # yaml_path = _find_config_file(kwargs.get("config_path"))
-    # if yaml_path is not None:
-    #     logging.info(f"Loading config from: {yaml_path}")
-    #     with open(yaml_path) as f:
-    #         config = yaml.safe_load(f) or {}
-    # else:
-    #     config = {}
-
-    # Apply any runtime overrides (highest priority)
-    # config.update(kwargs)
-
-    logging.info(
-        f"EnvHub make_env: environment={config.get('environment')}, n_envs={n_envs}"
-    )
-    logging.info(
-        f"Config: headless={config.get('headless')}, "
-        f"enable_cameras={config.get('enable_cameras')}"
-    )
+
+    logging.info(f"EnvHub make_env: environment={config.get('environment')}, n_envs={n_envs}")
+    logging.info(f"Config: headless={config.get('headless')}, enable_cameras={config.get('enable_cameras')}")
 
     return _create_isaaclab_env(config, n_envs)