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

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

2 

3from __future__ import annotations 

4 

5import os 

6import secrets 

7import warnings 

8from functools import lru_cache 

9 

10from flask import Flask, Response, render_template, stream_with_context 

11from loguru import logger 

12 

13from orchestrant.logging_config import setup_logging 

14from orchestrant.streaming.capture import FrameCapture 

15from orchestrant.streaming.generator import gen_frames 

16 

17 

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

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

20 setup_logging() 

21 

22 app = Flask(__name__, template_folder="template-files") 

23 app.config["SEND_FILE_MAX_AGE_DEFAULT"] = 0 

24 

25 secret_key = os.getenv("KATAGLYPHIS_SECRET_KEY") 

26 if secret_key: 

27 app.config["SECRET_KEY"] = secret_key 

28 else: 

29 generated_key = secrets.token_hex(32) 

30 app.config["SECRET_KEY"] = generated_key 

31 warnings.warn( 

32 "KATAGLYPHIS_SECRET_KEY environment variable not set. " 

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

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

35 UserWarning, 

36 stacklevel=2, 

37 ) 

38 logger.warning( 

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

40 "Sessions will not persist across restarts." 

41 ) 

42 

43 capture = frame_capture or FrameCapture() 

44 

45 @app.route("/video_feed") 

46 def video_feed() -> Response: 

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

48 response = Response( 

49 stream_with_context(gen_frames(capture)), 

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

51 ) 

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

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

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

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

56 return response 

57 

58 @app.route("/") 

59 def index() -> str: 

60 """Render the streaming index page.""" 

61 return render_template("index.html") 

62 

63 app.extensions["frame_capture"] = capture 

64 return app 

65 

66 

67@lru_cache(maxsize=1) 

68def _get_app() -> Flask: 

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

70 # Cached, not a function attribute: tests reset it with _get_app.cache_clear(). 

71 return create_app() 

72 

73 

74def run() -> None: 

75 """Run the streaming Flask app.""" 

76 app = _get_app() 

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

78 try: 

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

80 app.run( 

81 host=host, 

82 port=5000, 

83 debug=False, 

84 threaded=True, 

85 use_reloader=False, 

86 ) 

87 except KeyboardInterrupt: 

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

89 finally: 

90 if capture is not None: 

91 capture.stop() 

92 

93 

94if __name__ == "__main__": 

95 run()