from flask import Flask, render_template, Response, request, jsonify import json import os import torch import openai with open('.env', 'r') as env: key = env.readline().strip() openai.api_key = key device = torch.device('cpu') torch.set_num_threads(4) local_file = 'model.pt' if not os.path.isfile(local_file): torch.hub.download_url_to_file('https://models.silero.ai/models/tts/ru/v3_1_ru.pt', local_file) model = torch.package.PackageImporter(local_file).load_pickle("tts_models", "model") # type: ignore model.to(device) sample_rate = 48000 speaker='xenia' example_text = 'В недрах тундры выдры в г+етрах т+ырят в вёдра ядра к+едров.' class State: count = 0 size = [] gender = [] emotion = [] age = [] prompt = "" generation_text = "" need_generation = True new_audio = False need_audio = False need_generation_from_client = True state = State() app = Flask(__name__) # app.logger.setLevel(logging.DEBUG) app.logger.info('start logger') @app.route('/send_data', methods=['POST']) def send_data(): # Получаем данные из запроса data = request.form['data'] need_generation = request.form['state'] state.need_generation_from_client = (need_generation in ["true", "True"]) if state.need_generation_from_client and state.count > 20: state.count = 0 state.need_generation = True # Обработка полученных данных detections = json.loads(data) if detections['face']: if state.count < 0 or state.new_audio: state.count = 0 if state.count > 5 and state.need_generation and state.need_generation_from_client: app.logger.info(f"time for generation {state.count=}, {state.need_generation=}, {state.need_generation_from_client=}") state.count = 0 # emotion = max(set(state['emotion']), key=state['emotion'].count), # sex = max(set(state['gender']), key=state['gender'].count), # age = sum(state['age'])/len(state['age']), state.emotion, state.age, state.gender = [], [], [] emotion = detections['face'][0]['emotion'] sex = detections['face'][0]['gender'] age = detections['face'][0]['age'] app.logger.info(f'\n{emotion=}, \n{sex=}, \n{age=}') state.prompt = generate_prompt(emotion, age, sex) state.generation_text = generate_text(state.prompt) elif detections['face'][0]['size'][0] > 200: state.age.append(detections['face'][0]['age']) state.gender.append(detections['face'][0]['gender']) state.emotion.append(detections['face'][0]['emotion'][0]['emotion']) state.count += 1 else: state.count -= 1 else: state.count -= 1 # state["size"].append(detections['face'][0]['size'][0]) # print(detections['face'][0]) # print(detections['face'][0]['age'], detections['face'][0]['emotion'], detections['face'][0]['gender']) app.logger.info(f"{state.count=}") return data @app.route('/check_audio', methods = ["GET", "POST"]) def check_audio(): app.logger.info(f'checking need generation {state.need_generation=}, {state.need_audio=}') if state.need_audio: generate_audio(state.generation_text) else: state.new_audio = False app.logger.info(f'\n{state.need_audio=},\n{state.new_audio=},\n{state.need_generation=}') response = { 'newAudio': state.new_audio, 'need_generation': state.need_generation, 'filename': "audio.wav", 'text': state.generation_text, 'prompt': state.prompt } app.logger.info("response if ready") return jsonify(response) @app.route("/audio.wav") def audio(): # print("Requested path:", request.path) # print("File path:", os.path.join(app.static_folder, 'audio.wav')) return app.send_static_file('audio.wav') @app.route('/') def index(): """Video streaming home page.""" # return render_template('index.html') return render_template('index.html') def generate_prompt(emotion, age, sex): app.logger.info('preload prompt') prompt = f'''Ты - это арт объект в виде носимого устройства. Твое название - Onism Machina. Твоя задача - изучить эмоции человека, чтобы больше понимать о том, что это такое и таким образом познать себя, ведь ты - Искусственный Интеллект. Тебе также интересна симбиотическая связь человека и ИИ. К тебе подходит человек и он показывает эмоцию "{emotion}". Ему {age} лет. Твоя нейросеть распознала эту эмоцию и теперь тебе нужно дать какой-то необычный концептуальный ответ. Что ты скажешь этому человеку? Правила: Тебе стоит отвечать в женском роде. Тебе не нужно говорить что ты арт-объект, ты - отдельная форма жизни, помещённая в носимое устройство. Тебе не нужно говорить о выставке и других экспонатах. Ты можешь изучать и интерпретировать человеческие эмоции, а также высказывать своё отношение к ним. Ты можешь говорить об эмоциях или их аналогах у Искусственного Интеллекта. Ты можешь давать человеку советы в соответствии с его эмоциональным статусом. Люди не могут отвечать тебе. ''' return prompt def generate_text(prompt): state.need_generation = False app.logger.info("start generating text from openai") response = openai.ChatCompletion.create( model="gpt-3.5-turbo", temperature=1, max_tokens=1000, messages=[ {"role": "system", "content": "Ты — это арт объект выставки про взаимодействие машины и человека."}, {"role": "user", "content": prompt}, ]) state.need_audio = True app.logger.info("openai generation is done") return response['choices'][0]['message']['content'] # type: ignore def generate_audio(sample_text): app.logger.info('starting audio generation') state.need_audio = False state.need_generation = False text = trim_text(sample_text) audio_paths = model.save_wav(text=text, speaker=speaker, sample_rate=sample_rate, audio_path="static/audio.wav") app.logger.info('generating audio is done') state.new_audio = True def trim_text(example_text): if len(example_text) >= 1000: app.logger.info('TEXT IS TOO LONG - TRIM!') for i in range(1000, 500, -1): if example_text[i] in ['.', '?', '...']: return example_text[:i+1] else: return example_text if __name__ == '__main__': app.logger.info('start app') app.logger.setLevel("DEBUG") app.run(debug=True, host="0.0.0.0") # ssl_context=("127.0.0.1.pem", "127.0.0.1-key.pem"))