Spaces:
Sleeping
Sleeping
| """FastAPI backend for rmscript web demo.""" | |
| import logging | |
| from pathlib import Path | |
| from typing import Any, Dict, List | |
| from fastapi import FastAPI, HTTPException | |
| from fastapi.middleware.cors import CORSMiddleware | |
| from fastapi.responses import FileResponse | |
| from fastapi.staticfiles import StaticFiles | |
| from pydantic import BaseModel | |
| from rmscript import compile_script | |
| # Configure logging | |
| logging.basicConfig(level=logging.INFO) | |
| logger = logging.getLogger(__name__) | |
| # Create FastAPI app | |
| app = FastAPI( | |
| title="RMScript Web Demo API", | |
| description="Backend API for compiling and validating rmscript code", | |
| version="1.0.0" | |
| ) | |
| # Enable CORS for local development | |
| app.add_middleware( | |
| CORSMiddleware, | |
| allow_origins=["*"], # In production, specify exact origins | |
| allow_credentials=True, | |
| allow_methods=["*"], | |
| allow_headers=["*"], | |
| ) | |
| # Set up paths for serving frontend | |
| BACKEND_DIR = Path(__file__).parent | |
| FRONTEND_DIR = BACKEND_DIR.parent / "frontend" | |
| # Mount static files (CSS, JS) | |
| if FRONTEND_DIR.exists(): | |
| app.mount("/static", StaticFiles(directory=str(FRONTEND_DIR)), name="static") | |
| class ScriptInput(BaseModel): | |
| """Input model for script compilation.""" | |
| source: str | |
| class CompilationError(BaseModel): | |
| """Compilation error model.""" | |
| line: int | |
| column: int | |
| message: str | |
| severity: str | |
| class VerifyResponse(BaseModel): | |
| """Response model for script verification.""" | |
| success: bool | |
| errors: List[Dict[str, Any]] | |
| warnings: List[Dict[str, Any]] | |
| name: str = "" | |
| description: str = "" | |
| class IRActionModel(BaseModel): | |
| """Simplified IR action model for JSON serialization.""" | |
| type: str # "action", "wait", "picture", "sound" | |
| duration: float = 0.0 | |
| # For movement actions | |
| head_pose: List[List[float]] | None = None # 4x4 matrix as nested list | |
| antennas: List[float] | None = None # [left, right] in radians | |
| body_yaw: float | None = None # radians | |
| # For sound actions | |
| sound_name: str | None = None | |
| blocking: bool = False | |
| loop: bool = False | |
| # Metadata | |
| source_line: int = 0 | |
| class CompileResponse(BaseModel): | |
| """Response model for script compilation.""" | |
| success: bool | |
| errors: List[Dict[str, Any]] | |
| warnings: List[Dict[str, Any]] | |
| name: str = "" | |
| description: str = "" | |
| ir: List[IRActionModel] | |
| async def root(): | |
| """Serve the frontend application.""" | |
| index_path = FRONTEND_DIR / "index.html" | |
| if index_path.exists(): | |
| return FileResponse(index_path) | |
| return { | |
| "name": "RMScript Web Demo API", | |
| "version": "1.0.0", | |
| "endpoints": { | |
| "/api/verify": "POST - Verify rmscript syntax and semantics", | |
| "/api/compile": "POST - Compile rmscript to IR", | |
| } | |
| } | |
| async def verify_script(input: ScriptInput): | |
| """Verify rmscript without generating IR. | |
| Args: | |
| input: Script source code | |
| Returns: | |
| Verification result with errors and warnings | |
| """ | |
| logger.info(f"Verifying script ({len(input.source)} chars)") | |
| try: | |
| result = compile_script(input.source) | |
| # Convert errors and warnings to dict format | |
| errors = [ | |
| { | |
| "line": e.line, | |
| "column": e.column, | |
| "message": e.message, | |
| "severity": e.severity | |
| } | |
| for e in result.errors | |
| ] | |
| warnings = [ | |
| { | |
| "line": w.line, | |
| "column": w.column, | |
| "message": w.message, | |
| "severity": w.severity | |
| } | |
| for w in result.warnings | |
| ] | |
| return VerifyResponse( | |
| success=result.success, | |
| errors=errors, | |
| warnings=warnings, | |
| name=result.name, | |
| description=result.description | |
| ) | |
| except Exception as e: | |
| logger.error(f"Verification error: {e}") | |
| raise HTTPException(status_code=500, detail=str(e)) | |
| async def compile_rmscript(input: ScriptInput): | |
| """Compile rmscript to intermediate representation. | |
| Args: | |
| input: Script source code | |
| Returns: | |
| Compilation result with IR | |
| """ | |
| logger.info(f"Compiling script ({len(input.source)} chars)") | |
| try: | |
| from rmscript.ir import IRAction, IRWaitAction, IRPictureAction, IRPlaySoundAction | |
| result = compile_script(input.source) | |
| # Convert errors and warnings to dict format | |
| errors = [ | |
| { | |
| "line": e.line, | |
| "column": e.column, | |
| "message": e.message, | |
| "severity": e.severity | |
| } | |
| for e in result.errors | |
| ] | |
| warnings = [ | |
| { | |
| "line": w.line, | |
| "column": w.column, | |
| "message": w.message, | |
| "severity": w.severity | |
| } | |
| for w in result.warnings | |
| ] | |
| # Convert IR to JSON-serializable format | |
| ir_json = [] | |
| for action in result.ir: | |
| if isinstance(action, IRAction): | |
| ir_json.append(IRActionModel( | |
| type="action", | |
| duration=action.duration, | |
| head_pose=action.head_pose.tolist() if action.head_pose is not None else None, | |
| antennas=action.antennas if action.antennas is not None else None, | |
| body_yaw=action.body_yaw, | |
| source_line=action.source_line | |
| )) | |
| elif isinstance(action, IRWaitAction): | |
| ir_json.append(IRActionModel( | |
| type="wait", | |
| duration=action.duration, | |
| source_line=action.source_line | |
| )) | |
| elif isinstance(action, IRPictureAction): | |
| ir_json.append(IRActionModel( | |
| type="picture", | |
| source_line=action.source_line | |
| )) | |
| elif isinstance(action, IRPlaySoundAction): | |
| ir_json.append(IRActionModel( | |
| type="sound", | |
| duration=action.duration or 0.0, | |
| sound_name=action.sound_name, | |
| blocking=action.blocking, | |
| loop=action.loop, | |
| source_line=action.source_line | |
| )) | |
| logger.info(f"Compiled {len(ir_json)} IR actions") | |
| return CompileResponse( | |
| success=result.success, | |
| errors=errors, | |
| warnings=warnings, | |
| name=result.name, | |
| description=result.description, | |
| ir=ir_json | |
| ) | |
| except Exception as e: | |
| logger.error(f"Compilation error: {e}") | |
| raise HTTPException(status_code=500, detail=str(e)) | |
| if __name__ == "__main__": | |
| import uvicorn | |
| uvicorn.run(app, host="0.0.0.0", port=8001, log_level="info") | |