openenv/coding_env
21
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 Dictionary with 'emoji', 'colorFrom', and 'colorTo' keys78 """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 _copy_template_directory(302 template_pkg: str,303 template_dir: str,304 dest_dir: Path,305 replacements: Dict[str, str],306 env_name: str,307) -> List[Path]:308 """Recursively copy template directory and apply replacements."""309 created_files: List[Path] = []310 311 # Get the package path using importlib.resources but avoid importing the template package312 # We'll use the package's __file__ to get the directory path313 import importlib314 315 try:316 # Import the parent package (not the template package itself)317 if "." in template_pkg:318 parent_pkg = ".".join(template_pkg.split(".")[:-1])319 pkg = importlib.import_module(parent_pkg)320 template_path = Path(pkg.__file__).parent / template_pkg.split(".")[-1]321 else:322 pkg = importlib.import_module(template_pkg.split(".")[0])323 template_path = Path(pkg.__file__).parent / template_pkg.split(".")[-1]324 except Exception:325 # Fallback: try to use resources.files but handle import errors326 try:327 base = resources.files(template_pkg.split(".")[0])328 template_path = base.joinpath(*template_pkg.split(".")[1:])329 if not template_path.exists():330 raise FileNotFoundError(f"Template directory not found: {template_pkg}")331 except Exception as e:332 raise FileNotFoundError(333 f"Template directory not found: {template_pkg}"334 ) from e335 336 if template_dir:337 template_path = template_path / template_dir338 339 if not template_path.exists() or not template_path.is_dir():340 raise FileNotFoundError(341 f"Template directory not found: {template_pkg}.{template_dir}"342 )343 344 # Walk through all files in template directory using Path345 for item in template_path.rglob("*"):346 if item.is_file():347 rel_path = item.relative_to(template_path)348 dest_path = dest_dir / rel_path349 350 # Apply filename templating351 should_rename, new_name = _should_rename_file(dest_path.name, env_name)352 if should_rename:353 dest_path = dest_path.parent / new_name354 355 # Copy and apply replacements356 _copy_and_template_file(item, dest_path, replacements)357 created_files.append(dest_path)358 359 return created_files360 361 362def _generate_uv_lock(env_dir: Path) -> bool:363 """Generate uv.lock from pyproject.toml using uv."""364 pyproject_path = env_dir / "pyproject.toml"365 366 if not pyproject_path.exists():367 return False368 369 try:370 cmd = [371 "uv",372 "lock",373 "--directory",374 str(env_dir),375 ]376 377 result = subprocess.run(cmd, capture_output=True, text=True, check=True)378 379 if result.stdout:380 console.print(result.stdout)381 382 return True383 384 except subprocess.CalledProcessError as e:385 console.print(386 f"[yellow]Warning: Could not generate uv.lock: {e.stderr}[/yellow]"387 )388 return False389 except FileNotFoundError:390 console.print(391 "[yellow]Warning: 'uv' not found. Install it to generate uv.lock[/yellow]"392 )393 return False394 395 396@app.command()397def init(398 env_name: Annotated[399 str,400 typer.Argument(401 help="Name of the environment to create (snake_case, e.g., 'my_env')"402 ),403 ],404 output_dir: Annotated[405 str | None,406 typer.Option(407 "--output-dir",408 "-o",409 help="Output directory (defaults to current working directory)",410 ),411 ] = None,412) -> None:413 """414 Initialize a new OpenEnv environment.415 416 Creates a new directory with the environment name and generates all necessary417 files based on the OpenEnv template structure.418 419 Example:420 $ openenv init my_game_env421 $ openenv init my_env --output-dir /path/to/projects422 """423 # Validate environment name424 env_name = _validate_env_name(env_name)425 426 # Determine output directory427 base_dir = Path(output_dir).resolve() if output_dir else Path.cwd().resolve()428 env_dir = base_dir / env_name429 430 # Check if directory already exists431 if env_dir.exists():432 if env_dir.is_file():433 raise typer.BadParameter(f"Path '{env_dir}' exists and is a file")434 if any(env_dir.iterdir()):435 raise typer.BadParameter(436 f"Directory '{env_dir}' already exists and is not empty. "437 "Please choose a different name or remove the existing directory."438 )439 440 try:441 # Create template replacements442 replacements = _create_template_replacements(env_name)443 444 # Create environment directory445 env_dir.mkdir(parents=True, exist_ok=True)446 447 console.print(448 f"[bold cyan]Creating OpenEnv environment '{env_name}'...[/bold cyan]"449 )450 451 # Copy template files from template structure452 template_pkg = "openenv.cli.templates.openenv_env"453 created_files = _copy_template_directory(454 template_pkg,455 "",456 env_dir,457 replacements,458 env_name,459 )460 461 console.print(f"[bold green]✓[/bold green] Created {len(created_files)} files")462 463 # Generate uv.lock464 console.print("\n[bold]Generating uv.lock...[/bold]")465 if _generate_uv_lock(env_dir):466 console.print("[green]✓[/green] Generated uv.lock")467 else:468 console.print("[yellow]⚠[/yellow] Could not generate uv.lock automatically")469 console.print(" You can generate it manually with:")470 console.print(f" cd {env_dir} && uv lock")471 472 console.print(473 f"\n[bold green]Environment created successfully at: {env_dir}[/bold green]"474 )475 console.print("\n[bold]Next steps:[/bold]")476 console.print(f" cd {env_dir}")477 console.print(478 f" # Edit your environment implementation in server/{env_name}_environment.py"479 )480 console.print(" # Edit your models in models.py")481 console.print(" # Install dependencies: uv sync")482 console.print("\n # To integrate into OpenEnv repo:")483 console.print(f" # 1. Copy this directory to <repo_root>/envs/{env_name}_env")484 console.print(485 f" # 2. Build from repo root: docker build -t {env_name}_env:latest -f envs/{env_name}_env/server/Dockerfile ."486 )487 console.print(488 f" # 3. Run your image: docker run -p 8000:8000 {env_name}_env:latest"489 )490 491 except Exception as e:492 # Cleanup on error493 if env_dir.exists() and env_dir.is_dir():494 try:495 shutil.rmtree(env_dir)496 except Exception:497 pass498 499 console.print(f"[bold red]Error:[/bold red] {e}")500 raise typer.Exit(1) from e501