|
|
|
@@ -9,6 +9,7 @@ from typing import Any, Dict, List, Optional |
|
|
|
# import wave |
|
|
|
import os |
|
|
|
import shutil |
|
|
|
import re |
|
|
|
# import enviroment variables |
|
|
|
import config_env |
|
|
|
#other |
|
|
|
@@ -41,6 +42,7 @@ from models.cellular_hardwares import cellularHardwares |
|
|
|
from models.call import call, post_call |
|
|
|
from models.calls import calls |
|
|
|
from models.httpresponse import httpResponse400, httpResponse200, httpResponse500 |
|
|
|
from schemas.reslevis import CalibrationMetadata |
|
|
|
|
|
|
|
from fastapi_login import LoginManager |
|
|
|
|
|
|
|
@@ -192,6 +194,23 @@ app.include_router(majornet_router) |
|
|
|
app.include_router(contacts_router) |
|
|
|
app.include_router(reslevis_router, prefix="/reslevis", tags=["Reslevis"]) |
|
|
|
|
|
|
|
META_FILE_PATTERN = re.compile(r"^meta_(-?\d+)\.json$") |
|
|
|
|
|
|
|
def _model_to_dict(model: Any) -> Dict[str, Any]: |
|
|
|
if hasattr(model, "model_dump"): |
|
|
|
return model.model_dump() |
|
|
|
return model.dict() |
|
|
|
|
|
|
|
def _validate_metadata(payload: Dict[str, Any]) -> Dict[str, Any]: |
|
|
|
if hasattr(CalibrationMetadata, "model_validate"): |
|
|
|
return _model_to_dict(CalibrationMetadata.model_validate(payload)) |
|
|
|
return _model_to_dict(CalibrationMetadata.parse_obj(payload)) |
|
|
|
|
|
|
|
def _read_ble_ai_metadata(path: str) -> Dict[str, Any]: |
|
|
|
with open(path) as f: |
|
|
|
payload = json.load(f) |
|
|
|
return _validate_metadata(payload) |
|
|
|
|
|
|
|
|
|
|
|
@app.get("/") |
|
|
|
async def root(): |
|
|
|
@@ -239,6 +258,45 @@ async def get_ble_ai_infer(mac: Optional[List[str]] = Query(default=None)): |
|
|
|
return {"items": items, "count": len(items)} |
|
|
|
|
|
|
|
|
|
|
|
@app.get("/ble-ai/metadata", tags=["BLE-AI"], dependencies=[Depends(get_current_user)]) |
|
|
|
async def get_ble_ai_metadata(floor: Optional[int] = Query(default=None)): |
|
|
|
meta_dir = config_env.BLE_AI_META_DIR |
|
|
|
if not os.path.isdir(meta_dir): |
|
|
|
raise HTTPException(status_code=404, detail="Metadata directory not found") |
|
|
|
|
|
|
|
if floor is not None: |
|
|
|
file_path = os.path.join(meta_dir, f"meta_{floor}.json") |
|
|
|
if not os.path.isfile(file_path): |
|
|
|
raise HTTPException(status_code=404, detail="Metadata file not found") |
|
|
|
try: |
|
|
|
metadata = _read_ble_ai_metadata(file_path) |
|
|
|
except Exception: |
|
|
|
log.exception("Invalid metadata file: %s", file_path) |
|
|
|
raise HTTPException(status_code=500, detail="Invalid metadata file") |
|
|
|
return {"floor": floor, "metadata": metadata} |
|
|
|
|
|
|
|
items = [] |
|
|
|
for file_name in os.listdir(meta_dir): |
|
|
|
match = META_FILE_PATTERN.match(file_name) |
|
|
|
if not match: |
|
|
|
continue |
|
|
|
|
|
|
|
file_floor = int(match.group(1)) |
|
|
|
file_path = os.path.join(meta_dir, file_name) |
|
|
|
if not os.path.isfile(file_path): |
|
|
|
continue |
|
|
|
|
|
|
|
try: |
|
|
|
metadata = _read_ble_ai_metadata(file_path) |
|
|
|
except Exception: |
|
|
|
log.exception("Invalid metadata file: %s", file_path) |
|
|
|
raise HTTPException(status_code=500, detail="Invalid metadata file") |
|
|
|
items.append({"floor": file_floor, "metadata": metadata}) |
|
|
|
|
|
|
|
items.sort(key=lambda item: item["floor"]) |
|
|
|
return {"items": items, "count": len(items)} |
|
|
|
|
|
|
|
|
|
|
|
@app.get("/openapi.json/", tags=["Documentation"]) |
|
|
|
async def get_open_api_endpoint(): |
|
|
|
#async def get_open_api_endpoint(current_user: User = Depends(get_current_active_user)): |
|
|
|
|