49 lines
1.1 KiB
Python
49 lines
1.1 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
from dataclasses import dataclass, field
|
|
from datetime import datetime, timezone
|
|
from typing import Any, List, Optional
|
|
|
|
|
|
@dataclass
|
|
class MeasureSnapshot:
|
|
result: List[dict]
|
|
video_left: str
|
|
video_right: str
|
|
updated_at: Optional[datetime] = None
|
|
error: Optional[str] = None
|
|
raw_prediction_path: Optional[str] = None
|
|
|
|
|
|
@dataclass
|
|
class HealthSnapshot:
|
|
behavior_result: str
|
|
health_result: str
|
|
updated_at: Optional[datetime] = None
|
|
error: Optional[str] = None
|
|
raw_class_en: str = ""
|
|
|
|
|
|
@dataclass
|
|
class AppState:
|
|
measure_lock: asyncio.Lock = field(default_factory=asyncio.Lock)
|
|
action_lock: asyncio.Lock = field(default_factory=asyncio.Lock)
|
|
|
|
last_measure: MeasureSnapshot = field(
|
|
default_factory=lambda: MeasureSnapshot(result=[], video_left="", video_right="")
|
|
)
|
|
last_health: HealthSnapshot = field(
|
|
default_factory=lambda: HealthSnapshot(
|
|
behavior_result="",
|
|
health_result="",
|
|
)
|
|
)
|
|
|
|
# job status for optional polling
|
|
measure_status: str = "idle"
|
|
action_status: str = "idle"
|
|
|
|
|
|
app_state = AppState()
|