"""Dev command handler for script runner. Starts development servers using configuration from infrastructure/ports.yaml. """ import argparse import subprocess import sys from pathlib import Path from service_config import get_service_config, list_services def build_uvicorn_cmd(cfg: dict) -> list[str]: """Build uvicorn development command.""" return [ "uvicorn", cfg["app"], "--host", "0.0.0.0", "--port", str(cfg["port"]), "--reload", "--log-level", cfg["log_level"], ] def dev_command(args: list[str], workspace_root: Path) -> int: """Start development servers. Args: args: Command-line arguments workspace_root: Path to workspace root Returns: Exit code (0 = success, non-zero = failure) """ services = list_services("dev") parser = argparse.ArgumentParser( prog="./run dev", description="Start development servers", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=f""" Examples: ./run dev diffusion # Start diffusion dev server ./run dev classifier # Start classifier service ./run dev --list # List all available services Available services: classifier Request classification (port 8001) diffusion SDXL image generation (port 8002) [GPU] prompt LLM prompt enhancement (port 8003) [GPU] processing Image post-processing (port 8005) [TypeScript] aesthetic Aesthetic scoring (port 8006) [GPU] semantic Semantic analysis (port 8007) [GPU] moderator Content moderation (port 8008) identity Identity recognition (port 8009) [GPU] Note: Services run in foreground. Use Ctrl+C to stop. """, ) parser.add_argument( "service", nargs="?", choices=services, help="Service to start", ) parser.add_argument( "--list", action="store_true", help="List available services", ) parsed = parser.parse_args(args) # List services if parsed.list: print("Development services:\n") for svc_id in services: cfg = get_service_config(svc_id, "dev") svc_type = cfg["type"] print(f" {svc_id:15} {svc_type:10} port {cfg['port']}") return 0 # Require service argument if not parsed.service: parser.print_help() return 1 cfg = get_service_config(parsed.service, "dev") service_dir = workspace_root / cfg["dir"] if not service_dir.exists(): print(f"Error: Service directory not found: {service_dir}", file=sys.stderr) return 1 print(f"Starting {parsed.service} on port {cfg['port']}...") print(f"Directory: {service_dir}") print() # Build command based on service type if cfg["type"] == "python": cmd = build_uvicorn_cmd(cfg) # Check for venv venv_path = service_dir / ".venv" if not venv_path.exists(): print("Warning: No .venv found. You may need to create one:") print(f" cd {service_dir}") print(" python -m venv .venv") print(" source .venv/bin/activate") print(" pip install -e .") print() # For Python services, try to activate venv activate_script = venv_path / "bin" / "activate" if activate_script.exists(): full_cmd = f"source {activate_script} && {' '.join(cmd)}" cmd = ["bash", "-c", full_cmd] print(f"Command: uvicorn {cfg['app']} --reload") else: # TypeScript (NestJS) cmd = ["pnpm", "start:dev"] print(f"Command: pnpm start:dev") print() print("Press Ctrl+C to stop") print("-" * 50) print() # Run service try: result = subprocess.run(cmd, cwd=service_dir, check=False) return result.returncode except KeyboardInterrupt: print("\n\nStopped by user") return 0 def register_dev_command(runner): """Register the dev command with the script runner. Args: runner: ScriptRunner instance """ runner.register_command( "dev", dev_command, "Start development servers", )