Coverage for /usr/local/lib/python3.11/site-packages/twinpad_backend/api.py: 79%
369 statements
« prev ^ index » next coverage.py v7.9.2, created at 2025-07-04 13:35 +0000
1import os
2import logging
3import time
4from tempfile import mkdtemp
5from typing import Annotated
6from datetime import timedelta
7from pathlib import Path
8from pyinstrument import Profiler
10from fastapi import FastAPI, HTTPException, Depends, Query, Response, Request
11from fastapi.middleware.cors import CORSMiddleware
12from fastapi.responses import HTMLResponse
13from fastapi.security import OAuth2PasswordRequestForm
15from twinpad_backend import __version__
16from twinpad_backend.db import signal_datasize, get_signals_ids_from_collection_names
17from twinpad_backend.models import (
18 Signal,
19 SignalData,
20 SignalSample,
21 ServicesStatus,
22 Device,
23 DeviceUpdate,
24 DeviceSetup,
25 DeviceSetupUpdate,
26 DeviceState,
27 SignalUpdate,
28 SignalsData,
29 Event,
30 EventRule,
31 EventDay,
32 User,
33 UserUpdate,
34 Campaign,
35 Phase,
36 CustomView,
37 CustomViewCreation,
38 CustomViewUpdate,
39 Video,
40)
41from twinpad_backend.auth import (
42 Token,
43 authenticate_user,
44 get_current_active_user,
45 ACCESS_TOKEN_EXPIRE_MINUTES,
46 create_access_token,
47 get_password_hash,
48)
49from twinpad_backend.queries import SignalQuery, DeviceStatesQuery, EventQuery, EventRuleQuery
50from twinpad_backend.responses import ListResponse
52REQUEST_TIME_WARNING = 0.5
54DEBUG = os.environ.get("DEBUG", "false") == "true"
55PROFILING = os.environ.get("PROFILING", "false") == "true"
57logger = logging.getLogger("uvicorn.error")
58logger.propagate = False
59logger.info("Debug mode: %s", DEBUG)
60logger.info("log level: %s", logging.root.level)
63app = FastAPI(title="Twinpad backend", version=__version__)
65app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"])
67if PROFILING:
68 profiling_folder = mkdtemp()
69 logger.info("Profiling enabled")
71 @app.middleware("http")
72 async def profile_request(request: Request, call_next):
73 should_profile = True
74 url = str(request.url)
75 for segment in ["profiling", ".ico"]:
76 if segment in url:
77 should_profile = False
78 break
80 if should_profile: # avoid recursion
81 profiler = Profiler()
82 profiler.start()
83 await call_next(request)
84 profiler.stop()
85 url = "_".join(url.split("/")[3:]).rstrip("/")
86 if not url:
87 url = "slash"
88 # filename = f"{round(time.time())}_{url}"
89 filename = url.split("?", maxsplit=1)[0]
90 logger.info("saving profiling to %s", filename)
91 with open(os.path.join(profiling_folder, filename), "w", encoding="utf-8") as profiling_file:
92 profiling_file.write(profiler.output_html())
93 return await call_next(request)
95 @app.get("/profilings")
96 async def profilings():
97 return {"profilings": os.listdir(profiling_folder)}
99 @app.get("/profilings/{profiling_id}")
100 async def profiling(profiling_id):
102 filename = os.path.join(profiling_folder, profiling_id)
103 if os.path.exists(filename):
104 with open(filename, "r", encoding="utf-8") as profiling:
105 return HTMLResponse(profiling.read())
106 raise HTTPException(
107 status_code=404,
108 detail="Profiling not found",
109 )
111 # app.mount("/profilings", StaticFiles(directory=profiling_folder, html=True), name="profilings")
114@app.middleware("http")
115async def log_request_time(request: Request, call_next):
116 start_time = time.time() # Record the start time
117 response = await call_next(request) # Process the request
118 duration = time.time() - start_time # Calculate the time taken
119 client_ip = request.headers.get("x-forwarded-for", request.client.host)
120 message = f"{client_ip} {request.method} {request.url.path} - {response.status_code} - {round(1000*duration)}ms"
121 if duration > REQUEST_TIME_WARNING:
122 logger.warning(message)
123 else:
124 logger.info(message)
125 return response
128@app.get("/")
129async def slash():
130 return {"twinpad_version": __version__}
133@app.get("/status", dependencies=[Depends(get_current_active_user)])
134async def status():
135 """
136 Return service healthcheck
137 """
138 return {
139 "services": ServicesStatus.check(),
140 }
143@app.get("/devices", dependencies=[Depends(get_current_active_user)])
144async def get_devices() -> list[Device]:
145 return Device.get_all(sort_by="device_id")
148@app.get("/devices/{device_id}", dependencies=[Depends(get_current_active_user)])
149async def get_device(device_id):
150 device = Device.get_one_by_attribute("device_id", device_id)
151 if not device:
152 raise HTTPException(
153 status_code=404,
154 detail="Device not found",
155 )
156 return device
159@app.patch("/devices/{device_id}", response_model=DeviceUpdate, dependencies=[Depends(get_current_active_user)])
160async def update_item(device_id: str, device_update: DeviceUpdate):
161 device = Device.get_one_by_attribute("device_id", device_id)
162 if not device:
163 raise HTTPException(
164 status_code=404,
165 detail="Device not found",
166 )
167 device.update(device_update)
168 return device_update
171@app.get("/devices/{device_id}/states", dependencies=[Depends(get_current_active_user)])
172async def get_device_states(device_id: str, query: DeviceStatesQuery = Depends()) -> ListResponse[DeviceState]:
173 return DeviceState.get_from_id_and_query(device_id, query)
176@app.get("/device-setups", dependencies=[Depends(get_current_active_user)])
177async def get_device_setups() -> list[DeviceSetup]:
178 return DeviceSetup.get_all()
181@app.post("/device-setups", dependencies=[Depends(get_current_active_user)], status_code=201)
182async def create_device_setups(device_setup: DeviceSetup) -> DeviceSetup:
183 device_setup.insert()
184 return device_setup
187@app.get("/device-setups/{device_setup_id}", dependencies=[Depends(get_current_active_user)])
188async def get_device_setup(device_setup_id: str):
189 device_setup = DeviceSetup.get_from_id(device_setup_id)
190 if device_setup is None:
191 raise HTTPException(
192 status_code=404,
193 detail="Device setup not found",
194 )
195 return device_setup
198@app.patch("/device-setups/{device_setup_id}", dependencies=[Depends(get_current_active_user)])
199async def edit_device_setups(device_setup_id: str, device_setup_update: DeviceSetupUpdate) -> DeviceSetup:
200 device_setup = DeviceSetup.get_from_id(device_setup_id)
201 if device_setup is None:
202 raise HTTPException(
203 status_code=404,
204 detail="Device setup not found",
205 )
206 device_setup.update({k: v for k, v in device_setup_update.model_dump().items() if v is not None})
207 return device_setup
210@app.delete("/device-setups/{device_setup_id}", dependencies=[Depends(get_current_active_user)], status_code=200)
211async def delete_device_setups(device_setup_id: str) -> bool:
212 device_setup = DeviceSetup.get_from_id(device_setup_id)
213 if device_setup is None:
214 raise HTTPException(
215 status_code=404,
216 detail="Device setup not found",
217 )
218 deleted = device_setup.delete()
219 return deleted
222@app.get("/signals", dependencies=[Depends(get_current_active_user)])
223async def route_get_signals(query: SignalQuery = Depends()) -> ListResponse[Signal]:
224 return Signal.response_from_query(query).to_dict(exclude={"device"})
227@app.get("/signals/stats", dependencies=[Depends(get_current_active_user)])
228async def signal_stats():
229 """
230 Returns signals stats
231 """
232 # profiler = Profiler()
233 # profiler.start()
235 number_samples = 0
236 number_active_signals = 0
237 signals = [Signal.get_from_signal_id(signal_id=sid) for sid in get_signals_ids_from_collection_names()]
239 for s in signals:
240 if s is not None:
241 number_samples += await s.number_samples()
242 if s.status.status == "up":
243 number_active_signals += 1
245 number_signals = Signal.get_number_documents()
247 # profiler.stop()
248 # filename = "signals_stats_profiling.html"
249 # full_file_path = os.path.join(Path.home(), filename)
250 # logger.info(f"Saving profiling to %s", full_file_path)
251 # with open(full_file_path, "w", encoding="utf-8") as profiling_file:
252 # profiling_file.write(profiler.output_html())
254 return {
255 "signal_data_size": signal_datasize(),
256 "number_signal_samples": number_samples,
257 "number_active_signals": number_active_signals,
258 "number_signals": number_signals,
259 }
262@app.get("/signals/last-value", dependencies=[Depends(get_current_active_user)])
263async def get_last_values(signal_ids: list[str] = Query(default=[])) -> list[SignalSample | None]:
264 return SignalSample.get_last_from_signal_ids(signal_ids)
267@app.get("/signals/first-value", dependencies=[Depends(get_current_active_user)])
268async def get_first_values(signal_ids: list[str] = Query(default=[])) -> list[SignalSample | None]:
269 return SignalSample.get_first_from_signal_ids(signal_ids)
272@app.get("/signals/forcibility", dependencies=[Depends(get_current_active_user)])
273async def get_signals_forcibility(signal_ids: list[str] = Query(default=[])) -> dict[str, bool]:
274 return Signal.get_forcibility(signal_ids)
277@app.get("/signals/{signal_id}", dependencies=[Depends(get_current_active_user)])
278async def get_signal(signal_id):
279 signal = Signal.get_from_signal_id(signal_id)
280 if not signal:
281 raise HTTPException(
282 status_code=404,
283 detail="Signal not found",
284 )
285 return signal.to_dict()
288@app.patch("/signals/{signal_id}", response_model=SignalUpdate, dependencies=[Depends(get_current_active_user)])
289async def update_signal(signal_id: str, signal_update: SignalUpdate):
290 signal = Signal.get_from_signal_id(signal_id)
291 if not signal:
292 raise HTTPException(
293 status_code=404,
294 detail="Device not found",
295 )
296 signal.update(signal_update)
297 return signal_update
300@app.get("/signals/{signal_id}/data", dependencies=[Depends(get_current_active_user)])
301async def get_signal_data(
302 signal_id: str, number_samples_max: int = None, min_timestamp: float = None, max_timestamp: float = None
303) -> SignalData | None:
304 signal_data = SignalData.get_from_signal_id(signal_id, min_timestamp=min_timestamp, max_timestamp=max_timestamp)
306 if number_samples_max is not None:
307 signal_data = signal_data.uniform_desampling(number_samples_max=number_samples_max)
309 return signal_data
312@app.get("/signals/{signal_id}/last-value", dependencies=[Depends(get_current_active_user)])
313async def get_last_value(signal_id) -> SignalSample:
314 sample = SignalSample.get_last_from_signal_id(signal_id)
315 if sample is None:
316 raise HTTPException(status_code=404, detail="No data")
317 return sample
320@app.get("/signals/{signal_id}/first-value", dependencies=[Depends(get_current_active_user)])
321async def get_first_value(signal_id) -> SignalSample:
322 sample = SignalSample.get_first_from_signal_id(signal_id)
323 if sample is None:
324 raise HTTPException(status_code=404, detail="No data")
325 return sample
328@app.get("/signals/{signal_id}/number-samples", dependencies=[Depends(get_current_active_user)])
329async def get_signal_number_samples(signal_id):
330 signal = Signal.get_from_signal_id(signal_id)
331 if not signal:
332 raise HTTPException(
333 status_code=404,
334 detail="Device not found",
335 )
336 return {"signal_id": signal_id, "number_samples": await signal.number_samples(), "size": signal.sample_datasize()}
339@app.get("/signals-data", dependencies=[Depends(get_current_active_user)])
340async def get_signals_data(
341 signal_ids: list[str] = Query(default=[]),
342 number_samples_max: int = None,
343 min_timestamp: float = None,
344 max_timestamp: float = None,
345 interpolate_bounds: bool = True,
346) -> SignalsData | None:
347 # profiler = Profiler()
348 # profiler.start()
350 if min_timestamp and max_timestamp and min_timestamp > max_timestamp:
351 raise HTTPException(status_code=400, detail="min_timestamp should be less than max_timestamp")
353 signals_data = SignalsData.get_from_signal_ids(
354 signal_ids,
355 min_timestamp=min_timestamp,
356 max_timestamp=max_timestamp,
357 window_min_timestamp=min_timestamp,
358 window_max_timestamp=max_timestamp,
359 interpolate_bounds=interpolate_bounds,
360 )
361 if number_samples_max is not None:
362 signals_data = signals_data.uniform_desampling(number_samples_max=number_samples_max)
364 # profiler.stop()
365 # filename = "signals-data.html"
366 # full_file_path = os.path.join(Path.home(), filename)
367 # logger.info(f"Saving profiling to %s", full_file_path)
368 # with open(full_file_path, "w", encoding="utf-8") as profiling_file:
369 # profiling_file.write(profiler.output_html())
371 return signals_data
374@app.get("/signals-data/interest-window", dependencies=[Depends(get_current_active_user)])
375async def get_signals_data_interest_window(
376 window_max_number_samples: int = 600,
377 outside_max_number_samples: int = 150,
378 window_min_timestamp: float = None,
379 window_max_timestamp: float = None,
380 signal_ids: list[str] = Query(default=[]),
381 min_timestamp: float = None,
382 max_timestamp: float = None,
383) -> SignalsData | None:
384 # profiler = Profiler()
385 # profiler.start()
387 if window_min_timestamp and window_max_timestamp and window_min_timestamp > window_max_timestamp:
388 raise HTTPException(status_code=400, detail="window_min_timestamp should be less than window_max_timestamp")
390 if min_timestamp and max_timestamp and min_timestamp > max_timestamp:
391 raise HTTPException(status_code=400, detail="min_timestamp should be less than max_timestamp")
393 signals_data = SignalsData.get_from_signal_ids(
394 signal_ids,
395 min_timestamp=min_timestamp,
396 max_timestamp=max_timestamp,
397 window_min_timestamp=window_min_timestamp,
398 window_max_timestamp=window_max_timestamp,
399 max_documents=10 * (window_max_number_samples + outside_max_number_samples),
400 )
402 signals_data = signals_data.interest_window_desampling(
403 window_max_number_samples=window_max_number_samples,
404 outside_max_number_samples=outside_max_number_samples,
405 window_min_timestamp=window_min_timestamp,
406 window_max_timestamp=window_max_timestamp,
407 )
409 # profiler.stop()
410 # filename = "signals-data.html"
411 # full_file_path = os.path.join(Path.home(), filename)
412 # logger.info(f"Saving profiling to %s", full_file_path)
413 # with open(full_file_path, "w", encoding="utf-8") as profiling_file:
414 # profiling_file.write(profiler.output_html())
416 return signals_data
419@app.get("/signals-data/export", dependencies=[Depends(get_current_active_user)])
420async def export_signals_zip(
421 file_format: str,
422 signal_ids: list[str] = Query(default=[]),
423 min_timestamp: float = None,
424 max_timestamp: float = None,
425):
426 signals_data = SignalsData.get_from_signal_ids(
427 signal_ids, min_timestamp=min_timestamp, max_timestamp=max_timestamp, interpolate_bounds=False
428 )
429 zip_data = signals_data.zip_export(file_format)
430 return Response(
431 content=zip_data,
432 media_type="application/zip",
433 headers={"Content-Disposition": 'attachment; filename="signals.zip"'},
434 )
437@app.get("/events", dependencies=[Depends(get_current_active_user)])
438async def get_events(query: EventQuery = Depends()) -> ListResponse[Event]:
439 return Event.response_from_query(query)
442@app.get("/events/{event_id}", dependencies=[Depends(get_current_active_user)])
443async def get_event(event_id) -> Event:
444 event = Event.get_from_id(event_id)
445 if event is None:
446 raise HTTPException(status_code=404, detail="No such event")
447 return event
450@app.get("/number-events", dependencies=[Depends(get_current_active_user)])
451async def get_number_events(
452 min_timestamp: float | int, max_timestamp: float | int, recompute_events: bool = False
453) -> list[EventDay]:
454 return EventDay.get_number_events_timeframe(min_timestamp, max_timestamp, recompute_events)
457@app.get("/event-rules", dependencies=[Depends(get_current_active_user)])
458async def get_event_rules(query: EventRuleQuery = Depends()) -> ListResponse[EventRule]:
459 return EventRule.response_from_query(query)
462@app.get("/event-rules/{event_rule_id}", dependencies=[Depends(get_current_active_user)])
463async def get_event_rule(event_rule_id) -> EventRule:
464 event_rule = EventRule.get_from_id(event_rule_id)
465 if event_rule is None:
466 raise HTTPException(status_code=404, detail="No such event rule")
467 return event_rule
470@app.post("/users", status_code=201)
471async def create_user(user: User):
472 if User.get_one_by_attribute("user", user.email) is not None:
473 raise HTTPException(status_code=400, detail="An error occurred during account creation")
474 hashed_password = get_password_hash(user.password)
475 new_user = User.create(user.firstname, user.lastname, user.email, hashed_password, user.is_admin | False)
476 if new_user is None:
477 raise HTTPException(status_code=400, details="An error occurred during account creation")
478 return new_user
481@app.post("/token", status_code=201)
482async def login_for_access_token(
483 form_data: Annotated[OAuth2PasswordRequestForm, Depends()],
484) -> Token:
485 user = authenticate_user(form_data.username, form_data.password)
486 if not user:
487 raise HTTPException(status_code=401, detail="Bad Credentials", headers={"WWW-authenticate": "Bearer"})
488 access_token_expires = timedelta(minutes=float(ACCESS_TOKEN_EXPIRE_MINUTES))
489 if user.is_active:
490 raise HTTPException(status_code=402, detail="User Blocked", headers={"WWW-authenticate": "Bearer"})
491 access_token = create_access_token(
492 data={"sub": user.email, "admin": user.is_admin}, expires_delta=access_token_expires
493 )
494 return Token(access_token=access_token, token_type="bearer")
497@app.get("/users", dependencies=[Depends(get_current_active_user)])
498async def get_users():
499 return [u.to_dict(exclude={"password"}) for u in User.get_all(sort_by="email")]
502@app.get("/users/{user_id}", dependencies=[Depends(get_current_active_user)])
503async def get_user(user_id: str, current_user: Annotated[User, Depends(get_current_active_user)]):
504 user = None
506 if user_id == "me":
507 user = current_user
508 else:
509 user = User.get_from_id(user_id)
511 if user is None:
512 raise HTTPException(
513 status_code=404,
514 detail="User not found",
515 )
516 return user
519@app.patch("/users/{user_id}", response_model=UserUpdate, dependencies=[Depends(get_current_active_user)])
520async def patch_user(user: UserUpdate, user_id):
521 if user.password == "" or user.password is None:
522 del user.password
523 else:
524 user.password = get_password_hash(user.password)
525 return User.update(user, user_id)
528@app.post("/users/me", response_model=User)
529async def read_users_me(
530 current_user: Annotated[User, Depends(get_current_active_user)],
531):
532 del (current_user.password, current_user.is_active, current_user.is_connected)
533 return current_user
536@app.get("/campaigns", response_model=list[Campaign], dependencies=[Depends(get_current_active_user)])
537async def get_campaigns():
538 return Campaign.get_all()
541@app.get("/campaigns/{campaign_id}", response_model=Campaign, dependencies=[Depends(get_current_active_user)])
542async def get_campaign_by_id(campaign_id: str):
543 campaign = Campaign.get_from_id(campaign_id)
544 if campaign is None:
545 raise HTTPException(status_code=500, detail="An error occurred retrieving campaign")
546 return campaign
549@app.post("/campaigns", dependencies=[Depends(get_current_active_user)], status_code=201)
550async def add_campaign(campaign: Campaign):
551 new_campaign = Campaign.create(campaign)
552 if new_campaign is None:
553 raise HTTPException(status_code=500, detail="An error occurred during campaign creation")
554 return new_campaign
557@app.patch("/campaigns/{campaign_id}", response_model=Campaign, dependencies=[Depends(get_current_active_user)])
558async def edit_campaign(campaign_id: str, edit_campaign: Campaign):
559 campaign = Campaign.get_from_id(campaign_id)
560 if campaign is None:
561 raise HTTPException(status_code=500, detail="An error occurred during campaign edition")
562 campaign.name = edit_campaign.name
563 campaign.description = edit_campaign.description
564 return Campaign.update(campaign)
567@app.delete(
568 "/campaigns/{campaign_id}", response_model=bool, dependencies=[Depends(get_current_active_user)], status_code=200
569)
570async def delete_campaign(campaign_id: str):
571 exception = HTTPException(status_code=500, detail="An error occurred during campaign deletion")
572 campaign = Campaign.get_from_id(campaign_id)
573 if campaign is None:
574 raise exception
575 delete_phases = Phase.deleteMany(campaign_id)
576 if not delete_phases.acknowledged:
577 raise exception
578 campaign_deleted = Campaign.delete(campaign_id)
579 return campaign_deleted.acknowledged
582@app.get("/campaigns/{campaign_id}/phases", response_model=list[Phase], dependencies=[Depends(get_current_active_user)])
583async def get_campaign_phases(campaign_id: str):
584 return Phase.get_all(campaign_id)
587@app.get("/phases/{phase_id}", response_model=Phase, dependencies=[Depends(get_current_active_user)])
588async def get_phase(phase_id: str):
589 return Phase.get_from_id(phase_id)
592@app.post("/phases", dependencies=[Depends(get_current_active_user)], status_code=201)
593async def add_phase(phase: Phase):
594 new_phase = Phase.create(phase)
595 if new_phase is None:
596 raise HTTPException(status_code=500, detail="An error occurred during phase creation")
597 return new_phase
600@app.patch("/phases/{phase_id}", response_model=Phase, dependencies=[Depends(get_current_active_user)])
601async def edit_phase(phase_id, edit_phase: Phase):
602 phase = Phase.get_from_id(phase_id)
603 if phase is None:
604 raise HTTPException(status_code=500, detail="An error occurred during Phase edition")
605 phase.name = edit_phase.name
606 phase.description = edit_phase.description
607 phase.start_at = edit_phase.start_at
608 phase.end_at = edit_phase.end_at
609 return Phase.update(phase)
612@app.delete("/phases/{phase_id}", dependencies=[Depends(get_current_active_user)], status_code=200)
613async def delete_phase(phase_id: str):
614 phase = Phase.get_from_id(phase_id)
615 if phase is None:
616 raise HTTPException(status_code=500, detail="An error occurred during Phase deletion")
617 phase_deleted = Phase.delete(phase_id)
618 return phase_deleted.acknowledged
621@app.get("/custom-views", dependencies=[Depends(get_current_active_user)])
622async def get_custom_views():
623 return CustomView.get_all()
626@app.get("/users/{user_id}/custom-views", dependencies=[Depends(get_current_active_user)])
627async def get_custom_views_from_user_id(user_id: str):
628 return CustomView.get_by_attribute("user_id", user_id)
631@app.get("/custom-views/{custom_view_id}", dependencies=[Depends(get_current_active_user)])
632async def get_custom_view(custom_view_id: str):
633 return CustomView.get_from_id(custom_view_id)
636@app.post("/custom-views", dependencies=[Depends(get_current_active_user)])
637async def create_custom_view(
638 custom_view_creation: CustomViewCreation, current_user: User = Depends(get_current_active_user)
639):
640 custom_view = CustomView(**custom_view_creation.to_dict(), user_id=current_user.id)
641 custom_view.insert()
642 return custom_view
645@app.patch("/custom-views/{custom_view_id}", dependencies=[Depends(get_current_active_user)])
646async def update_custom_views(custom_view_id: str, custom_view_update: CustomViewUpdate):
647 custom_view = CustomView.get_from_id(custom_view_id)
648 return custom_view.update(custom_view_update.model_dump())
651@app.delete("/custom-views/{custom_view_id}", response_model=bool, dependencies=[Depends(get_current_active_user)])
652async def delete_custom_view(custom_view_id: str):
653 custom_view = CustomView.get_from_id(custom_view_id)
654 return custom_view.delete()
657@app.post("/videos", response_model=Video, dependencies=[Depends(get_current_active_user)])
658async def add_video(video: Video):
659 video.insert()
660 if not video:
661 raise HTTPException(status_code=500, detail="An error occurred during cctv creation")
662 return video
665@app.get("/videos", dependencies=[Depends(get_current_active_user)])
666async def get_videos():
667 return Video.get_all()
670@app.get("/videos/{video_id}", dependencies=[Depends(get_current_active_user)])
671def get_stream(video_id):
672 camera_name = Video.get_video(video_id)
673 if not camera_name:
674 raise HTTPException(status_code=404, detail="Camera not found")
675 return camera_name