Coverage for orchestr_ant_ion / streaming / app.py: 0%

1"""Flask app for streaming camera frames over HTTP.""" 

2 

3from __future__ import annotations 

4 

5import os 

6import secrets 

7import warnings 

8 

9from flask import Flask, Response, render_template, stream_with_context 

10from loguru import logger 

11 

12from orchestr_ant_ion.logging_config import setup_logging 

13from orchestr_ant_ion.streaming.capture import FrameCapture 

14from orchestr_ant_ion.streaming.generator import gen_frames 

15 

16 

17def create_app(frame_capture: FrameCapture | None = None) -> Flask: 

18 """Create and configure the streaming Flask app.""" 

19 setup_logging() 

20 

21 app = Flask(__name__) 

22 app.config["SEND_FILE_MAX_AGE_DEFAULT"] = 0 

23 

24 secret_key = os.getenv("KATAGLYPHIS_SECRET_KEY") 

25 if secret_key: 

26 app.config["SECRET_KEY"] = secret_key 

27 else: 

28 generated_key = secrets.token_hex(32) 

29 app.config["SECRET_KEY"] = generated_key 

30 warnings.warn( 

31 "KATAGLYPHIS_SECRET_KEY environment variable not set. " 

32 "Using a randomly generated secret key. Sessions will be invalidated " 

33 "on restart. Set KATAGLYPHIS_SECRET_KEY for production deployments.", 

34 UserWarning, 

35 stacklevel=2, 

36 ) 

37 logger.warning( 

38 "KATAGLYPHIS_SECRET_KEY not set - using ephemeral secret key. " 

39 "Sessions will not persist across restarts." 

40 ) 

41 

42 capture = frame_capture or FrameCapture() 

43 

44 @app.route("/video_feed") 

45 def video_feed() -> Response: 

46 """Return multipart MJPEG stream of camera frames.""" 

47 response = Response( 

48 stream_with_context(gen_frames(capture)), 

49 mimetype="multipart/x-mixed-replace; boundary=frame", 

50 ) 

51 # Disable caching so the browser always loads the newest frame 

52 response.headers["Cache-Control"] = "no-cache, no-store, must-revalidate" 

53 response.headers["Pragma"] = "no-cache" 

54 response.headers["Expires"] = "0" 

55 return response 

56 

57 @app.route("/") 

58 def index() -> str: 

59 """Render the streaming index page.""" 

60 return render_template("index.html") 

61 

62 app.extensions["frame_capture"] = capture 

63 return app 

64 

65 

66def _get_app() -> Flask: 

67 """Lazily create the module-level app on first access.""" 

68 if not hasattr(_get_app, "_instance"): 

69 _get_app._instance = create_app() # type: ignore[attr-defined] # noqa: SLF001 

70 return _get_app._instance # type: ignore[attr-defined] # noqa: SLF001 

71 

72 

73def run() -> None: 

74 """Run the streaming Flask app.""" 

75 app = _get_app() 

76 capture = app.extensions.get("frame_capture") 

77 try: 

78 host = os.getenv("KATAGLYPHIS_STREAM_HOST", "127.0.0.1") 

79 app.run( 

80 host=host, 

81 port=5000, 

82 debug=False, 

83 threaded=True, 

84 use_reloader=False, 

85 ) 

86 except KeyboardInterrupt: 

87 logger.info("Shutting down video stream") 

88 finally: 

89 if capture is not None: 

90 capture.stop() 

91 

92 

93if __name__ == "__main__": 

94 run()