Coverage for apps/streamcast/streamcast.py: 72%

1""" 

2StreamCast: Generate podcasts from documents using AI services. 

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

4""" 

5 

6import sys 

7import logging 

8 

9from typing import override 

10from typing import Dict 

11from typing import Any 

12 

13from streamcast_job import StreamCastJob 

14 

15# Local relative imports 

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

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

18 

19from streamwise_job import StreamWiseJob 

20from streamwise_app import StreamWiseApp 

21from streamwise_app import run_app 

22 

23from tts_utils import estimate_num_words_from_audio_duration 

24 

25 

26class StreamCastApp(StreamWiseApp): 

27 """Quart app for StreamCast podcast generation.""" 

28 

29 def __init__(self) -> None: 

30 super().__init__("streamcast") 

31 

32 def get_job_config_from_request( 

33 self, 

34 job_id: str, 

35 request_json: Dict[str, Any] 

36 ) -> Dict[str, Any]: 

37 """Process and return the job configuration from the request JSON.""" 

38 ret = super().get_job_config_from_request(job_id, request_json) 

39 

40 # Estimate dialogue length for a user requested duration 

41 if "video_duration_seconds" in ret: 

42 video_duration_seconds = float(ret["video_duration_seconds"]) 

43 del ret["video_duration_seconds"] 

44 DIALOGUE_DURATION_SECONDS = 5.0 # Each dialogue should be ~5 seconds 

45 num_words = estimate_num_words_from_audio_duration(DIALOGUE_DURATION_SECONDS) 

46 ret["max_words_per_dialogue"] = num_words 

47 num_dialogues = max(2, int(video_duration_seconds // DIALOGUE_DURATION_SECONDS)) 

48 ret["max_dialogues"] = num_dialogues 

49 logging.info( 

50 f"Estimated {num_dialogues} dialogues with {num_words} words each for " 

51 f"{video_duration_seconds} seconds video.") 

52 

53 return ret 

54 

55 @override 

56 def create_job( 

57 self, 

58 job_id: str, 

59 job_config: Dict[str, Any] 

60 ) -> StreamWiseJob: 

61 return StreamCastJob( 

62 job_id=job_id, 

63 config=job_config, 

64 service_manager=self.service_manager 

65 ) 

66 

67 

68if __name__ == "__main__": 

69 run_app( 

70 StreamCastApp, 

71 tmp_dir="/tmp/streamcast", 

72 log_files=[ 

73 "streamwise.log", 

74 "streamcast.log" 

75 ], 

76 app_name="StreamCast", 

77 )