Coverage for apps/streamanimate/streamanimate.py: 94%

1""" 

2StreamAnimate: Generate an animated video. 

3Starts an HTTP server to accept job submissions and monitor job status. 

4""" 

5 

6import sys 

7 

8from typing import override 

9from typing import Dict 

10from typing import Any 

11 

12from streamanimate_job import StreamAnimateJob 

13 

14# Local relative imports 

15sys.path.append("..") # noqa: E402 

16sys.path.append("../..") # noqa: E402 

17 

18from streamwise_job import StreamWiseJob 

19from streamwise_app import StreamWiseApp 

20from streamwise_app import run_app 

21 

22 

23class StreamAnimateApp(StreamWiseApp): 

24 """Quart app for StreamAnimate video generation.""" 

25 

26 def __init__(self) -> None: 

27 super().__init__("streamanimate") 

28 

29 @override 

30 def create_job( 

31 self, 

32 job_id: str, 

33 job_config: Dict[str, Any] 

34 ) -> StreamWiseJob: 

35 return StreamAnimateJob( 

36 job_id=job_id, 

37 config=job_config, 

38 service_manager=self.service_manager 

39 ) 

40 

41 

42if __name__ == "__main__": 

43 run_app( 

44 StreamAnimateApp, 

45 tmp_dir="/tmp/streamanimate", 

46 log_files=[ 

47 "streamwise.log", 

48 "streamanimate.log" 

49 ], 

50 app_name="StreamAnimate", 

51 )