openenv/echo_env
6
1"""Initialize a new OpenEnv environment."""2 3from __future__ import annotations4 5import random6import shutil7import subprocess8from importlib import resources9from pathlib import Path10from typing import Annotated, Dict, List, Tuple11 12import typer13 14from .._cli_utils import console15 16app = typer.Typer(help="Initialize a new OpenEnv environment")17 18 19def _snake_to_pascal(snake_str: str) -> str:20 """Convert snake_case to PascalCase (e.g., 'my_env' -> 'MyEnv')."""21 return "".join(word.capitalize() for word in snake_str.split("_"))22 23 24def _get_env_prefix(env_name: str) -> str:25 """Extract the prefix for class names (e.g., 'my_env' -> 'My', 'test_env' -> 'Test')."""26 # Remove trailing '_env' if present27 if env_name.endswith("_env"):28 base = env_name[:-4] # Remove '_env'29 else:30 base = env_name31 32 # If empty or just one part, use the whole thing33 if not base or "_" not in base:34 return base.capitalize() if base else env_name.capitalize()35 36 # PascalCase all parts except the last37 parts = base.split("_")38 return "".join(word.capitalize() for word in parts)39 40 41def _snake_to_camel(snake_str: str) -> str:42 """Convert snake_case to camelCase (e.g., 'my_env' -> 'myEnv')."""43 parts = snake_str.split("_")44 return parts[0] + "".join(word.capitalize() for word in parts[1:])45 46 47def _snake_to_title(snake_str: str) -> str:48 """Convert snake_case to Title Case (e.g., 'my_env' -> 'My Env')."""49 return " ".join(word.capitalize() for word in snake_str.split("_"))50 51 52def _validate_env_name(name: str) -> str:53 """Validate environment name (must be valid Python identifier in snake_case)."""54 if not name:55 raise typer.BadParameter("Environment name cannot be empty")56 57 # Check if it's a valid Python identifier58 if not name.isidentifier():59 raise typer.BadParameter(60 f"Environment name '{name}' is not a valid Python identifier. Use snake_case (e.g., 'my_env', 'game_env')."61 )62 63 # Check if it starts with a number64 if name[0].isdigit():65 raise typer.BadParameter(66 f"Environment name '{name}' cannot start with a number."67 )68 69 return name70 71 72def _get_random_hf_space_config() -> Dict[str, str]:73 """74 Get random Hugging Face Space configuration values.75 76 Returns:77 `dict` with keys `emoji`, `colorFrom`, and `colorTo`.78 """79 # Valid emojis (emoji-only characters)80 emojis = [81 "🎮",82 "🎯",83 "🚀",84 "🌟",85 "🎨",86 "🎪",87 "🎭",88 "🎬",89 "🎤",90 "🎧",91 "🎵",92 "🎶",93 "🎸",94 "🎹",95 "🥁",96 "🎺",97 "🎻",98 "🎼",99 "🎯",100 "🎲",101 "🎳",102 "🎰",103 "🎴",104 "🃏",105 "🀄",106 "🎴",107 "🎨",108 "🖼️",109 "🎬",110 "🎭",111 "🎪",112 "🎤",113 "🎧",114 "🎵",115 "🎶",116 "🎸",117 "🎹",118 "🎺",119 "🎻",120 "🥁",121 "🎯",122 "🎲",123 "🎳",124 "🎰",125 "🏀",126 "⚽",127 "🏈",128 "⚾",129 "🎾",130 "🏐",131 "🏉",132 "🎱",133 "🏓",134 "🏸",135 "🥅",136 "🏒",137 "🏑",138 "🏏",139 "⛳",140 "🏹",141 "🎣",142 "🥊",143 "🥋",144 "🎽",145 "🏅",146 "🎖️",147 "🏆",148 "🥇",149 "🥈",150 "🥉",151 "🔊",152 "🔉",153 "🔈",154 "🔇",155 "📢",156 "📣",157 "📯",158 "🔔",159 "🔕",160 "📻",161 "📡",162 "💻",163 "🖥️",164 "🖨️",165 "⌨️",166 "🖱️",167 "🖲️",168 "🕹️",169 "🗜️",170 "💾",171 "💿",172 "📀",173 "📼",174 "📷",175 "📸",176 "📹",177 "🎥",178 "📽️",179 "🎞️",180 "📞",181 "☎️",182 "📟",183 "📠",184 "📺",185 "📻",186 "🎙️",187 "🎚️",188 "🎛️",189 "⏱️",190 "⏲️",191 "⏰",192 "🕰️",193 "⌚",194 "📱",195 "📲",196 "💻",197 "⌨️",198 "🖥️",199 "🖨️",200 "🖱️",201 ]202 203 # Valid colors from HF Spaces config reference204 colors = ["red", "yellow", "green", "blue", "indigo", "purple", "pink", "gray"]205 206 return {207 "emoji": random.choice(emojis),208 "colorFrom": random.choice(colors),209 "colorTo": random.choice(colors),210 }211 212 213def _create_template_replacements(env_name: str) -> Dict[str, str]:214 """215 Create comprehensive template replacement dictionary.216 217 Supports all naming conventions:218 - PascalCase for class names219 - camelCase for variable names220 - snake_case for module names, file paths221 """222 env_prefix = _get_env_prefix(env_name)223 env_camel = _snake_to_camel(env_name)224 env_title = _snake_to_title(env_name)225 226 # Get random HF Space config values227 hf_config = _get_random_hf_space_config()228 229 replacements = {230 # Template placeholders (MUST come first - full class names before partial)231 "__ENV_CLASS_NAME__Environment": f"{env_prefix}Environment",232 "__ENV_CLASS_NAME__Action": f"{env_prefix}Action",233 "__ENV_CLASS_NAME__Observation": f"{env_prefix}Observation",234 "__ENV_CLASS_NAME__Env": f"{env_prefix}Env",235 # Template placeholders (partial - must come after full replacements)236 "__ENV_NAME__": env_name,237 "__ENV_CLASS_NAME__": env_prefix, # Use prefix, not full PascalCase238 "__ENV_TITLE_NAME__": env_title,239 "__ENV_CAMEL_NAME__": env_camel,240 # Hugging Face Space config placeholders241 "__HF_EMOJI__": hf_config["emoji"],242 "__HF_COLOR_FROM__": hf_config["colorFrom"],243 "__HF_COLOR_TO__": hf_config["colorTo"],244 }245 246 return replacements247 248 249def _replace_in_content(content: str, replacements: Dict[str, str]) -> str:250 """Replace all occurrences in content using case-sensitive replacements."""251 result = content252 # Sort by length (longest first) to avoid partial replacements253 for old, new in sorted(replacements.items(), key=lambda x: len(x[0]), reverse=True):254 result = result.replace(old, new)255 return result256 257 258def _should_rename_file(filename: str, env_name: str) -> Tuple[bool, str]:259 """260 Check if a file should be renamed and return the new name.261 262 Handles template placeholders in filenames like:263 - `__ENV_NAME___environment.py` → `<env_name>_environment.py`264 """265 # Check for template placeholder266 if "__ENV_NAME__" in filename:267 new_name = filename.replace("__ENV_NAME__", env_name)268 return True, new_name269 270 return False, filename271 272 273def _copy_and_template_file(274 src_path: Path,275 dest_path: Path,276 replacements: Dict[str, str],277) -> None:278 """Copy a file and apply template replacements."""279 dest_path.parent.mkdir(parents=True, exist_ok=True)280 281 try:282 # Read source file283 content = src_path.read_bytes()284 285 # Try to decode as text and apply replacements286 try:287 text = content.decode("utf-8")288 # Normalize line endings to LF before applying replacements289 text = text.replace("\r\n", "\n").replace("\r", "\n")290 text = _replace_in_content(text, replacements)291 dest_path.write_text(text, encoding="utf-8", newline="\n")292 except UnicodeDecodeError:293 # Binary file, just copy294 dest_path.write_bytes(content)295 except Exception as e:296 raise RuntimeError(297 f"Failed to copy template file {src_path} to {dest_path}: {e}"298 ) from e299 300 301def _resolve_template_path(template_pkg: str, template_dir: str) -> Path:302 """Resolve the template package directory used by `openenv init`."""303 # Get the package path using importlib.resources but avoid importing the template package304 # We'll use the package's __file__ to get the directory path305 import importlib306 307 try:308 # Import the parent package (not the template package itself)309 if "." in template_pkg:310 parent_pkg = ".".join(template_pkg.split(".")[:-1])311 pkg = importlib.import_module(parent_pkg)312 template_path = Path(pkg.__file__).parent / template_pkg.split(".")[-1]313 else:314 pkg = importlib.import_module(template_pkg.split(".")[0])315 template_path = Path(pkg.__file__).parent / template_pkg.split(".")[-1]316 except Exception:317 # Fallback: try to use resources.files but handle import errors318 try:319 base = resources.files(template_pkg.split(".")[0])320 template_path = base.joinpath(*template_pkg.split(".")[1:])321 if not template_path.exists():322 raise FileNotFoundError(f"Template directory not found: {template_pkg}")323 except Exception as e:324 raise FileNotFoundError(325 f"Template directory not found: {template_pkg}"326 ) from e327 328 if template_dir:329 template_path = template_path / template_dir330 331 if not template_path.exists() or not template_path.is_dir():332 raise FileNotFoundError(333 f"Template directory not found: {template_pkg}.{template_dir}"334 )335 336 return template_path337 338 339def _copy_template_directory(340 template_pkg: str,341 template_dir: str,342 dest_dir: Path,343 replacements: Dict[str, str],344 env_name: str,345) -> List[Path]:346 """Recursively copy template directory and apply replacements."""347 created_files: List[Path] = []348 template_path = _resolve_template_path(template_pkg, template_dir)349 350 # Walk through all files in template directory using Path351 for item in template_path.rglob("*"):352 if item.is_file():353 rel_path = item.relative_to(template_path)354 dest_path = dest_dir / rel_path355 356 # Apply filename templating357 should_rename, new_name = _should_rename_file(dest_path.name, env_name)358 if should_rename:359 dest_path = dest_path.parent / new_name360 361 # Copy and apply replacements362 _copy_and_template_file(item, dest_path, replacements)363 created_files.append(dest_path)364 365 return created_files366 367 368def _generate_uv_lock(env_dir: Path) -> bool:369 """Generate uv.lock from pyproject.toml using uv."""370 pyproject_path = env_dir / "pyproject.toml"371 372 if not pyproject_path.exists():373 return False374 375 try:376 cmd = [377 "uv",378 "lock",379 "--directory",380 str(env_dir),381 ]382 383 result = subprocess.run(cmd, capture_output=True, text=True, check=True)384 385 if result.stdout:386 console.print(result.stdout)387 388 return True389 390 except subprocess.CalledProcessError as e:391 console.print(392 f"[yellow]Warning: Could not generate uv.lock: {e.stderr}[/yellow]"393 )394 return False395 except FileNotFoundError:396 console.print(397 "[yellow]Warning: 'uv' not found. Install it to generate uv.lock[/yellow]"398 )399 return False400 401 402@app.command()403def init(404 env_name: Annotated[405 str,406 typer.Argument(407 help="Name of the environment to create (snake_case, e.g., 'my_env')"408 ),409 ],410 output_dir: Annotated[411 str | None,412 typer.Option(413 "--output-dir",414 "-o",415 help="Output directory (defaults to current working directory)",416 ),417 ] = None,418) -> None:419 """420 Initialize a new OpenEnv environment.421 422 Creates a new directory with the environment name and generates all necessary423 files based on the OpenEnv template structure.424 425 Examples:426 427 ```bash428 $ openenv init my_game_env429 $ openenv init my_env --output-dir /path/to/projects430 ```431 """432 # Validate environment name433 env_name = _validate_env_name(env_name)434 435 # Determine output directory436 base_dir = Path(output_dir).resolve() if output_dir else Path.cwd().resolve()437 env_dir = base_dir / env_name438 439 # Check if directory already exists440 if env_dir.exists():441 if env_dir.is_file():442 raise typer.BadParameter(f"Path '{env_dir}' exists and is a file")443 if any(env_dir.iterdir()):444 raise typer.BadParameter(445 f"Directory '{env_dir}' already exists and is not empty. "446 "Please choose a different name or remove the existing directory."447 )448 449 try:450 # Create template replacements451 replacements = _create_template_replacements(env_name)452 453 # Create environment directory454 env_dir.mkdir(parents=True, exist_ok=True)455 456 console.print(457 f"[bold cyan]Creating OpenEnv environment '{env_name}'...[/bold cyan]"458 )459 460 # Copy template files from template structure461 template_pkg = "openenv.cli.templates.openenv_env"462 created_files = _copy_template_directory(463 template_pkg,464 "",465 env_dir,466 replacements,467 env_name,468 )469 470 console.print(f"[bold green]✓[/bold green] Created {len(created_files)} files")471 472 # Generate uv.lock473 console.print("\n[bold]Generating uv.lock...[/bold]")474 if _generate_uv_lock(env_dir):475 console.print("[green]✓[/green] Generated uv.lock")476 else:477 console.print("[yellow]⚠[/yellow] Could not generate uv.lock automatically")478 console.print(" You can generate it manually with:")479 console.print(f" cd {env_dir} && uv lock")480 481 console.print(482 f"\n[bold green]Environment created successfully at: {env_dir}[/bold green]"483 )484 console.print("\n[bold]Next steps:[/bold]")485 console.print(f" cd {env_dir}")486 console.print(487 f" # Edit your environment implementation in server/{env_name}_environment.py"488 )489 console.print(" # Edit your models in models.py")490 console.print(" # Install dependencies: uv sync")491 console.print("\n # To integrate into OpenEnv repo:")492 console.print(f" # 1. Copy this directory to <repo_root>/envs/{env_name}_env")493 console.print(494 f" # 2. Build from repo root: docker build -t {env_name}_env:latest -f envs/{env_name}_env/server/Dockerfile ."495 )496 console.print(497 f" # 3. Run your image: docker run -p 8000:8000 {env_name}_env:latest"498 )499 500 except Exception as e:501 # Cleanup on error502 if env_dir.exists() and env_dir.is_dir():503 try:504 shutil.rmtree(env_dir)505 except Exception:506 pass507 508 console.print(f"[bold red]Error:[/bold red] {e}")509 raise typer.Exit(1) from e510