Coverage for integrations / service_tools / servers / tts_audio_suite_server.py: 0.0%

1""" 

2TTS-Audio-Suite Sidecar Server — Flask API on a dynamic port. 

3 

4Launched as a subprocess by RuntimeToolManager. On startup: 

51. Finds a free port (OS-assigned) 

62. Prints PORT=NNNNN to stdout (parent reads this) 

73. Lazy-loads TTS models 

84. Serves TTS requests 

9 

10Usage (standalone test): 

11 python -m integrations.service_tools.servers.tts_audio_suite_server 

12 

13Pattern from: ltx2_server.py 

14""" 

15 

16import json 

17import logging 

18import os 

19import socket 

20import sys 

21import uuid 

22from pathlib import Path 

23from threading import Lock 

24 

25from flask import Flask, request, jsonify, send_file 

26 

27try: 

28 from integrations.service_tools.vram_manager import clear_cuda_cache 

29except ImportError: 

30 def clear_cuda_cache(): 

31 try: 

32 import torch 

33 if torch.cuda.is_available(): 

34 torch.cuda.empty_cache() 

35 if hasattr(torch.backends, 'mps') and torch.backends.mps.is_available(): 

36 torch.mps.empty_cache() 

37 except Exception: 

38 pass 

39 

40logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s') 

41logger = logging.getLogger('tts_audio_suite_server') 

42 

43app = Flask(__name__) 

44 

45# Global state 

46_model = None 

47_model_lock = Lock() 

48_model_dir = None 

49OUTPUT_DIR = os.path.join(Path.home(), '.hevolve', 'outputs', 'tts_audio_suite') 

50os.makedirs(OUTPUT_DIR, exist_ok=True) 

51 

52 

53def _get_model_dir(): 

54 global _model_dir 

55 if _model_dir: 

56 return _model_dir 

57 _model_dir = os.environ.get( 

58 'TTS_AUDIO_SUITE_MODEL_DIR', 

59 str(Path.home() / '.hevolve' / 'models' / 'tts-audio-suite') 

60 ) 

61 return _model_dir 

62 

63 

64def _load_model(): 

65 """Lazy-load TTS-Audio-Suite model.""" 

66 global _model 

67 if _model is not None: 

68 return _model 

69 

70 with _model_lock: 

71 if _model is not None: 

72 return _model 

73 

74 model_dir = _get_model_dir() 

75 logger.info(f"Loading TTS-Audio-Suite from {model_dir}...") 

76 

77 if model_dir not in sys.path: 

78 sys.path.insert(0, model_dir) 

79 

80 try: 

81 offload_mode = os.environ.get('TTS_OFFLOAD', 'gpu') 

82 

83 # TTS-Audio-Suite supports multiple TTS backends 

84 # Actual loading depends on repo structure 

85 _model = { 

86 'loaded': True, 

87 'model_dir': model_dir, 

88 'offload_mode': offload_mode, 

89 } 

90 logger.info(f"TTS-Audio-Suite loaded (mode: {offload_mode})") 

91 return _model 

92 except Exception as e: 

93 logger.error(f"Failed to load TTS-Audio-Suite: {e}") 

94 _model = {'loaded': False, 'error': str(e)} 

95 return _model 

96 

97 

98@app.route('/health', methods=['GET']) 

99def health(): 

100 """Health check with VRAM stats.""" 

101 status = {'status': 'ok', 'service': 'tts_audio_suite'} 

102 try: 

103 import torch 

104 if torch.cuda.is_available(): 

105 status['gpu'] = torch.cuda.get_device_name(0) 

106 status['vram_total_gb'] = round(torch.cuda.get_device_properties(0).total_memory / 1e9, 2) 

107 status['vram_used_gb'] = round(torch.cuda.memory_allocated(0) / 1e9, 2) 

108 except ImportError: 

109 pass 

110 return jsonify(status) 

111 

112 

113@app.route('/synthesize', methods=['POST']) 

114def synthesize(): 

115 """Synthesize speech from text.""" 

116 data = request.get_json() or {} 

117 text = data.get('text', '') 

118 model_name = data.get('model', 'default') 

119 language = data.get('language', 'en') 

120 

121 if not text: 

122 return jsonify({'error': 'text is required'}), 400 

123 

124 model = _load_model() 

125 if not model.get('loaded'): 

126 return jsonify({'error': f"Model not loaded: {model.get('error', 'unknown')}"}), 503 

127 

128 try: 

129 output_id = str(uuid.uuid4())[:8] 

130 output_path = os.path.join(OUTPUT_DIR, f"tts_{output_id}.wav") 

131 

132 # TODO: Replace with actual TTS-Audio-Suite inference call once repo is cloned 

133 return jsonify({ 

134 'success': True, 

135 'audio_url': f"/audio/{output_id}", 

136 'text': text, 

137 'model': model_name, 

138 'language': language, 

139 'message': 'TTS-Audio-Suite synthesis placeholder — model integration pending repo clone', 

140 }) 

141 except Exception as e: 

142 logger.error(f"Synthesis failed: {e}") 

143 return jsonify({'error': str(e)}), 500 

144 

145 

146@app.route('/models', methods=['GET']) 

147def list_models(): 

148 """List available TTS models.""" 

149 return jsonify({ 

150 'models': [ 

151 {'name': 'default', 'description': 'Default TTS model', 'languages': ['en']}, 

152 ], 

153 'message': 'Model list placeholder — populated after model download', 

154 }) 

155 

156 

157@app.route('/audio/<filename>', methods=['GET']) 

158def serve_audio(filename): 

159 """Serve generated audio file.""" 

160 path = os.path.join(OUTPUT_DIR, f"tts_{filename}.wav") 

161 if os.path.exists(path): 

162 return send_file(path, mimetype='audio/wav') 

163 return jsonify({'error': 'not found'}), 404 

164 

165 

166@app.route('/unload', methods=['POST']) 

167def unload(): 

168 """Unload model to free memory.""" 

169 global _model 

170 _model = None 

171 clear_cuda_cache() 

172 return jsonify({'status': 'unloaded'}) 

173 

174 

175def _find_free_port() -> int: 

176 """Find a free port using OS assignment.""" 

177 sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) 

178 sock.bind(('127.0.0.1', 0)) 

179 port = sock.getsockname()[1] 

180 sock.close() 

181 return port 

182 

183 

184if __name__ == '__main__': 

185 port = _find_free_port() 

186 print(f"PORT={port}", flush=True) 

187 app.run(host='127.0.0.1', port=port, threaded=True)