146 lines
4.3 KiB
Python
146 lines
4.3 KiB
Python
"""
|
|
REST API for S-57 ENC chart management — GPS Navigator.
|
|
"""
|
|
import asyncio
|
|
import logging
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
from fastapi import APIRouter, HTTPException, UploadFile, File
|
|
from fastapi.responses import JSONResponse
|
|
from pydantic import BaseModel
|
|
|
|
|
|
class ScanPathRequest(BaseModel):
|
|
path: str
|
|
|
|
from backend.chart_manager import (
|
|
install_from_zip, install_from_enc, list_cells,
|
|
delete_cell, get_all_features, get_all_depths,
|
|
get_all_land, get_all_hazards, get_all_zones,
|
|
CHARTS_DIR, set_meta,
|
|
install_from_csv_zip, scan_and_install,
|
|
)
|
|
|
|
router = APIRouter(prefix="/charts", tags=["charts"])
|
|
log = logging.getLogger(__name__)
|
|
|
|
|
|
@router.get("/cells")
|
|
def get_cells():
|
|
return list_cells()
|
|
|
|
|
|
@router.post("/upload")
|
|
async def upload_chart(file: UploadFile = File(...)):
|
|
"""
|
|
Upload a chart file.
|
|
Accepts:
|
|
• .000 — single S-57 ENC cell
|
|
• .zip — NOAA ENC zip (contains .000) OR CSV-based custom zip
|
|
"""
|
|
suffix = Path(file.filename or "").suffix.lower()
|
|
if suffix not in (".zip", ".000"):
|
|
raise HTTPException(400, "Only .zip or .000 files accepted")
|
|
|
|
data = await file.read()
|
|
with tempfile.NamedTemporaryFile(suffix=suffix, delete=False) as tmp:
|
|
tmp.write(data)
|
|
tmp_path = Path(tmp.name)
|
|
|
|
try:
|
|
if suffix == ".zip":
|
|
# Auto-detect CSV vs ENC zip
|
|
import zipfile as _zf
|
|
with _zf.ZipFile(tmp_path) as z:
|
|
names = z.namelist()
|
|
has_csv = any(n.lower().endswith(".csv") for n in names)
|
|
has_enc = any(n.upper().endswith(".000") for n in names)
|
|
if has_csv and not has_enc:
|
|
installed = await asyncio.get_event_loop().run_in_executor(
|
|
None, install_from_csv_zip, tmp_path)
|
|
else:
|
|
installed = await asyncio.get_event_loop().run_in_executor(
|
|
None, install_from_zip, tmp_path)
|
|
else:
|
|
orig_name = Path(file.filename).stem.upper() if file.filename else None
|
|
cell_id = await asyncio.get_event_loop().run_in_executor(
|
|
None, install_from_enc, tmp_path, orig_name)
|
|
installed = [cell_id]
|
|
except Exception as e:
|
|
log.exception("Chart upload failed: %s", e)
|
|
raise HTTPException(500, "Chart processing failed — check server logs for details")
|
|
finally:
|
|
tmp_path.unlink(missing_ok=True)
|
|
|
|
return {"installed": installed}
|
|
|
|
|
|
@router.delete("/cells/{cell_id}")
|
|
def remove_cell(cell_id: str):
|
|
delete_cell(cell_id)
|
|
return {"deleted": cell_id.upper()}
|
|
|
|
|
|
@router.patch("/cells/{cell_id}/region")
|
|
def set_cell_region(cell_id: str, region: str):
|
|
region = (region or "").upper().strip()
|
|
if region not in ("A", "B"):
|
|
raise HTTPException(400, "region must be 'A' or 'B'")
|
|
try:
|
|
set_meta(cell_id, region=region)
|
|
except FileNotFoundError:
|
|
raise HTTPException(404, f"Cell {cell_id} not installed")
|
|
return {"id": cell_id.upper(), "region": region}
|
|
|
|
|
|
@router.get("/features")
|
|
def chart_features():
|
|
return JSONResponse(get_all_features())
|
|
|
|
|
|
@router.get("/depths")
|
|
def chart_depths(w: float | None = None, s: float | None = None,
|
|
e: float | None = None, n: float | None = None):
|
|
bbox = (w, s, e, n) if None not in (w, s, e, n) else None
|
|
return JSONResponse(get_all_depths(bbox))
|
|
|
|
|
|
@router.get("/land")
|
|
def chart_land():
|
|
return JSONResponse(get_all_land())
|
|
|
|
|
|
@router.get("/hazards")
|
|
def chart_hazards():
|
|
return JSONResponse(get_all_hazards())
|
|
|
|
|
|
@router.get("/zones")
|
|
def chart_zones():
|
|
return JSONResponse(get_all_zones())
|
|
|
|
|
|
@router.post("/scan-path")
|
|
async def scan_path(body: ScanPathRequest):
|
|
"""
|
|
Scan a local directory (e.g. SD card drive letter) for .000 / .zip chart
|
|
files and install them.
|
|
|
|
Body: { "path": "E:\\ENC_Charts" }
|
|
"""
|
|
directory = (body.path or "").strip()
|
|
if not directory:
|
|
raise HTTPException(400, "path is required")
|
|
|
|
try:
|
|
result = await asyncio.get_event_loop().run_in_executor(
|
|
None, scan_and_install, directory)
|
|
except (FileNotFoundError, NotADirectoryError) as exc:
|
|
raise HTTPException(404, str(exc))
|
|
except Exception as exc:
|
|
log.exception("scan-path failed: %s", exc)
|
|
raise HTTPException(500, "Scan failed — check server logs for details")
|
|
|
|
return result
|