analyzer/app/main.py - master - Sturon/Diplom

analyzer/app/main.py
120 строк4 KB
Sturon
Initial
03 июн 2026, 15:01
03 июн 2026, 15:0117b44a2
import json
from pathlib import Path

from fastapi import FastAPI, File, Form, HTTPException, UploadFile
from fastapi.responses import FileResponse
from fastapi.staticfiles import StaticFiles

from app.analyzer import evaluate, fetch_snapshot, read_image
from app.models import EvaluationParameters, EvaluationResponse
from app.runtime import runtime
from app.runtime_models import (
    AnalyzerConfig,
    AnalyzerStateResponse,
    CameraCreateRequest,
    SettingsRequest,
)

app = FastAPI(
    title="Diplom Edge Analyzer",
    version="0.1.0",
    description="Local inspection service for camera-side devices.",
)
app.mount("/static", StaticFiles(directory=Path(__file__).parent / "static"), name="static")


@app.on_event("startup")
def start_runtime() -> None:
    runtime.start()


@app.on_event("shutdown")
def stop_runtime() -> None:
    runtime.stop()


@app.get("/")
def index() -> FileResponse:
    return FileResponse(Path(__file__).parent / "static" / "index.html")


@app.get("/health")
def health() -> dict[str, str]:
    return {"status": "ok"}


@app.get("/api/app/state", response_model=AnalyzerStateResponse, response_model_by_alias=True)
def get_state() -> AnalyzerStateResponse:
    return runtime.snapshot()


@app.post("/api/app/settings", response_model=AnalyzerConfig, response_model_by_alias=True)
def update_settings(request: SettingsRequest) -> AnalyzerConfig:
    return runtime.update_settings(request.backend_url, request.watch_dir)


@app.post("/api/app/cameras")
def create_camera(request: CameraCreateRequest) -> dict:
    if not request.name.strip():
        raise HTTPException(status_code=400, detail="Camera name is required")
    try:
        return runtime.create_camera(request.name)
    except Exception as exc:
        raise HTTPException(status_code=502, detail=str(exc)) from exc


@app.post("/api/app/refresh-profile")
def refresh_profile() -> dict[str, str]:
    runtime.refresh_profile_once()
    return {"status": "ok"}


@app.get("/api/app/last-image")
def last_image() -> FileResponse:
    snapshot = runtime.snapshot()
    if snapshot.last_image_url is None or runtime.last_image_path is None:
        raise HTTPException(status_code=404, detail="No image has been processed yet")
    return FileResponse(runtime.last_image_path)


@app.post("/api/v1/evaluate", response_model=EvaluationResponse, response_model_by_alias=True)
async def evaluate_frame(
    parameters: str = Form(...),
    image: UploadFile | None = File(default=None),
) -> EvaluationResponse:
    request = parse_parameters(parameters)
    image_bytes = await resolve_image_bytes(request, image)

    try:
        frame = read_image(image_bytes)
        return evaluate(frame, request)
    except ValueError as exc:
        raise HTTPException(status_code=400, detail=str(exc)) from exc


def parse_parameters(raw: str) -> EvaluationParameters:
    try:
        return EvaluationParameters(**json.loads(raw))
    except json.JSONDecodeError as exc:
        raise HTTPException(status_code=400, detail="parameters must be a JSON object") from exc
    except Exception as exc:
        raise HTTPException(status_code=422, detail=str(exc)) from exc


async def resolve_image_bytes(
    parameters: EvaluationParameters,
    image: UploadFile | None,
) -> bytes:
    if image is not None:
        image_bytes = await image.read()
        if not image_bytes:
            raise HTTPException(status_code=400, detail="image file is empty")
        return image_bytes

    if parameters.camera and parameters.camera.snapshot_url:
        try:
            return fetch_snapshot(parameters.camera.snapshot_url, parameters.camera.timeout_ms)
        except Exception as exc:
            raise HTTPException(status_code=502, detail="failed to fetch camera snapshot") from exc

    raise HTTPException(status_code=400, detail="image file or camera.snapshotUrl is required")