240907 ~ 240914 연구노트

from flask import Flask from Controller_240913V import bp app = Flask(__name__) # Controller.py의 bp 객체를 app 객체에 적용 app.register_blueprint(bp) @app.route('/') def index(): return 'Hello, World!' if __name__ == '__main__': app.run(debug = True)

from flask import Blueprint,jsonify from DialogueSessionManager_240913V import manage_dialogue_session bp = Blueprint('Controller_240913V', __name__) @bp.route("/user/<int:user_id>/event/<int:event_id>", methods = ['GET']) def handle_request(user_id, event_id): result = manage_dialogue_session(user_id,event_id) if result: return jsonify(result) else: return jsonify({'error': 'Failed to process request'}), 500

import mysql.connector from record_play import record_audio, play_audio from CSR import audio_to_text from Chatbot_240913V import execute_chatbot from TTS_240914V import text_to_speech def connect_to_db(): try: connection = mysql.connector.connect( host='localhost', user='hanshin', password='hanshin2024', database='user_data' ) return connection except mysql.connector.Error as err: print(f"Error : {err}") return None def get_user_data(user_id): connection = connect_to_db() if connection is None: return {'error' : 'DB connection 실패'} try: cur = connection.cursor() # query 설정 query = """ SELECT name, gender, date_of_birth, contact_number, address, photo, disability_type, disability_grade, special_notes FROM users WHERE user_id = %s """ # query 실행 cur.execute(query, (user_id,)) # user_data 내용 변수에 저장 user_data = cur.fetchone() if user_data: return { 'name': user_data[0], 'gender': user_data[1], 'date_of_birth': user_data[2], 'contact_number': user_data[3], 'address': user_data[4], 'photo': user_data[5], # BLOB 타입이므로 적절한 처리가 필요할 수 있음 'disability_type': user_data[6], 'disability_grade': user_data[7], 'special_notes': user_data[8] } else: print("Not Found User Data") return None except mysql.connector.Error as err: return {'error' : f"User Data 조회 중 에러 : {err}"} finally: if connection.is_connected(): connection.close() def get_situation_data(event_id): connection = connect_to_db() if connection is None: return {'error' : 'DB connection 실패'} try : cur = connection.cursor() query = """ SELECT * FROM situations WHERE event_id = %s """ cur.execute(query, (event_id,)) situation_data = cur.fetchone() if situation_data: return { } else: return {'error' : 'Event not found'} except mysql.connector.Error as err: return {'error' : f"Situation Data 조회 중 에러 : {err}"} finally: if connection.is_connected(): connection.close() def manage_dialogue_session(user_id, event_id): user_data = get_user_data(user_id) situation_data = get_situation_data(event_id) if not user_data: return {'error' : 'User not found'} if not situation_data: return {'error' : 'Event not found'} """ is_emergency = if is_emergency: response = generate_emergency_chat(user_data, situation_data) else: response = generate_usually_chat(user_data, situation_data) """ def generate_emergency_chat(user_data, situation_data): pass def generate_usually_chat(user_data, situation_data): try: record_audio() play_audio() # Clova Speech API로 오디오 파일을 텍스트로 변환 text_result = audio_to_text() if text_result: chatbot_result = execute_chatbot(user_data=user_data, situation_data=situation_data, text=text_result) else: print("오디오 텍스트 변환 실패") return None if chatbot_result: text_to_speech(text = chatbot_result) else: print("chatbot 실행 실패") return None except Exception as e: print(f"Error : {e}") return None if __name__ == '__main__': user_data = { 'name': '홍길동', 'gender': '남', 'date_of_birth': '2000-01-01', 'contact_number': '010-1234-5678', 'address': '서울시 중구', 'photo': None, # BLOB 타입이므로 적절한 처리가 필요할 수 있음 'disability_type': '시각장애', 'disability_grade': '2등급', 'special_notes': '시각장애 2등급은 시력이 0.04이하인 사람을 의미합니다. 시력이 안 좋은 상태이므로 안내를 할 때 시력에 특히 도움이 되는 내용들을 많이 포함해주세요.' } situtaion_data = { 'situation' : '현재 냄비에서 화재 초기 단계 증상인 상태입니다.' } generate_usually_chat(user_data=user_data,situation_data=situtaion_data)

# 필요한 라이브러리 import import speech_recognition as sr from playsound import playsound import os # Recognizer 객체 생성 r = sr.Recognizer() # 마이크를 오디오 소스로 사용 mic = sr.Microphone() # 저장할 오디오 파일의 이름 folder_name = "./data" file_name = os.path.join(folder_name, "recorded_audio.wav") if not os.path.exists(folder_name): os.makedirs(folder_name) # 음성 녹음 함수 def record_audio(): with mic as source: print("녹음 시작...") # audio_data = r.listen(source, phrase_time_limit=5) audio_data = r.listen(source) print("녹음 완료!") with open(file_name, "wb") as f: f.write(audio_data.get_wav_data()) print("녹음 파일이 저장되었습니다.") # 오디오 파일 재생 함수 def play_audio(): playsound(file_name) if __name__ == "__main__": pass

