Coverage for orchestr_ant_ion / pipeline / capture / core.py: 43%

1"""Capture orchestration helpers.""" 

2 

3from __future__ import annotations 

4 

5from typing import TYPE_CHECKING, Protocol 

6 

7from loguru import logger 

8 

9from orchestr_ant_ion.pipeline.capture.gstreamer import ( 

10 GStreamerSubprocessCapture, 

11) 

12from orchestr_ant_ion.pipeline.capture.opencv import OpenCVCapture 

13from orchestr_ant_ion.pipeline.types import CaptureBackend 

14 

15 

16if TYPE_CHECKING: 

17 import numpy as np 

18 

19 from orchestr_ant_ion.pipeline.types import CameraConfig 

20 

21 

22class CaptureProtocol(Protocol): 

23 """Protocol for capture backends.""" 

24 

25 def open(self) -> bool: 

26 """Open the capture backend.""" 

27 ... 

28 

29 def read(self) -> tuple[bool, np.ndarray | None]: 

30 """Read a frame from the backend.""" 

31 ... 

32 

33 def release(self) -> None: 

34 """Release backend resources.""" 

35 ... 

36 

37 def is_opened(self) -> bool: 

38 """Return True when the backend is open.""" 

39 ... 

40 

41 def get_info(self) -> dict: 

42 """Return backend metadata for diagnostics.""" 

43 ... 

44 

45 

46class CameraCapture: 

47 """Unified camera capture supporting OpenCV and GStreamer subprocess backends.""" 

48 

49 def __init__(self, config: CameraConfig) -> None: 

50 """Create a capture wrapper with the requested backend.""" 

51 self.config = config 

52 self._capture: CaptureProtocol | None = None 

53 self.backend_name = "" 

54 

55 def open(self) -> bool: 

56 """Open the selected backend, falling back to OpenCV if needed.""" 

57 if self.config.backend == CaptureBackend.GSTREAMER: 

58 logger.info("Using GStreamer subprocess capture...") 

59 self._capture = GStreamerSubprocessCapture(self.config) 

60 if self._capture.open(): 

61 self.backend_name = "GStreamer (subprocess)" 

62 return True 

63 logger.warning("GStreamer failed, falling back to OpenCV...") 

64 

65 self._capture = OpenCVCapture(self.config) 

66 if self._capture.open(): 

67 self.backend_name = "OpenCV" 

68 return True 

69 

70 return False 

71 

72 def read(self) -> tuple[bool, np.ndarray | None]: 

73 """Read a frame from the active backend.""" 

74 if self._capture is None: 

75 return False, None 

76 return self._capture.read() 

77 

78 def release(self) -> None: 

79 """Release the active backend.""" 

80 if self._capture is not None: 

81 self._capture.release() 

82 self._capture = None 

83 

84 def is_opened(self) -> bool: 

85 """Return True if the active backend is open.""" 

86 return self._capture is not None and self._capture.is_opened() 

87 

88 def get_info(self) -> dict: 

89 """Return backend metadata for diagnostics.""" 

90 if self._capture is not None: 

91 return self._capture.get_info() 

92 return {"backend": "None", "pipeline": "", "width": 0, "height": 0, "fps": 0}