Coverage for orchestrant/streaming/app.py: 90%

1"""The camera as a live Flet web page: ``python -m orchestrant.streaming``.""" 

2 

3from __future__ import annotations 

4 

5import asyncio 

6import os 

7import threading 

8from typing import TYPE_CHECKING, Any 

9 

10import cv2 

11from loguru import logger 

12 

13from orchestrant.logging_config import setup_logging 

14from orchestrant.pipeline.ui.flet_viewer import require_flet, web_server 

15from orchestrant.streaming.capture import FrameCapture 

16 

17 

18if TYPE_CHECKING: 

19 from collections.abc import Callable 

20 

21 import numpy as np 

22 

23 

24HOST_ENV = "KATAGLYPHIS_STREAM_HOST" 

25DEFAULT_HOST = "127.0.0.1" 

26PORT = 5000 

27JPEG_QUALITY = 30 

28FRAME_TIMEOUT_S = 2.0 

29WAIT_ON_EMPTY_S = 0.5 

30 

31 

32def jpeg(frame: np.ndarray, quality: int = JPEG_QUALITY) -> bytes | None: 

33 """One frame as JPEG bytes; None when encoding fails.""" 

34 ok, buffer = cv2.imencode(".jpg", frame, [cv2.IMWRITE_JPEG_QUALITY, quality]) 

35 return buffer.tobytes() if ok else None 

36 

37 

38def pump( 

39 loop: asyncio.AbstractEventLoop, 

40 image: Any, # noqa: ANN401 - a flet.RawImage 

41 capture: FrameCapture, 

42 stop: threading.Event, 

43) -> None: 

44 """Draw the capture's frames into `image` until the tab closes.""" 

45 while not stop.is_set(): 

46 frame = capture.get_frame() 

47 if frame is None: 

48 stop.wait(WAIT_ON_EMPTY_S) 

49 continue 

50 data = jpeg(frame) 

51 if data is None: 

52 logger.warning("Frame encoding failed; skipping frame") 

53 continue 

54 future = asyncio.run_coroutine_threadsafe(image.render_encoded(data), loop) 

55 try: 

56 future.result(timeout=FRAME_TIMEOUT_S) 

57 except TimeoutError: 

58 future.cancel() 

59 except Exception as exc: # the tab is gone; its pump ends, the server stays 

60 logger.debug("Stream session ended: {}", exc) 

61 stop.set() 

62 

63 

64def stream_page(capture: FrameCapture) -> Callable[[Any], None]: 

65 """The page each browser tab gets: a title and the live frame, fed by its own pump.""" 

66 flet = require_flet() 

67 

68 def main(page: Any) -> None: # noqa: ANN401 - a flet.Page 

69 page.title = "Live Streaming" 

70 page.theme = flet.Theme(color_scheme_seed=flet.Colors.GREEN_ACCENT) 

71 image = flet.RawImage(expand=True, fit=flet.BoxFit.CONTAIN) 

72 stop = threading.Event() 

73 page.on_close = page.on_disconnect = lambda _event: stop.set() 

74 page.add( 

75 flet.Text("Live Streaming", size=24, weight=flet.FontWeight.BOLD), 

76 flet.Container(content=image, expand=True), 

77 ) 

78 page.run_thread(pump, page.loop, image, capture, stop) 

79 

80 return main 

81 

82 

83def run() -> None: 

84 """Serve the Picamera2 stream on port 5000 until Ctrl+C; $KATAGLYPHIS_STREAM_HOST binds it.""" 

85 setup_logging() 

86 host = os.getenv(HOST_ENV, DEFAULT_HOST) 

87 capture = FrameCapture() 

88 try: 

89 logger.info("Streaming at http://{}:{}", host, PORT) 

90 web_server(stream_page(capture), host, PORT).run() 

91 finally: 

92 capture.stop() 

93 

94 

95if __name__ == "__main__": 

96 run()