47 lines
1.6 KiB
Python
47 lines
1.6 KiB
Python
from __future__ import annotations
|
|
|
|
from contextlib import asynccontextmanager
|
|
|
|
from fastapi import FastAPI
|
|
|
|
from app.config import settings
|
|
from app.models.schemas import DataLayersHealth, HealthResponse
|
|
from app.routers import coverage, elevation, link
|
|
from app.services.buildings import BuildingService
|
|
from app.services.canopy import CanopyService
|
|
from app.services.coverage import CoverageService
|
|
from app.services.link_calculator import LinkCalculator
|
|
from app.services.terrain import TerrainService
|
|
|
|
terrain_service = TerrainService(settings.srtm_path, cache_size=settings.raster_cache_size)
|
|
canopy_service = CanopyService(settings.canopy_path, cache_size=settings.raster_cache_size)
|
|
building_service = BuildingService(settings.database_url)
|
|
link_calculator = LinkCalculator(terrain_service, canopy_service, building_service)
|
|
coverage_service = CoverageService(link_calculator)
|
|
|
|
|
|
@asynccontextmanager
|
|
async def lifespan(_: FastAPI):
|
|
building_service.connect()
|
|
yield
|
|
building_service.close()
|
|
terrain_service.close()
|
|
canopy_service.close()
|
|
|
|
|
|
app = FastAPI(title="Radio Link Service", version="1.0.0", lifespan=lifespan)
|
|
app.include_router(link.router)
|
|
app.include_router(coverage.router)
|
|
app.include_router(elevation.router)
|
|
|
|
|
|
@app.get("/health", response_model=HealthResponse)
|
|
async def health() -> HealthResponse:
|
|
layers = DataLayersHealth(
|
|
srtm=terrain_service.is_available(),
|
|
canopy=canopy_service.is_available(),
|
|
buildings=building_service.is_available(),
|
|
)
|
|
status = "ok" if all(layers.model_dump().values()) else "degraded"
|
|
return HealthResponse(status=status, data_layers=layers)
|