import asyncio
import logging
import contextlib
from fastapi import FastAPI, Request, WebSocket, WebSocketDisconnect
from fastapi.responses import HTMLResponse
import uuid
import html
logger = logging.getLogger(__name__)
class JSExecutionContext:
def __init__(self, endpoint: str, result_future: asyncio.Future):
self.endpoint = endpoint
self.result_future = result_future
self.hidden_iframe_name = "hidden-iframe"
class JSExecutionRouter:
def __init__(self):
self.app = None
self.active_routes = set()
def init_app(self, app: FastAPI):
self.app = app
def get_eval_js_html(self, js_code: str, result_id: str):
# escaped_js = html.escape(js_code)
logger.info(f"Generating HTML for JavaScript execution. Result ID: {result_id}")
return f"""
Execute JavaScript
"""
@contextlib.asynccontextmanager
async def create_dynamic_endpoint(self, js_code: str):
if not self.app:
raise RuntimeError("DynamicRouter not initialized with FastAPI app")
result_id = str(uuid.uuid4())
result_future = asyncio.Future()
@self.app.get(f"/eval-js/{result_id}", response_class=HTMLResponse)
async def serve_eval_js_page():
logger.info(f"Serving eval-js page for result ID: {result_id}")
return HTMLResponse(content=self.get_eval_js_html(js_code, result_id))
@self.app.post(f"/js-result/{result_id}")
async def receive_js_result(data: dict):
logger.info(f"Received result for ID: {result_id}")
if 'error' in data:
result_future.set_exception(Exception(data['error']))
else:
result_future.set_result(data['result'])
return {"status": "received"}
self.active_routes.add(result_id)
try:
yield JSExecutionContext(f"/eval-js/{result_id}", result_future)
finally:
self.cleanup_route(result_id)
def cleanup_route(self, result_id: str):
if result_id in self.active_routes:
self.app.router.routes = [route for route in self.app.router.routes if route.path not in [f"/eval-js/{result_id}", f"/js-result/{result_id}"]]
self.active_routes.remove(result_id)
js_execution_router = JSExecutionRouter()
def add_js_execution_routes(app: FastAPI):
js_execution_router.init_app(app)
@app.websocket("/ws/camera")
async def websocket_endpoint(websocket: WebSocket):
await websocket.accept()
try:
while True:
data = await websocket.receive_text()
await broadcast_camera_frame(data)
except WebSocketDisconnect:
pass
except Exception as e:
logger.error(f"WebSocket error: {str(e)}")
# Add this new function to handle broadcasting
async def broadcast_camera_frame(frame_data: str):
# This function will be called from the Flet app to set the callback
if hasattr(broadcast_camera_frame, 'callback'):
await broadcast_camera_frame.callback(frame_data)
# Function to set the callback from the Flet app
def set_camera_frame_callback(callback):
broadcast_camera_frame.callback = callback
def mount_flet_app(app: FastAPI, route: str, flet_app):
# Mount the Flet app
app.mount("/flet-app-internal", flet_app)
app.flet_app = flet_app # Store reference to the Flet app
@app.get(route, response_class=HTMLResponse)
async def serve_main_page(request: Request):
html_content = f"""
Camera Stream App
"""
return HTMLResponse(content=html_content)
# Add the JS execution routes
add_js_execution_routes(app)