Coverage for orchestrant/pipeline/capture/__init__.py: 92%

1"""Camera capture backends for monitoring pipelines.""" 

2 

3from __future__ import annotations 

4 

5import importlib 

6from typing import TYPE_CHECKING 

7 

8 

9if TYPE_CHECKING: 

10 from orchestrant.pipeline.capture.core import CameraCapture, CaptureProtocol 

11 from orchestrant.pipeline.capture.gstreamer import ( 

12 GStreamerSubprocessCapture, 

13 find_gstreamer_launch, 

14 get_gstreamer_env, 

15 ) 

16 from orchestrant.pipeline.capture.opencv import OpenCVCapture 

17 

18# Lazy (PEP 562): the GStreamer backend needs no cv2, which has no free-threaded (3.14t) wheel. 

19_EXPORTS = { 

20 "CameraCapture": "orchestrant.pipeline.capture.core", 

21 "CaptureProtocol": "orchestrant.pipeline.capture.core", 

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

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

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

25 "OpenCVCapture": "orchestrant.pipeline.capture.opencv", 

26} 

27 

28 

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

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

31 if name in _EXPORTS: 

32 return getattr(importlib.import_module(_EXPORTS[name]), name) 

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

34 raise AttributeError(msg) 

35 

36 

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

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

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

40 

41 

42__all__ = [ 

43 "CameraCapture", 

44 "CaptureProtocol", 

45 "GStreamerSubprocessCapture", 

46 "OpenCVCapture", 

47 "find_gstreamer_launch", 

48 "get_gstreamer_env", 

49]