Coverage for orchestrant/pipeline/types.py: 100%

1"""Shared data structures for monitoring pipelines.""" 

2 

3from __future__ import annotations 

4 

5from dataclasses import dataclass 

6from enum import Enum 

7from typing import TYPE_CHECKING 

8 

9 

10if TYPE_CHECKING: 

11 from collections import deque 

12 

13 

14RESOLUTION_MIN = 64 

15RESOLUTION_MAX = 7680 

16FPS_MIN = 1 

17FPS_MAX = 240 

18DEVICE_INDEX_MIN = 0 

19 

20 

21class CaptureBackend(Enum): 

22 """Video capture backend options.""" 

23 

24 OPENCV = "opencv" 

25 GSTREAMER = "gstreamer" 

26 

27 

28@dataclass 

29class CameraConfig: 

30 """Camera configuration settings with validation.""" 

31 

32 device_index: int = 0 

33 width: int = 1920 

34 height: int = 1080 

35 fps: int = 30 

36 backend: CaptureBackend = CaptureBackend.OPENCV 

37 

38 def __post_init__(self) -> None: 

39 """Validate field ranges as soon as the config is constructed.""" 

40 self._validate() 

41 

42 def _validate(self) -> None: 

43 if self.device_index < DEVICE_INDEX_MIN: 

44 msg = f"device_index must be >= {DEVICE_INDEX_MIN}" 

45 raise ValueError(msg) 

46 if not RESOLUTION_MIN <= self.width <= RESOLUTION_MAX: 

47 msg = f"width must be in [{RESOLUTION_MIN}, {RESOLUTION_MAX}]" 

48 raise ValueError(msg) 

49 if not RESOLUTION_MIN <= self.height <= RESOLUTION_MAX: 

50 msg = f"height must be in [{RESOLUTION_MIN}, {RESOLUTION_MAX}]" 

51 raise ValueError(msg) 

52 if not FPS_MIN <= self.fps <= FPS_MAX: 

53 msg = f"fps must be in [{FPS_MIN}, {FPS_MAX}]" 

54 raise ValueError(msg) 

55 

56 

57@dataclass 

58class SystemStats: 

59 """Container for system statistics.""" 

60 

61 cpu_percent: float = 0.0 

62 ram_percent: float = 0.0 

63 ram_used_gb: float = 0.0 

64 ram_total_gb: float = 0.0 

65 gpu_percent: float = 0.0 

66 gpu_memory_used_gb: float = 0.0 

67 gpu_memory_total_gb: float = 0.0 

68 gpu_memory_percent: float = 0.0 

69 gpu_temp_celsius: float = 0.0 

70 gpu_power_watts: float = 0.0 

71 gpu_name: str = "N/A" 

72 

73 

74@dataclass 

75class PerformanceMetrics: 

76 """Container for performance metrics.""" 

77 

78 camera_fps: float = 0.0 

79 inference_ms: float = 0.0 

80 inference_capacity_fps: float = 0.0 

81 frame_budget_percent: float = 0.0 

82 actual_throughput_fps: float = 0.0 

83 

84 

85@dataclass 

86class Track: 

87 """Tracked object with normalized trajectory points. 

88 

89 Attributes: 

90 track_id: Unique identifier for this track. 

91 points_norm: Deque of (x, y) normalized coordinates (0.0-1.0) showing 

92 the object's recent trajectory. New points are appended to the right. 

93 last_seen_ts: Timestamp of the most recent detection associated with 

94 this track. Used to expire stale tracks. 

95 """ 

96 

97 track_id: int 

98 points_norm: deque[tuple[float, float]] 

99 last_seen_ts: float