25'ten fazla konu seçemezsiniz Konular bir harf veya rakamla başlamalı, kısa çizgiler ('-') içerebilir ve en fazla 35 karakter uzunluğunda olabilir.
 
 
 
 

541 satır
16 KiB

  1. import asyncio
  2. from fastapi import APIRouter, Depends, HTTPException, Query, Request
  3. import httpx
  4. import config_env
  5. from typing import List, Optional
  6. from schemas.reslevis import (
  7. BuildingItem,
  8. FloorItem,
  9. ZoneItem,
  10. GatewayItem,
  11. TrackerItem,
  12. OperatorItem,
  13. SubjectItem,
  14. AlarmItem,
  15. TrackItem,
  16. TrackHistoryItem,
  17. TrackerZoneItem,
  18. SettingItem,
  19. )
  20. from logica_reslevis.gateway import GatewayJsonRepository
  21. from logica_reslevis.building import BuildingJsonRepository
  22. from logica_reslevis.floor import FloorJsonRepository
  23. from logica_reslevis.zone import ZoneJsonRepository
  24. from logica_reslevis.tracker import TrackerJsonRepository
  25. from logica_reslevis.operator import OperatorJsonRepository
  26. from logica_reslevis.setting import SettingJsonRepository
  27. from logica_reslevis.subject import SubjectJsonRepository
  28. from logica_reslevis.alarm import AlarmJsonRepository
  29. from logica_reslevis.track import TrackJsonRepository
  30. from logica_reslevis.tracker_zone import TrackerZoneJsonRepository
  31. from security import get_current_user
  32. #CORE SYNC
  33. CORE_BASE_URL = config_env.CORE_API_URL.rstrip("/")
  34. CORE_TIMEOUT = 2.0 # secondi
  35. async def sync_core_get(request: Request) -> None:
  36. if request.method != "GET":
  37. return
  38. sync = CORE_GET_SYNC.get(request.url.path)
  39. if sync is None:
  40. return
  41. repo, normalizer = sync
  42. try:
  43. async with httpx.AsyncClient(timeout=CORE_TIMEOUT) as client:
  44. resp = await client.get(
  45. f"{CORE_BASE_URL}{request.url.path}",
  46. params=request.query_params,
  47. )
  48. if 200 <= resp.status_code < 300:
  49. data = resp.json()
  50. if isinstance(data, list):
  51. if normalizer:
  52. data = [normalizer(r) for r in data if isinstance(r, dict)]
  53. repo._write_all(data) # aggiorna i file locali
  54. except (httpx.RequestError, ValueError):
  55. # CORE giù o risposta non valida -> uso il file locale
  56. pass
  57. router = APIRouter(dependencies=[Depends(sync_core_get)])
  58. gateway_repo = GatewayJsonRepository()
  59. building_repo = BuildingJsonRepository()
  60. floor_repo = FloorJsonRepository()
  61. zone_repo = ZoneJsonRepository()
  62. tracker_repo = TrackerJsonRepository()
  63. operator_repo = OperatorJsonRepository()
  64. subject_repo = SubjectJsonRepository()
  65. alarm_repo = AlarmJsonRepository()
  66. track_repo = TrackJsonRepository()
  67. tracker_zone_repo = TrackerZoneJsonRepository()
  68. setting_repo = SettingJsonRepository()
  69. def _none_if_empty(v):
  70. return None if v in ("", None, 0, "0") else v
  71. def _str_or_none(v):
  72. if v in ("", None):
  73. return None
  74. if isinstance(v, (int, float, bool)):
  75. return str(v)
  76. return v
  77. def _uuid_list(values):
  78. if values in ("", None):
  79. return []
  80. if isinstance(values, str):
  81. values = [v for v in values.split(",") if v]
  82. if isinstance(values, (list, tuple, set)):
  83. cleaned = []
  84. for v in values:
  85. if isinstance(v, dict):
  86. v = v.get("id") or v.get("uuid")
  87. if v in ("", None, 0, "0"):
  88. continue
  89. cleaned.append(v)
  90. return cleaned
  91. return [values] if values not in ("", None, 0, "0") else []
  92. def _normalize_gateway(row: dict) -> dict:
  93. row = dict(row)
  94. row["floor"] = _none_if_empty(row.get("floor"))
  95. row["building"] = _none_if_empty(row.get("building"))
  96. return row
  97. def _normalize_tracker(row: dict) -> dict:
  98. row = dict(row)
  99. row["floor"] = _none_if_empty(row.get("floor"))
  100. row["building"] = _none_if_empty(row.get("building"))
  101. row["battery"] = _str_or_none(row.get("battery"))
  102. row["temperature"] = _str_or_none(row.get("temperature"))
  103. row["acceleration"] = _str_or_none(row.get("acceleration"))
  104. row["heartRate"] = _str_or_none(row.get("heartRate"))
  105. return row
  106. def _normalize_track(row: dict) -> dict:
  107. row = dict(row)
  108. row["ID"] = row.get("ID")
  109. row["gateway"] = _none_if_empty(row.get("gateway"))
  110. row["tracker"] = _none_if_empty(row.get("tracker"))
  111. row["subject"] = _none_if_empty(row.get("subject"))
  112. row["floor"] = _none_if_empty(row.get("floor"))
  113. row["building"] = _none_if_empty(row.get("building"))
  114. row["timestamp"] = _str_or_none(row.get("timestamp"))
  115. row["type"] = _str_or_none(row.get("type"))
  116. row["status"] = _str_or_none(row.get("status"))
  117. row["gatewayMac"] = _str_or_none(row.get("gatewayMac"))
  118. row["trackerMac"] = _str_or_none(row.get("trackerMac"))
  119. row["subjectName"] = _str_or_none(row.get("subjectName"))
  120. row["x"] = None if row.get("x") in ("", None) else row.get("x")
  121. row["y"] = None if row.get("y") in ("", None) else row.get("y")
  122. row["z"] = None if row.get("z") in ("", None) else row.get("z")
  123. # signal resta float o None
  124. row["signal"] = None if row.get("signal") in ("", None) else row.get("signal")
  125. return row
  126. def _normalize_zone(row: dict) -> dict:
  127. row = dict(row)
  128. row["floor"] = _none_if_empty(row.get("floor"))
  129. row["building"] = _none_if_empty(row.get("building"))
  130. row["groups"] = _uuid_list(row.get("groups"))
  131. return row
  132. CORE_GET_SYNC = {
  133. "/reslevis/getGateways": (gateway_repo, _normalize_gateway),
  134. "/reslevis/getZones": (zone_repo, _normalize_zone),
  135. "/reslevis/getTrackers": (tracker_repo, _normalize_tracker),
  136. }
  137. async def _core_get_json(
  138. client: httpx.AsyncClient,
  139. path: str,
  140. params: Optional[dict] = None,
  141. ):
  142. resp = await client.get(f"{CORE_BASE_URL}{path}", params=params)
  143. if resp.status_code >= 400:
  144. detail = resp.text.strip() or "CORE request failed"
  145. raise HTTPException(status_code=resp.status_code, detail=detail)
  146. try:
  147. return resp.json()
  148. except ValueError as exc:
  149. raise HTTPException(status_code=502, detail="Invalid CORE response") from exc
  150. async def _fetch_tracks_for_tracker(
  151. client: httpx.AsyncClient,
  152. tracker_id: str,
  153. params: Optional[dict] = None,
  154. ) -> List[dict]:
  155. payload = await _core_get_json(client, f"/reslevis/getTracks/{tracker_id}", params=params)
  156. if not isinstance(payload, list):
  157. raise HTTPException(status_code=502, detail="Unexpected CORE response type")
  158. return [_normalize_track(row) for row in payload if isinstance(row, dict)]
  159. def _sort_tracks_desc(rows: List[dict]) -> List[dict]:
  160. return sorted(rows, key=lambda row: row.get("timestamp") or "", reverse=True)
  161. async def _fetch_all_tracks(params: dict) -> List[dict]:
  162. async with httpx.AsyncClient(timeout=30.0, verify=False) as client:
  163. trackers_payload = await _core_get_json(client, "/reslevis/getTrackers")
  164. if not isinstance(trackers_payload, list):
  165. raise HTTPException(status_code=502, detail="Unexpected CORE tracker response type")
  166. tracker_ids = []
  167. for item in trackers_payload:
  168. if not isinstance(item, dict):
  169. continue
  170. tracker_id = item.get("id")
  171. if tracker_id:
  172. tracker_ids.append(str(tracker_id))
  173. batches = await asyncio.gather(
  174. *[_fetch_tracks_for_tracker(client, tracker_id, params) for tracker_id in tracker_ids]
  175. )
  176. merged = [row for batch in batches for row in batch]
  177. merged = _sort_tracks_desc(merged)
  178. limit = params.get("limit")
  179. if isinstance(limit, int):
  180. return merged[:limit]
  181. return merged
  182. @router.get(
  183. "/getGateways",
  184. response_model=List[GatewayItem],
  185. tags=["Reslevis"],
  186. dependencies=[Depends(get_current_user)],
  187. )
  188. def getGateways():
  189. return gateway_repo.list()
  190. @router.post("/postGateway", tags=["Reslevis"], dependencies=[Depends(get_current_user)])
  191. def postGateway(item: GatewayItem):
  192. gateway_repo.add(item)
  193. return {"message": "OK"}
  194. @router.put("/updateGateway", tags=["Reslevis"], dependencies=[Depends(get_current_user)])
  195. def updateGateway(item: GatewayItem):
  196. gateway_repo.update(item)
  197. return {"message": "OK"}
  198. @router.delete("/removeGateway/{gateway_id}", tags=["Reslevis"], dependencies=[Depends(get_current_user)])
  199. def removeGateway(gateway_id: str):
  200. gateway_repo.remove(gateway_id)
  201. return {"message": "OK"}
  202. @router.get(
  203. "/getBuildings",
  204. response_model=List[BuildingItem],
  205. tags=["Reslevis"],
  206. dependencies=[Depends(get_current_user)],
  207. )
  208. def getBuildings():
  209. return building_repo.list()
  210. @router.post("/postBuilding", tags=["Reslevis"], dependencies=[Depends(get_current_user)])
  211. def postBuilding(item: BuildingItem):
  212. building_repo.add(item)
  213. return {"message": "OK"}
  214. @router.put("/updateBuilding", tags=["Reslevis"], dependencies=[Depends(get_current_user)])
  215. def updateBuilding(item: BuildingItem):
  216. building_repo.update(item)
  217. return {"message": "OK"}
  218. @router.delete("/removeBuilding/{building_id}", tags=["Reslevis"], dependencies=[Depends(get_current_user)])
  219. def removeBuilding(building_id: str):
  220. building_repo.remove(building_id)
  221. return {"message": "OK"}
  222. @router.get(
  223. "/getFloors",
  224. response_model=List[FloorItem],
  225. tags=["Reslevis"],
  226. dependencies=[Depends(get_current_user)],
  227. )
  228. def getFloors():
  229. return floor_repo.list()
  230. @router.post("/postFloor", tags=["Reslevis"], dependencies=[Depends(get_current_user)])
  231. def postFloor(item: FloorItem):
  232. floor_repo.add(item)
  233. return {"message": "OK"}
  234. @router.put("/updateFloor", tags=["Reslevis"], dependencies=[Depends(get_current_user)])
  235. def updateFloor(item: FloorItem):
  236. floor_repo.update(item)
  237. return {"message": "OK"}
  238. @router.delete("/removeFloor/{floor_id}", tags=["Reslevis"], dependencies=[Depends(get_current_user)])
  239. def removeFloor(floor_id: str):
  240. floor_repo.remove(floor_id)
  241. return {"message": "OK"}
  242. @router.get(
  243. "/getZones",
  244. response_model=List[ZoneItem],
  245. tags=["Reslevis"],
  246. dependencies=[Depends(get_current_user)],
  247. )
  248. def getZones():
  249. return zone_repo.list()
  250. @router.post("/postZone", tags=["Reslevis"], dependencies=[Depends(get_current_user)])
  251. def postZone(item: ZoneItem):
  252. zone_repo.add(item)
  253. return {"message": "OK"}
  254. @router.put("/updateZone", tags=["Reslevis"], dependencies=[Depends(get_current_user)])
  255. def updateZone(item: ZoneItem):
  256. zone_repo.update(item)
  257. return {"message": "OK"}
  258. @router.delete("/removeZone/{zone_id}", tags=["Reslevis"], dependencies=[Depends(get_current_user)])
  259. def removeZone(zone_id: str):
  260. zone_repo.remove(zone_id)
  261. return {"message": "OK"}
  262. @router.get(
  263. "/getTrackers",
  264. response_model=List[TrackerItem],
  265. tags=["Reslevis"],
  266. dependencies=[Depends(get_current_user)],
  267. )
  268. def getTrackers():
  269. return tracker_repo.list()
  270. @router.post("/postTracker", tags=["Reslevis"], dependencies=[Depends(get_current_user)])
  271. def postTracker(item: TrackerItem):
  272. tracker_repo.add(item)
  273. return {"message": "OK"}
  274. @router.put("/updateTracker", tags=["Reslevis"], dependencies=[Depends(get_current_user)])
  275. def updateTracker(item: TrackerItem):
  276. tracker_repo.update(item)
  277. return {"message": "OK"}
  278. @router.delete("/removeTracker/{tracker_id}", tags=["Reslevis"], dependencies=[Depends(get_current_user)])
  279. def removeTracker(tracker_id: str):
  280. tracker_repo.remove(tracker_id)
  281. return {"message": "OK"}
  282. @router.get(
  283. "/getTrackerZones",
  284. response_model=List[TrackerZoneItem],
  285. tags=["Reslevis"],
  286. dependencies=[Depends(get_current_user)],
  287. )
  288. def getTrackerZones():
  289. return tracker_zone_repo.list()
  290. @router.post("/postTrackerZone", tags=["Reslevis"], dependencies=[Depends(get_current_user)])
  291. def postTrackerZone(item: TrackerZoneItem):
  292. tracker_zone_repo.add(item)
  293. return {"message": "OK"}
  294. @router.put("/updateTrackerZone", tags=["Reslevis"], dependencies=[Depends(get_current_user)])
  295. def updateTrackerZone(item: TrackerZoneItem):
  296. tracker_zone_repo.update(item)
  297. return {"message": "OK"}
  298. @router.delete("/removeTrackerZone/{tracker_zone_id}", tags=["Reslevis"], dependencies=[Depends(get_current_user)])
  299. def removeTrackerZone(tracker_zone_id: str):
  300. tracker_zone_repo.remove(tracker_zone_id)
  301. return {"message": "OK"}
  302. @router.get(
  303. "/getTracks",
  304. response_model=List[TrackHistoryItem],
  305. tags=["Reslevis"],
  306. dependencies=[Depends(get_current_user)],
  307. )
  308. async def getTracks(
  309. request: Request,
  310. tracker_id: Optional[str] = Query(None),
  311. limit: Optional[int] = Query(None, ge=1),
  312. from_: Optional[str] = Query(None, alias="from"),
  313. to: Optional[str] = Query(None),
  314. ):
  315. params = {}
  316. if limit is not None:
  317. params["limit"] = limit
  318. if from_:
  319. params["from"] = from_
  320. if to:
  321. params["to"] = to
  322. selected_tracker_id = tracker_id or request.query_params.get("id")
  323. try:
  324. if selected_tracker_id:
  325. async with httpx.AsyncClient(timeout=30.0, verify=False) as client:
  326. return await _fetch_tracks_for_tracker(client, selected_tracker_id, params)
  327. return await _fetch_all_tracks(params)
  328. except httpx.RequestError as exc:
  329. raise HTTPException(status_code=502, detail=f"CORE request failed: {exc}") from exc
  330. @router.get(
  331. "/getTracks/{tracker_id}",
  332. response_model=List[TrackHistoryItem],
  333. tags=["Reslevis"],
  334. dependencies=[Depends(get_current_user)],
  335. )
  336. async def getTrack(
  337. tracker_id: str,
  338. limit: Optional[int] = Query(None, ge=1),
  339. from_: Optional[str] = Query(None, alias="from"),
  340. to: Optional[str] = Query(None),
  341. ):
  342. params = {}
  343. if limit is not None:
  344. params["limit"] = limit
  345. if from_:
  346. params["from"] = from_
  347. if to:
  348. params["to"] = to
  349. try:
  350. async with httpx.AsyncClient(timeout=30.0, verify=False) as client:
  351. return await _fetch_tracks_for_tracker(client, tracker_id, params)
  352. except httpx.RequestError as exc:
  353. raise HTTPException(status_code=502, detail=f"CORE request failed: {exc}") from exc
  354. @router.post("/postTrack", tags=["Reslevis"], dependencies=[Depends(get_current_user)])
  355. def postTrack(item: TrackItem):
  356. track_repo.add(item)
  357. return {"message": "OK"}
  358. @router.put("/updateTrack", tags=["Reslevis"], dependencies=[Depends(get_current_user)])
  359. def updateTrack(item: TrackItem):
  360. track_repo.update(item)
  361. return {"message": "OK"}
  362. @router.delete("/removeTrack/{track_id}", tags=["Reslevis"], dependencies=[Depends(get_current_user)])
  363. def removeTrack(track_id: str):
  364. track_repo.remove(track_id)
  365. return {"message": "OK"}
  366. @router.get(
  367. "/getOperators",
  368. response_model=List[OperatorItem],
  369. tags=["Reslevis"],
  370. dependencies=[Depends(get_current_user)],
  371. )
  372. def getOperators():
  373. return operator_repo.list()
  374. @router.post("/postOperator", tags=["Reslevis"], dependencies=[Depends(get_current_user)])
  375. def postOperator(item: OperatorItem):
  376. operator_repo.add(item)
  377. return {"message": "OK"}
  378. @router.put("/updateOperator", tags=["Reslevis"], dependencies=[Depends(get_current_user)])
  379. def updateOperator(item: OperatorItem):
  380. operator_repo.update(item)
  381. return {"message": "OK"}
  382. @router.delete("/removeOperator/{operator_id}", tags=["Reslevis"], dependencies=[Depends(get_current_user)])
  383. def removeOperator(operator_id: str):
  384. operator_repo.remove(operator_id)
  385. return {"message": "OK"}
  386. @router.get(
  387. "/getSubjects",
  388. response_model=List[SubjectItem],
  389. tags=["Reslevis"],
  390. dependencies=[Depends(get_current_user)],
  391. )
  392. def getSubjects():
  393. return subject_repo.list()
  394. @router.post("/postSubject", tags=["Reslevis"], dependencies=[Depends(get_current_user)])
  395. def postSubject(item: SubjectItem):
  396. subject_repo.add(item)
  397. return {"message": "OK"}
  398. @router.put("/updateSubject", tags=["Reslevis"], dependencies=[Depends(get_current_user)])
  399. def updateSubject(item: SubjectItem):
  400. subject_repo.update(item)
  401. return {"message": "OK"}
  402. @router.delete("/removeSubject/{subject_id}", tags=["Reslevis"], dependencies=[Depends(get_current_user)])
  403. def removeSubject(subject_id: str):
  404. subject_repo.remove(subject_id)
  405. return {"message": "OK"}
  406. @router.get(
  407. "/getSettings",
  408. response_model=List[SettingItem],
  409. tags=["Reslevis"],
  410. dependencies=[Depends(get_current_user)],
  411. )
  412. def getSettings():
  413. return setting_repo.list()
  414. @router.post("/postSetting", tags=["Reslevis"], dependencies=[Depends(get_current_user)])
  415. def postSetting(item: SettingItem):
  416. setting_repo.add(item)
  417. return {"message": "OK"}
  418. @router.put("/updateSetting", tags=["Reslevis"], dependencies=[Depends(get_current_user)])
  419. def updateSetting(item: SettingItem):
  420. setting_repo.update(item)
  421. return {"message": "OK"}
  422. @router.delete("/removeSetting/{setting_id}", tags=["Reslevis"], dependencies=[Depends(get_current_user)])
  423. def removeSetting(setting_id: str):
  424. setting_repo.remove(setting_id)
  425. return {"message": "OK"}