| from __future__ import annotations |
|
|
| import asyncio |
| import logging |
| import os |
| from contextlib import asynccontextmanager |
| from pathlib import Path |
|
|
| from fastapi import FastAPI |
| from fastapi.middleware.cors import CORSMiddleware |
|
|
| from core.config import settings |
| from core.database import SessionLocal, init_db |
| from core.seed import ensure_test_patient, seed_if_empty, seed_users |
| from core.simulator import run_simulator |
| from routers import ( |
| auth, |
| ingest, |
| jarvis, |
| jarvis_agent, |
| jarvis_chat, |
| jarvis_learning, |
| notes, |
| patients, |
| platform, |
| sensors, |
| vitals, |
| websocket, |
| wound_profiles, |
| ) |
|
|
| logging.basicConfig( |
| level=logging.DEBUG if settings.DEBUG else logging.INFO, |
| format="%(asctime)s %(levelname)s [%(name)s] %(message)s", |
| ) |
| logger = logging.getLogger("biopatch") |
|
|
|
|
| def _ensure_persistent_dirs() -> None: |
| """Make sure SQLite DB parent + LLM cache + wound photo dirs exist. |
| |
| On HF Spaces with persistent storage enabled, /data is mounted at |
| runtime and the env vars point here. On other hosts (Fly.io, local |
| docker), the same code path falls back to /app/data without extra |
| config. Either way we want the dirs to exist before init_db opens |
| the SQLite file. |
| """ |
| targets: list[Path] = [] |
| db_url = settings.DATABASE_URL |
| if db_url.startswith("sqlite:///"): |
| |
| path = db_url.split("sqlite:///", 1)[1] |
| if path.startswith("/"): |
| targets.append(Path(path).parent) |
| else: |
| targets.append(Path(path).resolve().parent) |
| for var in ("JARVIS_LLM_CACHE", "WOUND_PHOTO_DIR", "HF_HOME"): |
| value = os.environ.get(var) |
| if value: |
| targets.append(Path(value)) |
| for target in targets: |
| try: |
| target.mkdir(parents=True, exist_ok=True) |
| except PermissionError: |
| logger.warning("Cannot create %s — read-only filesystem", target) |
|
|
|
|
| @asynccontextmanager |
| async def lifespan(app: FastAPI): |
| _ensure_persistent_dirs() |
| init_db() |
| if settings.AUTO_SEED: |
| with SessionLocal() as db: |
| if settings.HARDWARE_MODE: |
| |
| |
| seed_users(db) |
| ensure_test_patient(db) |
| else: |
| seed_if_empty(db) |
|
|
| sim_task: asyncio.Task | None = None |
| |
| |
| if settings.ENABLE_SIMULATOR and not settings.HARDWARE_MODE: |
| sim_task = asyncio.create_task(run_simulator(settings.SIMULATOR_INTERVAL_SECONDS)) |
| logger.info("Sensor simulator started") |
|
|
| try: |
| yield |
| finally: |
| if sim_task is not None: |
| sim_task.cancel() |
| try: |
| await sim_task |
| except asyncio.CancelledError: |
| pass |
|
|
|
|
| app = FastAPI( |
| title=settings.APP_NAME, |
| version=settings.APP_VERSION, |
| description="BioPatch — predictive SSI / sepsis monitoring command center", |
| lifespan=lifespan, |
| ) |
|
|
| app.add_middleware( |
| CORSMiddleware, |
| allow_origins=settings.CORS_ORIGINS, |
| allow_origin_regex=settings.CORS_ORIGIN_REGEX, |
| allow_credentials=True, |
| allow_methods=["*"], |
| allow_headers=["*"], |
| ) |
|
|
|
|
| @app.get("/api/health", tags=["meta"]) |
| def health() -> dict: |
| return {"status": "ok", "service": settings.APP_NAME, "version": settings.APP_VERSION} |
|
|
|
|
| app.include_router(auth.router) |
| app.include_router(ingest.router) |
| app.include_router(patients.router) |
| app.include_router(sensors.router) |
| app.include_router(vitals.router) |
| app.include_router(notes.router) |
| app.include_router(jarvis.router) |
| app.include_router(jarvis_agent.router) |
| app.include_router(jarvis_chat.router) |
| app.include_router(jarvis_learning.router) |
| app.include_router(wound_profiles.router) |
| app.include_router(platform.router) |
| app.include_router(websocket.router) |
|
|
