gardentech
f34817d5b6
FastAPI backend + vanilla JS frontend for LiDAR/camera startup, camera settings, and rosbag recording, unifying the previous scan_gui/scan_gui_dual/ scan_gui_triple desktop tools into one web app.
68 lines
2.3 KiB
Python
68 lines
2.3 KiB
Python
from __future__ import annotations
|
|
|
|
import logging
|
|
from contextlib import asynccontextmanager
|
|
from pathlib import Path
|
|
|
|
from fastapi import FastAPI
|
|
from fastapi.staticfiles import StaticFiles
|
|
from starlette.types import Scope
|
|
|
|
import config
|
|
from process_manager import ProcessManager
|
|
from ros_bridge import ROSBridge
|
|
|
|
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s")
|
|
logger = logging.getLogger("scan_web.app")
|
|
|
|
FRONTEND_DIR = Path(__file__).resolve().parent.parent / "frontend"
|
|
|
|
|
|
class SessionState:
|
|
def __init__(self):
|
|
self.camera_count = config.DEFAULT_CAMERA_COUNT
|
|
self.recording = False
|
|
self.recording_path: str | None = None
|
|
self.recording_started_at: float | None = None
|
|
self.camera_values: dict = {} # cam_id -> camera_params.CameraParamValues
|
|
|
|
|
|
@asynccontextmanager
|
|
async def lifespan(app: FastAPI):
|
|
app.state.pm = ProcessManager()
|
|
app.state.session = SessionState()
|
|
app.state.ros = ROSBridge(camera_count=app.state.session.camera_count)
|
|
app.state.ros.start()
|
|
logger.info("scan_web backend started (camera_count=%s)", app.state.session.camera_count)
|
|
try:
|
|
yield
|
|
finally:
|
|
logger.info("shutting down — stopping all managed processes")
|
|
app.state.pm.stop_all()
|
|
app.state.ros.stop()
|
|
|
|
|
|
app = FastAPI(title="scan_web", lifespan=lifespan)
|
|
|
|
from routers import system, camera, recording, telemetry # noqa: E402 (needs app.state types defined above)
|
|
|
|
app.include_router(system.router, prefix="/api")
|
|
app.include_router(camera.router)
|
|
app.include_router(recording.router, prefix="/api")
|
|
app.include_router(telemetry.router)
|
|
|
|
class NoCacheStaticFiles(StaticFiles):
|
|
"""Frontend is actively edited during field testing — a browser caching a
|
|
stale index.html/app.js/style.css (no Cache-Control was set before) led
|
|
to confusing "it's not applying" reports that were actually just the
|
|
prior version still running. Static assets here are small and local, so
|
|
the no-caching cost is negligible."""
|
|
|
|
async def get_response(self, path: str, scope: Scope):
|
|
response = await super().get_response(path, scope)
|
|
response.headers["Cache-Control"] = "no-store"
|
|
return response
|
|
|
|
|
|
app.mount("/", NoCacheStaticFiles(directory=str(FRONTEND_DIR), html=True), name="frontend")
|