Coverage for / usr / local / lib / python3.14 / site-packages / twinpad_backend / api.py: 95%
552 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-01-06 16:39 +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 DeviceDeployerUpdate,
19 DeviceFromDeployer,
20 DeviceFromDeployerCreation,
21 DeviceId,
22 DeviceUpdateFromDeployer,
23 MongoId,
24 Signal,
25 ForcedSignal,
26 SignalData,
27 SignalSample,
28 ServicesStatus,
29 Device,
30 DeviceUpdate,
31 DeviceSetup,
32 DeviceSetupUpdate,
33 DeviceState,
34 SignalUpdate,
35 SignalsData,
36 Event,
37 EventRule,
38 TwinPadActivity,
39 User,
40 UserUpdate,
41 Campaign,
42 Phase,
43 CustomView,
44 Command,
45 CustomViewCreation,
46 CustomViewUpdate,
47 Video,
48 SignalsPreset,
49 SignalsPresetCreation,
50 SignalsPresetUpdate,
51 PrivateGraphTheme,
52 PublicGraphTheme,
53 GraphThemeCreation,
54 GraphThemeUpdate,
55 DeviceDeployer,
56 SINGLE_POST_PROCESSING_FUNCTION,
57 DOUBLE_POST_PROCESSING_FUNCTION,
58 MULTIPLE_POST_PROCESSING_FUNCTION,
59)
60from twinpad_backend.auth import (
61 Token,
62 authenticate_user,
63 get_current_active_user,
64 ACCESS_TOKEN_EXPIRE_MINUTES,
65 create_access_token,
66 get_password_hash,
67)
68from twinpad_backend.queries import (
69 SignalQuery,
70 ForcedSignalQuery,
71 DeviceStatesQuery,
72 EventQuery,
73 EventRuleQuery,
74 CommandQuery,
75 GraphThemeQuery,
76)
77from twinpad_backend.responses import ListResponse
79REQUEST_TIME_WARNING = 0.5
81DEBUG = os.environ.get("DEBUG", "false") == "true"
82PROFILING = os.environ.get("PROFILING", "false") == "true"
84logger = logging.getLogger("uvicorn.error")
85logger.propagate = False
86logger.info("Debug mode: %s", DEBUG)
87logger.info("log level: %s", logging.root.level)
90app = FastAPI(title="Twinpad backend", version=__version__)
92app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"])
94if PROFILING: # pragma: no cover
95 profiling_folder = mkdtemp()
96 logger.info("Profiling enabled")
98 @app.middleware("http")
99 async def profile_request(request: Request, call_next):
100 should_profile = True
101 url = str(request.url)
102 for segment in ("profiling", ".ico"):
103 if segment in url:
104 should_profile = False
105 break
107 if should_profile: # avoid recursion
108 profiler = Profiler()
109 profiler.start()
111 response = await call_next(request)
113 profiler.stop()
114 url = "_".join(url.split("/")[3:]).rstrip("/")
115 if not url:
116 url = "slash"
117 filename = url.split("?", maxsplit=1)[0]
118 logger.info("saving profiling to %s", filename)
119 with open(os.path.join(profiling_folder, filename), "w", encoding="utf-8") as profiling_file:
120 profiling_file.write(profiler.output_html())
122 return response
124 return await call_next(request)
126 @app.get("/profilings")
127 async def profilings():
128 return {"profilings": os.listdir(profiling_folder)}
130 @app.get("/profilings/{file_name}")
131 async def profiling(file_name):
132 file_path = os.path.join(profiling_folder, file_name)
134 if not os.path.exists(file_path):
135 raise HTTPException(
136 status_code=404,
137 detail=f"Profiling file '{file_name}' not found",
138 )
140 with open(file_path, "r", encoding="utf-8") as profiling_file:
141 return Response(
142 content=profiling_file.read(),
143 media_type="application/html",
144 headers={"Content-Disposition": f'attachment; filename="{file_name}_profiling.html"'},
145 )
148@app.middleware("http")
149async def log_request_time(request: Request, call_next):
150 start_time = time.time() # Record the start time
151 response = await call_next(request) # Process the request
152 duration = time.time() - start_time # Calculate the time taken
153 client_ip = request.headers.get("x-forwarded-for", request.client.host)
154 message = f"{client_ip} {request.method} {request.url.path} - {response.status_code} - {round(1000*duration)}ms"
155 if duration > REQUEST_TIME_WARNING:
156 logger.warning(message)
157 else:
158 logger.info(message)
159 return response
162@app.get("/")
163async def slash():
164 return {"twinpad_version": __version__}
167@app.get("/status", dependencies=[Depends(get_current_active_user)])
168async def status():
169 """
170 Return service healthcheck
171 """
172 return {
173 "services": ServicesStatus.check(),
174 }
177@app.get("/devices", dependencies=[Depends(get_current_active_user)])
178async def get_devices() -> list[Device]:
179 return Device.get_all(sort_by="device_id")
182@app.get("/devices/{device_id}", dependencies=[Depends(get_current_active_user)])
183async def get_device(device_id) -> Device:
184 device = Device.get_one_by_attribute("device_id", device_id)
185 if not device:
186 raise HTTPException(
187 status_code=404,
188 detail="Device not found",
189 )
190 return device
193@app.patch("/devices/{device_id}", dependencies=[Depends(get_current_active_user)])
194async def update_item(
195 device_id: DeviceId, device_update: DeviceUpdate, current_user: Annotated[User, Depends(get_current_active_user)]
196):
197 device = Device.get_one_by_attribute("device_id", device_id)
198 if not device:
199 raise HTTPException(
200 status_code=404,
201 detail="Device not found",
202 )
203 result = await device.change_mode(device_update, current_user)
204 if result.get("error", False) is True:
205 raise HTTPException(
206 status_code=result.get("status_code", 500),
207 detail=result.get("message", "An error has occurred"),
208 )
209 return result
212@app.get("/devices/{device_id}/states", dependencies=[Depends(get_current_active_user)])
213async def get_device_states(device_id: DeviceId, query: DeviceStatesQuery = Depends()) -> ListResponse[DeviceState]:
214 return DeviceState.get_from_id_and_query(device_id, query)
217@app.get("/device-setups", dependencies=[Depends(get_current_active_user)])
218async def get_device_setups() -> list[DeviceSetup]:
219 return DeviceSetup.get_all()
222@app.post("/device-setups", dependencies=[Depends(get_current_active_user)], status_code=201)
223async def create_device_setups(device_setup: DeviceSetup) -> DeviceSetup:
224 device_setup.insert()
225 return device_setup
228@app.get("/device-setups/{device_setup_id}", dependencies=[Depends(get_current_active_user)])
229async def get_device_setup(device_setup_id: str):
230 device_setup = DeviceSetup.get_from_id(device_setup_id)
231 if device_setup is None:
232 raise HTTPException(
233 status_code=404,
234 detail="Device setup not found",
235 )
236 return device_setup
239@app.patch("/device-setups/{device_setup_id}", dependencies=[Depends(get_current_active_user)])
240async def edit_device_setups(device_setup_id: str, device_setup_update: DeviceSetupUpdate) -> DeviceSetup:
241 device_setup = DeviceSetup.get_from_id(device_setup_id)
242 if device_setup is None:
243 raise HTTPException(
244 status_code=404,
245 detail="Device setup not found",
246 )
247 device_setup.update({k: v for k, v in device_setup_update.model_dump().items() if v is not None})
248 return device_setup
251@app.delete("/device-setups/{device_setup_id}", dependencies=[Depends(get_current_active_user)], status_code=200)
252async def delete_device_setups(device_setup_id: str) -> bool:
253 device_setup = DeviceSetup.get_from_id(device_setup_id)
254 if device_setup is None:
255 raise HTTPException(
256 status_code=404,
257 detail="Device setup not found",
258 )
259 deleted = device_setup.delete()
260 return deleted
263@app.get("/number-samples", dependencies=[Depends(get_current_active_user)])
264async def get_number_samples(
265 min_timestamp: float | int, max_timestamp: float | int, recompute_amount: bool = False
266) -> list[TwinPadActivity]:
267 return TwinPadActivity.get_number_samples_timeframe(min_timestamp, max_timestamp, recompute_amount)
270@app.get("/signals", dependencies=[Depends(get_current_active_user)])
271async def route_get_signals(query: SignalQuery = Depends()) -> ListResponse[Signal]:
272 if "signal_id" not in query.sort_by:
273 query.sort_by += ",signal_id:1"
274 return Signal.response_from_query(query).to_dict(exclude={"device"})
277@app.get("/signals/ids", dependencies=[Depends(get_current_active_user)])
278async def signals_names() -> list[str]:
279 return Signal.get_all_ids()
282@app.get("/signals/stats", dependencies=[Depends(get_current_active_user)])
283async def signal_stats():
284 """
285 Returns signals stats
286 """
287 signal_statuses = Signal.get_all_statuses()
288 signal_ids = [signal["signal_id"] for signal in signal_statuses]
290 number_samples_by_signal_id = await Signal.number_samples_batch(signal_ids)
291 number_samples = sum(number_samples_by_signal_id.values())
293 number_active_signals = sum(1 for signal in signal_statuses if signal["status"] == "up")
295 number_signals = Signal.get_number_documents()
297 return {
298 "signal_data_size": signal_datasize(),
299 "number_signal_samples": number_samples,
300 "number_active_signals": number_active_signals,
301 "number_signals": number_signals,
302 }
305@app.get("/signals/last-value", dependencies=[Depends(get_current_active_user)])
306async def get_last_values(signal_ids: list[str] = Query(default=[])) -> list[SignalSample | None]:
307 return SignalSample.get_last_from_signal_ids(signal_ids)
310@app.get("/signals/first-value", dependencies=[Depends(get_current_active_user)])
311async def get_first_values(signal_ids: list[str] = Query(default=[])) -> list[SignalSample | None]:
312 return SignalSample.get_first_from_signal_ids(signal_ids)
315@app.get("/signals/forcibility", dependencies=[Depends(get_current_active_user)])
316async def get_signals_forcibility(signal_ids: list[str] = Query(default=[])) -> dict[str, bool]:
317 return Signal.get_forcibility(signal_ids)
320@app.get("/signals/forced", response_model=ListResponse[ForcedSignal], dependencies=[Depends(get_current_active_user)])
321async def get_forced_signals(
322 current_user: Annotated[User, Depends(get_current_active_user)], query: ForcedSignalQuery = Depends()
323):
324 if not current_user.is_admin:
325 raise HTTPException(401)
326 return ForcedSignal.response_from_query(query)
329@app.get("/signals/{signal_id}", dependencies=[Depends(get_current_active_user)])
330async def get_signal(signal_id):
331 signal = Signal.get_from_signal_id(signal_id)
332 if not signal:
333 raise HTTPException(
334 status_code=404,
335 detail="Signal not found",
336 )
337 return signal.to_dict()
340@app.patch("/signals/{signal_id}", dependencies=[Depends(get_current_active_user)])
341async def update_signal(
342 signal_id: str, signal_update: SignalUpdate, current_user: Annotated[User, Depends(get_current_active_user)]
343):
344 signal = Signal.get_from_signal_id(signal_id)
345 if not signal:
346 raise HTTPException(
347 status_code=404,
348 detail="Device not found",
349 )
350 forced_signal = ForcedSignal.get_one_by_attribute("signal_id", signal_id)
351 if forced_signal is not None:
352 if forced_signal.forcing_user_id != current_user.id and not current_user.is_admin:
353 raise HTTPException(
354 status_code=403,
355 detail="Cannot override another user's forcing",
356 )
358 result = await signal.send_command(signal_update, current_user)
359 if result.get("error", False) is True:
360 raise HTTPException(
361 status_code=result.get("status_code", 500),
362 detail=result.get("message", "An error has occurred"),
363 )
365 if forced_signal is not None and signal_update.forced_value is None:
366 forced_signal.delete()
367 elif signal_update.forced_value is not None:
368 forced_signal = ForcedSignal(
369 signal_id=signal_id,
370 forcing_user_id=current_user.id,
371 forced_at=time.time(),
372 value=signal_update.forced_value,
373 )
374 forced_signal.insert()
376 return result
379@app.get("/signals/{signal_id}/can-force", dependencies=[Depends(get_current_active_user)])
380async def get_signal_forcibility(
381 signal_id: str, current_user: Annotated[User, Depends(get_current_active_user)]
382) -> bool:
383 return ForcedSignal.can_force(signal_id, current_user)
386@app.get("/signals/{signal_id}/data", dependencies=[Depends(get_current_active_user)])
387async def get_signal_data(
388 signal_id: str, number_samples_max: int = None, min_timestamp: float = None, max_timestamp: float = None
389) -> SignalData | None:
390 signal_data = SignalData.get_from_signal_id(signal_id, min_timestamp=min_timestamp, max_timestamp=max_timestamp)
392 if number_samples_max is not None:
393 signal_data = signal_data.uniform_desampling(number_samples_max=number_samples_max)
395 return signal_data
398@app.get("/signals/{signal_id}/last-value", dependencies=[Depends(get_current_active_user)])
399async def get_last_value(signal_id) -> SignalSample:
400 sample = SignalSample.get_last_from_signal_id(signal_id)
401 if sample is None:
402 raise HTTPException(status_code=404, detail="No data")
403 return sample
406@app.get("/signals/{signal_id}/first-value", dependencies=[Depends(get_current_active_user)])
407async def get_first_value(signal_id) -> SignalSample:
408 sample = SignalSample.get_first_from_signal_id(signal_id)
409 if sample is None:
410 raise HTTPException(status_code=404, detail="No data")
411 return sample
414@app.get("/signals/{signal_id}/number-samples", dependencies=[Depends(get_current_active_user)])
415async def get_signal_number_samples(signal_id):
416 signal = Signal.get_from_signal_id(signal_id)
417 if not signal:
418 raise HTTPException(
419 status_code=404,
420 detail="Device not found",
421 )
422 return {"signal_id": signal_id, "number_samples": await signal.number_samples(), "size": signal.sample_datasize()}
425@app.get("/signals-data", dependencies=[Depends(get_current_active_user)])
426async def get_signals_data(
427 signal_ids: list[str] = Query(default=[]),
428 number_samples_max: int = None,
429 min_timestamp: float = None,
430 max_timestamp: float = None,
431 interpolate_bounds: bool = True,
432) -> SignalsData | None:
433 if min_timestamp and max_timestamp and min_timestamp > max_timestamp:
434 raise HTTPException(status_code=400, detail="min_timestamp should be less than max_timestamp")
436 signals_data = SignalsData.get_from_signal_ids(
437 signal_ids,
438 min_timestamp=min_timestamp,
439 max_timestamp=max_timestamp,
440 window_min_timestamp=min_timestamp,
441 window_max_timestamp=max_timestamp,
442 interpolate_bounds=interpolate_bounds,
443 )
444 if number_samples_max is not None:
445 signals_data = signals_data.uniform_desampling(number_samples_max=number_samples_max)
447 return signals_data
450@app.get("/signals-data/interest-window", dependencies=[Depends(get_current_active_user)])
451async def get_signals_data_interest_window(
452 window_max_number_samples: int | None = None,
453 outside_max_number_samples: int | None = None,
454 window_min_timestamp: float = None,
455 window_max_timestamp: float = None,
456 signal_ids: list[str] = Query(default=[]),
457 min_timestamp: float = None,
458 max_timestamp: float = None,
459) -> SignalsData | None:
460 if window_min_timestamp and window_max_timestamp and window_min_timestamp > window_max_timestamp:
461 raise HTTPException(status_code=400, detail="window_min_timestamp should be less than window_max_timestamp")
463 if min_timestamp and max_timestamp and min_timestamp > max_timestamp:
464 raise HTTPException(status_code=400, detail="min_timestamp should be less than max_timestamp")
466 max_documents = 0
468 if window_max_number_samples is not None:
469 max_documents += 10 * window_max_number_samples
470 if outside_max_number_samples is not None:
471 max_documents += 10 * outside_max_number_samples
473 if max_documents == 0:
474 max_documents = None
476 signals_data = SignalsData.get_from_signal_ids(
477 signal_ids,
478 min_timestamp=min_timestamp,
479 max_timestamp=max_timestamp,
480 window_min_timestamp=window_min_timestamp,
481 window_max_timestamp=window_max_timestamp,
482 max_documents=max_documents,
483 )
485 signals_data = signals_data.interest_window_desampling(
486 window_max_number_samples=window_max_number_samples,
487 outside_max_number_samples=outside_max_number_samples,
488 window_min_timestamp=window_min_timestamp,
489 window_max_timestamp=window_max_timestamp,
490 )
492 return signals_data
495@app.get("/signals-data/export_zip", dependencies=[Depends(get_current_active_user)])
496async def export_signals_zip(
497 file_format: str,
498 signal_ids: list[str] = Query(default=[]),
499 min_timestamp: float = None,
500 max_timestamp: float = None,
501):
502 signals_data = SignalsData.get_from_signal_ids(
503 signal_ids, min_timestamp=min_timestamp, max_timestamp=max_timestamp, interpolate_bounds=False
504 )
505 zip_data = signals_data.zip_export(file_format)
506 return Response(
507 content=zip_data,
508 media_type="application/zip",
509 headers={"Content-Disposition": 'attachment; filename="signals.zip"'},
510 )
513@app.get("/signals-data/export_hdf5", dependencies=[Depends(get_current_active_user)])
514async def export_signals_hdf5(
515 signal_ids: list[str] = Query(default=[]),
516 min_timestamp: float = None,
517 max_timestamp: float = None,
518):
519 signals_data = SignalsData.get_from_signal_ids(
520 signal_ids, min_timestamp=min_timestamp, max_timestamp=max_timestamp, interpolate_bounds=False
521 )
522 data = signals_data.hdf5_export()
523 return Response(
524 content=data,
525 media_type="application/hdf5",
526 headers={"Content-Disposition": 'attachment; filename="signals.hdf5"'},
527 )
530@app.get("/post-processing/signals-data", dependencies=[Depends(get_current_active_user)])
531async def get_signals_data_post_processing(
532 phase_ids: list[str] = Query(default=[]),
533 phase_sync_times: list[float | None] = Query(default=[]),
534 signal_ids: list[str] = Query(default=[]),
535 window_min_timestamps: list[float | None] = Query(default=[]),
536 window_max_timestamps: list[float | None] = Query(default=[]),
537 number_samples_max: int = None,
538) -> SignalsData | None:
539 if len(phase_sync_times) == 0:
540 phase_sync_times = [None for _ in range(len(phase_ids))]
541 if len(window_min_timestamps) == 0:
542 window_min_timestamps = [None for _ in range(len(phase_ids))]
543 if len(window_max_timestamps) == 0:
544 window_max_timestamps = [None for _ in range(len(phase_ids))]
546 if (
547 len(phase_ids) != len(phase_sync_times)
548 or len(phase_ids) != len(window_min_timestamps)
549 or len(phase_ids) != len(window_max_timestamps)
550 ):
551 raise HTTPException(
552 400, "Each phase should have corresponding synchronization time, minimum and maximum timestamps."
553 )
555 phases = [Phase.get_from_id(phase_id) for phase_id in phase_ids]
557 if None in phases:
558 raise HTTPException(404, "Phase not found")
560 signals_data = SignalsData.get_from_phase_and_signal_ids(
561 phases=phases,
562 phase_sync_times=phase_sync_times,
563 signal_ids=signal_ids,
564 window_min_timestamps=window_min_timestamps,
565 window_max_timestamps=window_max_timestamps,
566 )
568 if number_samples_max is not None:
569 signals_data = signals_data.min_max_downsampling(number_samples_max)
571 return signals_data
574@app.get("/post-processing/functions/single", dependencies=[Depends(get_current_active_user)])
575async def apply_single_post_processing_function(
576 phase_id: str,
577 base_signal_id: str,
578 function: SINGLE_POST_PROCESSING_FUNCTION,
579 phase_sync_time: float = None,
580 window_min_timestamp: float = None,
581 window_max_timestamp: float = None,
582 number_samples_max: int = None,
583) -> SignalsData | None:
584 phase = Phase.get_from_id(phase_id)
586 if phase is None:
587 raise HTTPException(404, "Phase not found")
588 if phase_sync_time is None:
589 phase_sync_time = phase.start_at / 1000
591 signals_data = await SignalsData.apply_single_function(
592 phase,
593 base_signal_id,
594 function,
595 window_min_timestamp=window_min_timestamp,
596 window_max_timestamp=window_max_timestamp,
597 )
599 if signals_data is None:
600 raise HTTPException(500, "There was en error while applying the function")
602 if number_samples_max is not None and number_samples_max < len(signals_data.signals_data[0].time_vector):
603 signals_data = signals_data.min_max_downsampling(number_samples_max)
604 signals_data = signals_data.zero_time_vector(phase_sync_time)
606 return signals_data
609@app.get("/post-processing/functions/multiple", dependencies=[Depends(get_current_active_user)])
610async def apply_multiple_post_processing_function(
611 function: DOUBLE_POST_PROCESSING_FUNCTION | MULTIPLE_POST_PROCESSING_FUNCTION,
612 phase_ids: list[str] = Query(default=[]),
613 phase_sync_times: list[float] = Query(default=[]),
614 signal_ids: list[str] = Query(default=[]),
615 window_min_timestamp: float = None,
616 window_max_timestamp: float = None,
617 number_samples_max: int = None,
618) -> SignalsData | None:
619 if len(phase_ids) != len(signal_ids):
620 raise HTTPException(400, "Each selected signal should correspond to a phase")
622 if len(phase_ids) < 2:
623 raise HTTPException(400, "These functions can only be applied to multiple signals")
625 phases = [Phase.get_from_id(phase_id) for phase_id in phase_ids]
626 if None in phases:
627 raise HTTPException(404, "Phase not found")
629 if len(phase_sync_times) == 0:
630 phase_sync_times = [phase.start_at / 1000 for phase in phases]
631 if len(phases) != len(phase_sync_times):
632 raise HTTPException(400, "Number of synchronization times does not match the number of phases")
634 signals_data = await SignalsData.apply_multiple_function(
635 phases,
636 signal_ids,
637 function,
638 window_min_timestamp=window_min_timestamp,
639 window_max_timestamp=window_max_timestamp,
640 )
642 if signals_data is None:
643 raise HTTPException(500, "There was en error while applying the function")
645 if number_samples_max is not None and number_samples_max < len(signals_data.signals_data[0].time_vector):
646 signals_data = signals_data.min_max_downsampling(number_samples_max)
647 if function in {"Align-X", "Using-X"}:
648 signals_data = signals_data.zero_time_vector(phase_sync_times[1])
649 else:
650 signals_data = signals_data.zero_time_vector(phase_sync_times[0])
652 return signals_data
655@app.get("/post-processing/export_zip", dependencies=[Depends(get_current_active_user)])
656async def export_post_processing_zip(
657 file_format: str,
658 phase_ids: list[MongoId] = Query(default=[]),
659 phase_sync_times: list[float | None] = Query(default=[]),
660 signal_ids: list[str] = Query(default=[]),
661 window_min_timestamps: list[float | None] = Query(default=[]),
662 window_max_timestamps: list[float | None] = Query(default=[]),
663):
664 signals_data = await get_signals_data_post_processing(
665 phase_ids,
666 phase_sync_times,
667 signal_ids,
668 window_min_timestamps,
669 window_max_timestamps,
670 )
672 zip_data = signals_data.zip_export(file_format, post_processing=True, phase_ids=phase_ids)
674 return Response(
675 content=zip_data,
676 media_type="application/zip",
677 headers={"Content-Disposition": 'attachment; filename="signals.zip"'},
678 )
681@app.get("/post-processing/export_hdf5", dependencies=[Depends(get_current_active_user)])
682async def export_post_processing_hdf5(
683 phase_ids: list[str] = Query(default=[]),
684 phase_sync_times: list[float | None] = Query(default=[]),
685 signal_ids: list[str] = Query(default=[]),
686 window_min_timestamps: list[float | None] = Query(default=[]),
687 window_max_timestamps: list[float | None] = Query(default=[]),
688):
689 signals_data = await get_signals_data_post_processing(
690 phase_ids,
691 phase_sync_times,
692 signal_ids,
693 window_min_timestamps,
694 window_max_timestamps,
695 )
697 data = signals_data.hdf5_export(post_processing=True, phase_ids=phase_ids)
699 return Response(
700 content=data,
701 media_type="application/hdf5",
702 headers={"Content-Disposition": 'attachment; filename="signals.hdf5"'},
703 )
706@app.get("/events", dependencies=[Depends(get_current_active_user)])
707async def get_events(query: EventQuery = Depends()) -> ListResponse[Event]:
708 return Event.response_from_query(query)
711@app.get("/events/{event_id}", dependencies=[Depends(get_current_active_user)])
712async def get_event(event_id) -> Event:
713 event = Event.get_from_id(event_id)
714 if event is None:
715 raise HTTPException(status_code=404, detail="No such event")
716 return event
719@app.get("/number-events", dependencies=[Depends(get_current_active_user)])
720async def get_number_events(
721 min_timestamp: float | int, max_timestamp: float | int, recompute_amount: bool = False
722) -> list[TwinPadActivity]:
723 return TwinPadActivity.get_number_events_timeframe(min_timestamp, max_timestamp, recompute_amount)
726@app.get("/number-commands", dependencies=[Depends(get_current_active_user)])
727async def get_number_commands(
728 min_timestamp: float | int, max_timestamp: float | int, recompute_amount: bool = False
729) -> list[TwinPadActivity]:
730 return TwinPadActivity.get_number_commands_timeframe(min_timestamp, max_timestamp, recompute_amount)
733@app.get("/event-rules", dependencies=[Depends(get_current_active_user)])
734async def get_event_rules(query: EventRuleQuery = Depends()) -> ListResponse[EventRule]:
735 return EventRule.response_from_query(query)
738@app.get("/event-rules/{event_rule_id}", dependencies=[Depends(get_current_active_user)])
739async def get_event_rule(event_rule_id) -> EventRule:
740 event_rule = EventRule.get_from_id(event_rule_id)
741 if event_rule is None:
742 raise HTTPException(status_code=404, detail="No such event rule")
743 return event_rule
746@app.post("/users", status_code=201)
747async def create_user(user: User):
748 if User.get_one_by_attribute("email", user.email) is not None:
749 raise HTTPException(status_code=400, detail="An error occurred during account creation")
750 hashed_password = get_password_hash(user.password)
751 new_user = User.create(user.firstname, user.lastname, user.email, hashed_password, user.is_admin | False)
752 if new_user is None:
753 raise HTTPException(status_code=400, detail="An error occurred during account creation")
754 return new_user
757@app.post("/token", status_code=201)
758async def login_for_access_token(
759 form_data: Annotated[OAuth2PasswordRequestForm, Depends()],
760) -> Token:
761 user = authenticate_user(form_data.username, form_data.password)
762 if not user:
763 raise HTTPException(status_code=401, detail="Bad Credentials", headers={"WWW-authenticate": "Bearer"})
764 access_token_expires = timedelta(minutes=float(ACCESS_TOKEN_EXPIRE_MINUTES))
765 if user.is_active:
766 raise HTTPException(status_code=402, detail="User Blocked", headers={"WWW-authenticate": "Bearer"})
767 access_token = create_access_token(
768 data={"sub": user.email, "admin": user.is_admin}, expires_delta=access_token_expires
769 )
770 return Token(access_token=access_token, token_type="bearer")
773@app.get("/users", dependencies=[Depends(get_current_active_user)])
774async def get_users():
775 return [u.to_dict() for u in User.get_all(sort_by="email")]
778@app.get("/users/me")
779async def get_current_user(
780 current_user: Annotated[User, Depends(get_current_active_user)],
781):
782 return current_user.to_dict()
785@app.get("/users/{user_id}", dependencies=[Depends(get_current_active_user)])
786async def get_user(user_id: str):
787 user = User.get_from_id(user_id)
789 if user is None:
790 raise HTTPException(
791 status_code=404,
792 detail="User not found",
793 )
794 return user.to_dict(exclude={"password", "is_connected"})
797@app.patch("/users/{user_id}", dependencies=[Depends(get_current_active_user)])
798async def patch_user(user: UserUpdate, user_id):
799 if user.password == "" or user.password is None:
800 del user.password
801 else:
802 user.password = get_password_hash(user.password)
803 return User.update_info(user, user_id).to_dict()
806@app.get("/commands", dependencies=[Depends(get_current_active_user)])
807async def get_commands(query: CommandQuery = Depends()) -> ListResponse[Command]:
808 return Command.response_from_query(query)
811@app.get("/campaigns", response_model=list[Campaign], dependencies=[Depends(get_current_active_user)])
812async def get_campaigns():
813 return Campaign.get_all()
816@app.get("/campaigns/{campaign_id}", response_model=Campaign, dependencies=[Depends(get_current_active_user)])
817async def get_campaign_by_id(campaign_id: str):
818 campaign = Campaign.get_from_id(campaign_id)
819 if campaign is None:
820 raise HTTPException(status_code=404, detail="Campaign not found")
821 return campaign
824@app.post("/campaigns", dependencies=[Depends(get_current_active_user)], status_code=201)
825async def add_campaign(campaign: Campaign):
826 campaign_id = campaign.insert()
827 if campaign_id is None:
828 raise HTTPException(status_code=500, detail="An error occurred during campaign creation")
829 return campaign
832@app.patch("/campaigns/{campaign_id}", response_model=Campaign, dependencies=[Depends(get_current_active_user)])
833async def edit_campaign(campaign_id: str, edit_campaign: Campaign):
834 campaign = Campaign.get_from_id(campaign_id)
835 if campaign is None:
836 raise HTTPException(status_code=404, detail="Campaign not found")
837 campaign.update(edit_campaign.model_dump(exclude_unset=True, mode="json"))
838 return campaign
841@app.delete(
842 "/campaigns/{campaign_id}", response_model=bool, dependencies=[Depends(get_current_active_user)], status_code=200
843)
844async def delete_campaign(campaign_id: str):
845 campaign = Campaign.get_from_id(campaign_id)
846 if campaign is None:
847 raise HTTPException(status_code=404, detail="Campaign not found")
848 delete_phases = Phase.deleteMany(campaign_id)
849 if not delete_phases.acknowledged:
850 raise HTTPException(status_code=500, detail="An error occurred during phases deletion")
851 campaign_deleted = campaign.delete()
852 if not campaign_deleted:
853 raise HTTPException(status_code=500, detail="An error occurred during campaign deletion")
854 return True
857@app.get("/campaigns/{campaign_id}/phases", response_model=list[Phase], dependencies=[Depends(get_current_active_user)])
858async def get_campaign_phases(campaign_id: str):
859 return Phase.get_by_attribute("campaign_id", campaign_id)
862@app.get("/phases/{phase_id}", response_model=Phase, dependencies=[Depends(get_current_active_user)])
863async def get_phase(phase_id: str):
864 phase = Phase.get_from_id(phase_id)
865 if phase is None:
866 raise HTTPException(status_code=404, detail="Phase not found")
867 return phase
870@app.post("/phases", dependencies=[Depends(get_current_active_user)], status_code=201)
871async def add_phase(phase: Phase):
872 phase_id = phase.insert()
873 if phase_id is None:
874 raise HTTPException(status_code=500, detail="An error occurred during phase creation")
875 return phase
878@app.patch("/phases/{phase_id}", response_model=Phase, dependencies=[Depends(get_current_active_user)])
879async def edit_phase(phase_id, edit_phase: Phase):
880 phase = Phase.get_from_id(phase_id)
881 if phase is None:
882 raise HTTPException(status_code=404, detail="Phase does not exists")
883 phase.update(edit_phase.model_dump(exclude_unset=True, mode="json"))
884 return phase
887@app.delete("/phases/{phase_id}", dependencies=[Depends(get_current_active_user)], status_code=200)
888async def delete_phase(phase_id: str):
889 phase = Phase.get_from_id(phase_id)
890 if phase is None:
891 raise HTTPException(status_code=404, detail="Phase not found")
892 deleted = phase.delete()
893 if not deleted:
894 raise HTTPException(status_code=500, detail="An error occurred during phase deletion")
895 return True
898@app.get("/custom-views", dependencies=[Depends(get_current_active_user)])
899async def get_custom_views():
900 return CustomView.get_all()
903@app.get("/users/{user_id}/custom-views", dependencies=[Depends(get_current_active_user)])
904async def get_custom_views_from_user_id(user_id: str):
905 return CustomView.get_by_attribute("user_id", user_id)
908@app.get("/custom-views/{custom_view_id}", dependencies=[Depends(get_current_active_user)])
909async def get_custom_view(custom_view_id: str):
910 return CustomView.get_from_id(custom_view_id)
913@app.post("/custom-views", dependencies=[Depends(get_current_active_user)])
914async def create_custom_view(
915 custom_view_creation: CustomViewCreation, current_user: User = Depends(get_current_active_user)
916):
917 custom_view = CustomView(**custom_view_creation.to_dict(), user_id=current_user.id)
918 custom_view.insert()
919 return custom_view
922@app.patch("/custom-views/{custom_view_id}", dependencies=[Depends(get_current_active_user)])
923async def update_custom_views(custom_view_id: str, custom_view_update: CustomViewUpdate):
924 custom_view = CustomView.get_from_id(custom_view_id)
925 return custom_view.update(custom_view_update.model_dump())
928@app.delete("/custom-views/{custom_view_id}", response_model=bool, dependencies=[Depends(get_current_active_user)])
929async def delete_custom_view(custom_view_id: str):
930 custom_view = CustomView.get_from_id(custom_view_id)
931 return custom_view.delete()
934@app.post("/videos", response_model=Video, dependencies=[Depends(get_current_active_user)])
935async def add_video(video: Video):
936 video.insert()
937 if not video:
938 raise HTTPException(status_code=500, detail="An error occurred during cctv creation")
939 return video
942@app.get("/videos", dependencies=[Depends(get_current_active_user)])
943async def get_videos():
944 return Video.get_all()
947@app.get("/videos/{video_id}", dependencies=[Depends(get_current_active_user)])
948def get_stream(video_id):
949 camera_name = Video.get_video(video_id)
950 if camera_name is None:
951 raise HTTPException(status_code=404, detail="Camera not found")
952 return camera_name
955@app.get("/signals-presets", dependencies=[Depends(get_current_active_user)])
956async def get_signals_preset(current_user: User = Depends(get_current_active_user)):
957 return SignalsPreset.get_by_attribute("user_id", current_user.id)
960@app.post("/signals-presets", dependencies=[Depends(get_current_active_user)])
961async def create_signals_preset(
962 signals_preset: SignalsPresetCreation, current_user: User = Depends(get_current_active_user)
963):
964 new_signals_preset = SignalsPreset.create(signals_preset=signals_preset, user_id=current_user.id)
965 return new_signals_preset
968@app.patch("/signals-presets/{signals_preset_id}", dependencies=[Depends(get_current_active_user)])
969async def update_signals_preset(signals_preset_id: str, signals_preset_update: SignalsPresetUpdate):
970 signals_preset = SignalsPreset.get_from_id(signals_preset_id)
971 return signals_preset.update(signals_preset_update.model_dump(exclude_unset=True))
974@app.delete(
975 "/signals-presets/{signals_preset_id}", response_model=bool, dependencies=[Depends(get_current_active_user)]
976)
977async def delete_signals_preset(signals_preset_id: str):
978 signals_preset = SignalsPreset.get_from_id(signals_preset_id)
979 return signals_preset.delete()
982@app.post("/graph-themes", dependencies=[Depends(get_current_active_user)])
983async def create_graph_theme(
984 graph_theme_creation: GraphThemeCreation, current_user: User = Depends(get_current_active_user)
985):
986 styled_signal = Signal.get_one_by_attribute("signal_id", graph_theme_creation.signal_id)
987 if styled_signal is None:
988 raise HTTPException(400, f"Signal ID '{graph_theme_creation.signal_id}' doesn't exist")
990 graph_theme = PrivateGraphTheme.create(
991 **graph_theme_creation.to_dict(exclude={"id": True}), creator_id=current_user.id
992 )
993 return PublicGraphTheme.get_from_id(graph_theme.id, current_user.id)
996@app.get("/graph-themes", dependencies=[Depends(get_current_active_user)])
997async def get_all_graph_themes(
998 query: GraphThemeQuery = Depends(), current_user: User = Depends(get_current_active_user)
999) -> ListResponse[PublicGraphTheme]:
1000 return PublicGraphTheme.response_from_query(query, current_user.id)
1003@app.get("/graph-themes/own", dependencies=[Depends(get_current_active_user)])
1004async def get_graph_themes_in_library(
1005 query: GraphThemeQuery = Depends(), current_user: User = Depends(get_current_active_user)
1006) -> ListResponse[PublicGraphTheme]:
1007 return PublicGraphTheme.response_from_query_in_user_library(query, current_user.id)
1010@app.patch("/graph-themes/{theme_id}", dependencies=[Depends(get_current_active_user)])
1011async def update_graph_theme(
1012 theme_id: str, theme_update: GraphThemeUpdate, current_user: User = Depends(get_current_active_user)
1013):
1014 graph_theme = PrivateGraphTheme.get_from_id(theme_id)
1015 update_dict = theme_update.model_dump(exclude_unset=True)
1016 if current_user.id != graph_theme.creator_id:
1017 for theme_property in update_dict.keys():
1018 if theme_property not in ["active_for_user", "in_user_library"]:
1019 raise HTTPException(401, "User is not allowed to edit a theme which wasn't created by him")
1020 graph_theme.update(theme_update.model_dump(exclude_unset=True), current_user.id)
1021 return PublicGraphTheme.get_from_id(graph_theme.id, current_user.id)
1024@app.delete("/graph-themes/{graph_theme_id}", response_model=bool, dependencies=[Depends(get_current_active_user)])
1025async def delete_graph_theme(graph_theme_id: str, current_user: User = Depends(get_current_active_user)):
1026 graph_theme = PrivateGraphTheme.get_from_id(graph_theme_id)
1027 if current_user.id != graph_theme.creator_id:
1028 raise HTTPException(401, "User is not allowed to delete a theme which wasn't created by him")
1029 return graph_theme.delete()
1032@app.get("/signals-appearances", dependencies=[Depends(get_current_active_user)])
1033async def get_signals_appearances(
1034 signal_ids: list[str] = Query(default=[]), current_user: User = Depends(get_current_active_user)
1035) -> dict:
1036 return PublicGraphTheme.get_signal_appearances(signal_ids, current_user.id)
1039@app.get("/device-deployers", dependencies=[Depends(get_current_active_user)])
1040async def get_device_deployers() -> list[DeviceDeployer]:
1041 return DeviceDeployer.get_all()
1044@app.get("/device-deployers/{deployer_id}", dependencies=[Depends(get_current_active_user)])
1045async def get_device_deployers(deployer_id: MongoId) -> DeviceDeployer:
1046 deployer = DeviceDeployer.get_from_id(deployer_id)
1047 if deployer is None:
1048 raise HTTPException(
1049 status_code=404,
1050 detail=f"Deployer not found",
1051 )
1052 return deployer
1055@app.patch("/device-deployers/{deployer_id}", dependencies=[Depends(get_current_active_user)])
1056async def update_device_deployers(deployer_id: MongoId, deployer_update: DeviceDeployerUpdate) -> DeviceDeployer:
1057 deployer = DeviceDeployer.get_from_id(deployer_id)
1058 if deployer is None:
1059 raise HTTPException(
1060 status_code=404,
1061 detail=f"Deployer not found",
1062 )
1063 return deployer.update(deployer_update.model_dump(exclude_unset=True, mode="json"))
1066@app.post("/device-deployers/{deployer_id}/devices", dependencies=[Depends(get_current_active_user)], status_code=201)
1067async def create_device_from_deployer(deployer_id: MongoId, device: DeviceFromDeployerCreation) -> DeviceFromDeployer:
1068 deployer = DeviceDeployer.get_from_id(deployer_id)
1069 device = deployer.create_device(device)
1070 if device is None:
1071 raise HTTPException(status_code=400, detail="Cannot create device")
1072 return device
1075@app.post("/device-deployers", dependencies=[Depends(get_current_active_user)], status_code=201)
1076async def add_deployer(deployer: DeviceDeployer):
1077 deployer.insert()
1078 return deployer
1081@app.delete("/device-deployers/{deployer_id}", dependencies=[Depends(get_current_active_user)])
1082async def add_deployer(deployer_id: MongoId):
1083 deployer = DeviceDeployer.get_from_id(deployer_id)
1084 if deployer is None:
1085 raise HTTPException(
1086 status_code=404,
1087 detail=f"Deployer not found",
1088 )
1089 deleted_info = deployer.delete()
1090 return deleted_info
1093@app.get("/device-deployers/{deployer_id}/devices", dependencies=[Depends(get_current_active_user)])
1094async def get_devices_from_deployer(deployer_id: MongoId) -> list[DeviceFromDeployer]:
1095 deployer = DeviceDeployer.get_from_id(deployer_id)
1096 if deployer is None:
1097 raise HTTPException(
1098 status_code=404,
1099 detail=f"Deployer not found",
1100 )
1101 devices = deployer.devices()
1102 if devices is None:
1103 raise HTTPException(
1104 status_code=500,
1105 detail="Error getting devices",
1106 )
1107 return devices
1110@app.get("/device-deployers/{deployer_id}/devices/{device_id}", dependencies=[Depends(get_current_active_user)])
1111async def update_device_from_deployer(deployer_id: MongoId, device_id: DeviceId) -> DeviceFromDeployer:
1112 deployer = DeviceDeployer.get_from_id(deployer_id)
1113 if deployer is None:
1114 raise HTTPException(
1115 status_code=404,
1116 detail=f"Deployer not found",
1117 )
1118 device = deployer.get_device(device_id=device_id)
1119 if device is None:
1120 raise HTTPException(
1121 status_code=404,
1122 detail=f"Device not found",
1123 )
1124 return device
1127@app.patch("/device-deployers/{deployer_id}/devices/{device_id}", dependencies=[Depends(get_current_active_user)])
1128async def update_device_from_deployer(
1129 deployer_id: MongoId,
1130 device_id: DeviceId,
1131 device_update: DeviceUpdateFromDeployer,
1132) -> DeviceFromDeployer:
1133 deployer = DeviceDeployer.get_from_id(deployer_id)
1134 device = deployer.update_device(device_id, device_update)
1135 return device
1138@app.delete("/device-deployers/{deployer_id}/devices/{device_id}", dependencies=[Depends(get_current_active_user)])
1139async def delete_device_from_deployer(deployer_id: MongoId, device_id: DeviceId) -> bool:
1140 deployer = DeviceDeployer.get_from_id(deployer_id)
1141 if deployer is None:
1142 raise HTTPException(
1143 status_code=404,
1144 detail=f"Deployer not found",
1145 )
1146 delete_info = deployer.delete_device(device_id=device_id)
1147 if not delete_info.is_deleted:
1148 raise HTTPException(
1149 status_code=500,
1150 detail=f"Cannot delete device: {delete_info.detail}",
1151 )
1152 return delete_info.is_deleted