36 lines
1.1 KiB
Python
36 lines
1.1 KiB
Python
from sqlalchemy import text
|
|
from sqlalchemy.exc import SQLAlchemyError
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.config import get_settings
|
|
from app.core.raster_sampling import raster_files
|
|
from app.models.status import DataStatusResponse, PostgisStatus, RasterStatus
|
|
|
|
|
|
def data_status(db: Session) -> DataStatusResponse:
|
|
settings = get_settings()
|
|
return DataStatusResponse(
|
|
dem=_raster_status(settings.dem_path),
|
|
landcover=_raster_status(settings.landcover_path),
|
|
canopy=_raster_status(settings.canopy_path),
|
|
postgis=_postgis_status(db),
|
|
)
|
|
|
|
|
|
def _raster_status(path: object) -> RasterStatus:
|
|
files = raster_files(path)
|
|
return RasterStatus(
|
|
configured=bool(files),
|
|
path=str(path),
|
|
files_count=len(files),
|
|
)
|
|
|
|
|
|
def _postgis_status(db: Session) -> PostgisStatus:
|
|
try:
|
|
buildings_count = db.execute(text("SELECT count(*) FROM buildings")).scalar_one()
|
|
except SQLAlchemyError as exc:
|
|
db.rollback()
|
|
return PostgisStatus(configured=False, error=str(exc))
|
|
return PostgisStatus(configured=True, buildings_count=int(buildings_count))
|