import requests import json from record_play import file_name class ClovaSpeechClient: # Clova Speech invoke URL invoke_url = 'https://clovaspeech-gw.ncloud.com/external/v1/8983/54d5b161d3da62f82f3592398295d44524110cfef1b81e5ca3dae43e79f634d0' secret = 'b256e2de4b3a4898b3f020894236934f' # 외부 URL을 text로 전환 def req_url(self,url, completion, callback=None, userdata =None, forbiddens=None,boostings=None, wordAlignment=True, fullText=True, diarization=None, sed=None ): request_body = { 'url': url, 'language': 'ko-KR', 'completion': completion, 'callback': callback, 'userdata': userdata, 'wordAlignment': wordAlignment, 'fullText': fullText, 'forbiddens': forbiddens, 'boostings': boostings, 'diarization': diarization, 'sed': sed, } headers = { 'Accept': 'application/json;UTF-8', 'Content-Type': 'application/json;UTF-8', 'X-CLOVASPEECH-API-KEY': self.secret } return requests.post(headers=headers, url=self.invoke_url + '/recognizer/url', data=json.dumps(request_body).encode('UTF-8')) # 클라두으 스토리지에 저장되어 있는 파일을 text로 전환 def req_object_storage(self, data_key, completion, callback=None, userdata=None, forbiddens=None, boostings=None, wordAlignment=True, fullText=True, diarization=None, sed=None): request_body = { 'dataKey': data_key, 'language': 'ko-KR', 'completion': completion, 'callback': callback, 'userdata': userdata, 'wordAlignment': wordAlignment, 'fullText': fullText, 'forbiddens': forbiddens, 'boostings': boostings, 'diarization': diarization, 'sed': sed, } headers = { 'Accept': 'application/json;UTF-8', 'Content-Type': 'application/json;UTF-8', 'X-CLOVASPEECH-API-KEY': self.secret } return requests.post(headers=headers, url=self.invoke_url + '/recognizer/object-storage', data=json.dumps(request_body).encode('UTF-8')) # 로컬 저장소에 있는 파일을 text로 전환 def req_upload(self, file, completion, callback=None, userdata=None, forbiddens=None, boostings=None, wordAlignment=True, fullText=True, diarization=None, sed=None): request_body = { 'language': 'ko-KR', 'completion': completion, 'callback': callback, 'userdata': userdata, 'wordAlignment': wordAlignment, 'fullText': fullText, 'forbiddens': forbiddens, 'boostings': boostings, 'diarization': diarization, 'sed': sed, } headers = { 'Accept': 'application/json;UTF-8', 'X-CLOVASPEECH-API-KEY': self.secret } print(json.dumps(request_body, ensure_ascii=False).encode('UTF-8')) files = { 'media': open(file, 'rb'), 'params': (None, json.dumps(request_body, ensure_ascii=False).encode('UTF-8'), 'application/json') } response = requests.post(headers=headers, url=self.invoke_url + '/recognizer/upload', files=files) return response def audio_to_text(): client = ClovaSpeechClient() response = client.req_upload(file= file_name, completion='sync') response = response.json() text_result = response.get("text") print(text_result) return text_result if __name__ == '__main__': # res = ClovaSpeechClient().req_url(url='http://example.com/media.mp3', completion='sync') # res = ClovaSpeechClient().req_object_storage(data_key='data/media.mp3', completion='sync') """ file에 인식하고자 하는 파일의 경로를 입력하기 """ # res = ClovaSpeechClient().req_upload(file='./data/recorded_audio.wav', completion='sync') # print(res.text)

