Coverage for orchestrant/pipeline/__init__.py: 41%

1"""Monitoring pipeline package: capture, tracking, metrics, and viewers.""" 

2 

3from __future__ import annotations 

4 

5import importlib 

6from typing import TYPE_CHECKING 

7 

8 

9if TYPE_CHECKING: 

10 from orchestrant.pipeline.ui.wx import ( 

11 WxPythonViewer as WxPythonViewerType, 

12 ) 

13 

14# Lazy (PEP 562): a light submodule import must not pull in cv2, DearPyGui or GStreamer. 

15_EXPORTS = { 

16 "CameraCapture": "orchestrant.pipeline.capture", 

17 "OpenCVCapture": "orchestrant.pipeline.capture", 

18 "GStreamerSubprocessCapture": "orchestrant.pipeline.capture.gstreamer", 

19 "find_gstreamer_launch": "orchestrant.pipeline.capture.gstreamer", 

20 "get_gstreamer_env": "orchestrant.pipeline.capture.gstreamer", 

21 "attach_log_buffer": "orchestrant.pipeline.logging", 

22 "configure_logging": "orchestrant.pipeline.logging", 

23 "create_log_buffer": "orchestrant.pipeline.logging", 

24 "log_buffer_lines": "orchestrant.pipeline.logging", 

25 "PerformanceTracker": "orchestrant.pipeline.metrics.performance", 

26 "PowerMonitor": "orchestrant.pipeline.monitoring.power", 

27 "get_cpu_freq_ratio": "orchestrant.pipeline.monitoring.power", 

28 "SystemMonitor": "orchestrant.pipeline.monitoring.system", 

29 "SimpleCentroidTracker": "orchestrant.pipeline.tracking.centroid", 

30 "CameraConfig": "orchestrant.pipeline.types", 

31 "CaptureBackend": "orchestrant.pipeline.types", 

32 "PerformanceMetrics": "orchestrant.pipeline.types", 

33 "SystemStats": "orchestrant.pipeline.types", 

34 "Track": "orchestrant.pipeline.types", 

35 "DearPyGuiViewer": "orchestrant.pipeline.ui.dearpygui", 

36 "AMD_AVAILABLE": "orchestrant.monitoring.gpu", 

37 "PYNVML_AVAILABLE": "orchestrant.monitoring.gpu", 

38} 

39 

40 

41def __getattr__(name: str) -> object: 

42 """Resolve a public name from its home module on first access.""" 

43 if name == "WxPythonViewer": 

44 # None, not ImportError, without wxPython: yolo/monitor.py checks for None. 

45 try: 

46 _wx_mod = importlib.import_module("orchestrant.pipeline.ui.wx") 

47 except Exception: # pragma: no cover - optional dependency 

48 return None 

49 return getattr(_wx_mod, "WxPythonViewer", None) 

50 if name in _EXPORTS: 

51 module = importlib.import_module(_EXPORTS[name]) 

52 return getattr(module, name) 

53 msg = f"module {__name__!r} has no attribute {name!r}" 

54 raise AttributeError(msg) 

55 

56 

57def __dir__() -> list[str]: 

58 """Advertise lazy exports to introspection alongside real attributes.""" 

59 return sorted(set(globals()) | set(__all__)) 

60 

61 

62__all__ = [ 

63 "AMD_AVAILABLE", 

64 "PYNVML_AVAILABLE", 

65 "CameraCapture", 

66 "CameraConfig", 

67 "CaptureBackend", 

68 "DearPyGuiViewer", 

69 "GStreamerSubprocessCapture", 

70 "OpenCVCapture", 

71 "PerformanceMetrics", 

72 "PerformanceTracker", 

73 "PowerMonitor", 

74 "SimpleCentroidTracker", 

75 "SystemMonitor", 

76 "SystemStats", 

77 "Track", 

78 "WxPythonViewer", 

79 "attach_log_buffer", 

80 "configure_logging", 

81 "create_log_buffer", 

82 "find_gstreamer_launch", 

83 "get_cpu_freq_ratio", 

84 "get_gstreamer_env", 

85 "log_buffer_lines", 

86]