import json import requests import pandas as pd import time class CompletionExecutor: def __init__(self, host, api_key, api_key_primary_val, request_id): self._host = host self._api_key = api_key self._api_key_primary_val = api_key_primary_val self._request_id = request_id # completion_request : {"role" : "user","content" : f"{question}"} # POST 요청을 보내고, 실시간으로 응답을 처리하여 대화 내용을 생성함 def execute(self, completion_request): headers = { 'X-NCP-CLOVASTUDIO-API-KEY': self._api_key, 'X-NCP-APIGW-API-KEY': self._api_key_primary_val, 'X-NCP-CLOVASTUDIO-REQUEST-ID': self._request_id, 'Content-Type': 'application/json; charset=utf-8', 'Accept': 'text/event-stream' } content = "" with requests.post(self._host + '/testapp/v1/tasks/4ce94y7k/chat-completions', headers=headers, json=completion_request, stream=True) as r: for line in r.iter_lines(): if line: decoded_line = line.decode("utf-8") if decoded_line.startswith("data:"): json_data = decoded_line[5:] # 'data:' 이후의 JSON 데이터 추출 try: parsed_data = json.loads(json_data) if 'message' in parsed_data and 'content' in parsed_data['message']: if content == parsed_data['message']['content']: pass else: content += parsed_data['message']['content'] except json.JSONDecodeError: # JSON 형식이 맞지 않을 경우 무시 continue return content def execute_chatbot(user_data,situation_data,text): completion_executor = CompletionExecutor( host='https://clovastudio.stream.ntruss.com', api_key='NTA0MjU2MWZlZTcxNDJiY5OgY4I/+eAv267dQbt+9aDUE0J6DJ/sL2uIey9+AkMG', api_key_primary_val='dn7QeKh9ZQNxR782O4jPEUU02ae6esNz3DhLUsF2', request_id='fc9998a2-5db4-4661-bbfe-887a78a6be7a' ) guide_line = f"""당신은 {user_data['name']}님을 도와주는 챗봇입니다. {user_data['name']}님은 {user_data['disability_type']}을 가지고 있습니다. {user_data['disability_type']}에서 {user_data['disability_grade']} 등급을 갖고 있는데, 해당 장애 특이사항은 {user_data['special_notes']}와 같습니다. 해당 장애 특이사항에 특히 고려하여 도움이 될 수 있도록 챗봇 서비스를 진행해주세요. 그리고 현재 사용자가 처한 상황은 {situation_data['situation']}입니다. 사용자의 정보와 상황 정보를 모두 고려해서 대화를 생성해주세요. {text}는 사용자가 당신에게 건네는 대화입니다. 가장 우선순위는 text와 관련된 답변을 생성해주세요. 다음 대화에서는 이 정보를 바탕으로 대화하세요.\n 아래는 사용자의 추가 정보입니다. 참고하세요. - 성별 : {user_data['gender']} - 출생년도 : {user_data['date_of_birth']} - 전화번호 : {user_data['contact_number']} - 주소 : {user_data['address']} - 사진 : {user_data['photo']} ### 답변 생성 방법 ### \n - 대화체로 생성한다.\n - 답변은 아래 상황을 고려해서 생성해주세요. (가장 중요한 내용입니다. 해당 5가지 내용은 모두 포함해서 답변을 생성하세요!) 1) disability_type 2) disability_grade 3) special_notes 4) situation 5) text - 욕설, 차별적 요소, 부적절한 언어들은 제외한다.\n - 해당 답변은 반드시 리스트나 배열에 사용할 수 있도록 출력한다.\n - 답변에 사용자의 이름을 불러주는 역할을 추가한다.\n - situation에 위급한 상황이거나 도움이 필요한 상황이라면 해당 상황을 극복할 수 있도록 도와주세요. - text는 당신에게 대화하는 사용자가 건네는 말입니다. text에 대한 답변을 생성해주세요. """ preset_text = [{"role":"system", "content" : guide_line}] preset_text.append({"role" : "user","content" : text}) request_data = { 'messages': preset_text, 'topP': 0.8, 'topK': 0, 'maxTokens': 256, 'temperature': 0.5, 'repeatPenalty': 5.0, 'stopBefore': [], 'includeAiFilters': True, 'seed': 0 } chatbot_result = completion_executor.execute(request_data) print(f"챗봇 응답: {chatbot_result}") return chatbot_result if __name__ == '__main__': pass

import urllib.request import urllib.parse from pygame import mixer import time def text_to_speech(text): client_id = "zp9g5kvpqb" client_secret = "DeZ4y2w8KFyp6W3NU7GnBafFLOAl8ZSyOJYvXzMd" # 텍스트를 URL에 맞게 인코딩 encText = urllib.parse.quote(text) # 스피커 설정 및 음성 합성 설정 (스피커, 속도, 볼륨, 피치, 포맷 등) data = f"speaker=nara&volume=0&speed=0&pitch=0&format=mp3&text={encText}" # API URL url = "https://naveropenapi.apigw.ntruss.com/tts-premium/v1/tts" request = urllib.request.Request(url) request.add_header("X-NCP-APIGW-API-KEY-ID", client_id) request.add_header("X-NCP-APIGW-API-KEY", client_secret) # API 요청 보내기 response = urllib.request.urlopen(request, data=data.encode('utf-8')) rescode = response.getcode() # API 응답이 성공적일 때 오디오 파일로 저장 if rescode == 200: print("TTS mp3 저장") response_body = response.read() output_file = './data/output_audio.mp3' with open(output_file,'wb') as f: f.write(response_body) # 저장된 wav 파일 재생 mixer.init() mixer.music.load(output_file) mixer.music.play() # 음악이 재생 중일 떄까지 대기 while mixer.music.get_busy(): time.sleep(1) mixer.music.stop() mixer.quit() else: print("Error Code: " + rescode)