569 lines
22 KiB
Python
569 lines
22 KiB
Python
import flet as ft
|
|
import configparser
|
|
import os
|
|
import numpy as np
|
|
import wave
|
|
import tempfile
|
|
from modules.audio_source import AudioSource
|
|
from modules.speech_recognition import SpeechRecognizer
|
|
from modules.conversation_analyzer import ConversationAnalyzer
|
|
from modules.database_manager import DatabaseManager
|
|
from modules.gui_components import (
|
|
AudioSourceSelector,
|
|
StatusIndicator,
|
|
ConversationView,
|
|
PreviousConversationsList,
|
|
TrainInfoPanel
|
|
)
|
|
|
|
def main(page: ft.Page):
|
|
# 설정 파일 로드
|
|
config = configparser.ConfigParser()
|
|
|
|
# 기본 설정 생성
|
|
if not os.path.exists("config.ini"):
|
|
config["api"] = {
|
|
"openai_api_key": "your_openai_api_key_here",
|
|
"huggingface_api_key": "your_huggingface_api_key_here"
|
|
}
|
|
config["audio"] = {
|
|
"sample_rate": "16000",
|
|
"silence_threshold": "0.02",
|
|
"silence_duration": "60",
|
|
"buffer_duration": "3"
|
|
}
|
|
config["app"] = {
|
|
"theme": "light"
|
|
}
|
|
config["model"] = {
|
|
"provider": "huggingface",
|
|
"name": "facebook/wav2vec2-base-960h"
|
|
}
|
|
|
|
with open("config.ini", "w") as configfile:
|
|
config.write(configfile)
|
|
else:
|
|
config.read("config.ini")
|
|
|
|
# 테마 설정
|
|
theme = config.get("app", "theme", fallback="light")
|
|
page.theme_mode = ft.ThemeMode.LIGHT if theme.lower() == "light" else ft.ThemeMode.DARK
|
|
|
|
page.title = "철도 음성인식 시스템"
|
|
page.window_width = 1200
|
|
page.window_height = 800
|
|
page.padding = 10
|
|
|
|
# 인스턴스 생성
|
|
audio_source = AudioSource()
|
|
speech_recognizer = SpeechRecognizer()
|
|
conversation_analyzer = ConversationAnalyzer()
|
|
db_manager = DatabaseManager()
|
|
|
|
# 테스트 모드 상태
|
|
is_test_mode = False
|
|
|
|
# GUI 컴포넌트 생성
|
|
audio_source_selector = AudioSourceSelector(audio_source)
|
|
status_indicator = StatusIndicator()
|
|
conversation_view = ConversationView()
|
|
previous_conversations_list = PreviousConversationsList(conversation_view)
|
|
train_info_panel = TrainInfoPanel(db_manager)
|
|
|
|
# 실시간 처리 상태 표시기
|
|
realtime_indicator = ft.Text("", italic=True, color=ft.colors.GREY_600)
|
|
|
|
# 파일 선택 콜백 함수 정의
|
|
def on_file_picked(e):
|
|
if e.files and len(e.files) > 0:
|
|
file_path = e.files[0].path
|
|
test_status.value = f"선택된 파일: {os.path.basename(file_path)}"
|
|
page.update()
|
|
|
|
# 선택된 파일 처리
|
|
process_audio_file(file_path)
|
|
|
|
# 테스트 모드 컨트롤
|
|
test_mode_switch = ft.Switch(label="테스트 모드", value=False, on_change=lambda e: toggle_test_mode(e.control.value))
|
|
file_picker = ft.FilePicker(on_result=on_file_picked)
|
|
page.overlay.append(file_picker)
|
|
|
|
test_file_button = ft.ElevatedButton(
|
|
"MP3 파일 선택",
|
|
icon=ft.icons.UPLOAD_FILE,
|
|
on_click=lambda _: file_picker.pick_files(
|
|
allow_multiple=False,
|
|
allowed_extensions=["mp3", "wav"]
|
|
),
|
|
disabled=True
|
|
)
|
|
|
|
# 테스트 모드 상태 표시
|
|
test_status = ft.Text("", italic=True, color=ft.colors.ORANGE)
|
|
|
|
# 레이아웃 구성
|
|
left_panel = ft.Column([
|
|
ft.Container(
|
|
content=previous_conversations_list,
|
|
height=320,
|
|
border=ft.border.all(1, ft.colors.GREY_400),
|
|
border_radius=5,
|
|
padding=5
|
|
),
|
|
ft.Container(
|
|
content=conversation_view,
|
|
height=480,
|
|
border=ft.border.all(1, ft.colors.GREY_400),
|
|
border_radius=5,
|
|
padding=5,
|
|
expand=True
|
|
),
|
|
], expand=True)
|
|
|
|
right_panel = ft.Container(
|
|
content=train_info_panel,
|
|
border=ft.border.all(1, ft.colors.GREY_400),
|
|
border_radius=5,
|
|
padding=10,
|
|
expand=True
|
|
)
|
|
|
|
# 상단 컨트롤에 설정 버튼 추가
|
|
top_controls = ft.Row([
|
|
audio_source_selector,
|
|
status_indicator,
|
|
ft.ElevatedButton("시작", on_click=lambda e: start_monitoring()),
|
|
ft.ElevatedButton("중지", on_click=lambda e: stop_monitoring()),
|
|
ft.IconButton(
|
|
icon=ft.icons.SETTINGS,
|
|
tooltip="설정",
|
|
on_click=lambda e: show_settings_dialog()
|
|
),
|
|
])
|
|
|
|
# 테스트 모드 컨트롤
|
|
test_controls = ft.Row([
|
|
test_mode_switch,
|
|
test_file_button,
|
|
test_status
|
|
], visible=True)
|
|
|
|
# 메인 레이아웃에 실시간 인디케이터 추가
|
|
main_layout = ft.Column([
|
|
top_controls,
|
|
test_controls,
|
|
realtime_indicator,
|
|
ft.Row([
|
|
left_panel,
|
|
right_panel,
|
|
], expand=True),
|
|
], expand=True)
|
|
|
|
page.add(main_layout)
|
|
|
|
# 테스트 모드 토글 함수
|
|
def toggle_test_mode(value):
|
|
nonlocal is_test_mode
|
|
is_test_mode = value
|
|
test_file_button.disabled = not value
|
|
|
|
if value:
|
|
test_status.value = "테스트 모드: 활성화됨"
|
|
audio_source_selector.disabled = True
|
|
else:
|
|
test_status.value = "테스트 모드: 비활성화됨"
|
|
audio_source_selector.disabled = False
|
|
|
|
page.update()
|
|
|
|
# 오디오 파일 처리
|
|
def process_audio_file(file_path):
|
|
try:
|
|
status_indicator.set_status("파일 처리 중")
|
|
status_indicator.set_detecting(True)
|
|
page.update()
|
|
|
|
# MP3 또는 WAV 파일 직접 처리
|
|
text = speech_recognizer.recognize_file(file_path)
|
|
|
|
# 변환 실패 시 MP3 파일인 경우 추가 처리
|
|
if text is None and file_path.lower().endswith('.mp3'):
|
|
try:
|
|
# MP3를 WAV로 변환 (librosa 사용)
|
|
import librosa
|
|
|
|
# MP3 파일 로드
|
|
audio_data, sample_rate = librosa.load(file_path, sr=16000, mono=True)
|
|
|
|
# WAV 파일로 변환
|
|
fd, temp_wav_path = tempfile.mkstemp(suffix=".wav")
|
|
os.close(fd) # 파일 디스크립터 즉시 닫기
|
|
|
|
try:
|
|
# 16비트 정수로 변환
|
|
audio_data_int = (audio_data * 32767).astype(np.int16)
|
|
|
|
# WAV 파일 저장 - Wave_write 객체 사용
|
|
wf = wave.open(temp_wav_path, 'wb')
|
|
try:
|
|
wf.setnchannels(1)
|
|
wf.setsampwidth(2) # 16-bit
|
|
wf.setframerate(16000)
|
|
wf.writeframes(audio_data_int.tobytes())
|
|
finally:
|
|
wf.close() # 명시적으로 닫아줌
|
|
|
|
# OpenAI의 API를 사용해 음성 인식
|
|
text = speech_recognizer.recognize_file(temp_wav_path)
|
|
finally:
|
|
# 임시 파일 항상 삭제
|
|
if os.path.exists(temp_wav_path):
|
|
try:
|
|
os.unlink(temp_wav_path)
|
|
except Exception as e:
|
|
print(f"임시 파일 삭제 오류: {e}")
|
|
except Exception as mp3_error:
|
|
print(f"MP3 처리 오류: {mp3_error}")
|
|
|
|
# 인식 결과 처리
|
|
if text:
|
|
realtime_indicator.value = f"파일 인식 결과: {text[:100]}..."
|
|
|
|
speakers = conversation_analyzer.identify_speakers(text)
|
|
context = conversation_analyzer.analyze_conversation(text, speakers)
|
|
|
|
conversation_view.update_conversation(text, speakers)
|
|
previous_conversations_list.add_conversation(context)
|
|
|
|
# 데이터베이스에서 관련 정보 검색
|
|
train_info = db_manager.get_train_info(context)
|
|
fault_info = db_manager.get_fault_info(context)
|
|
|
|
train_info_panel.update_info(train_info, fault_info)
|
|
else:
|
|
realtime_indicator.value = "파일 인식 실패: 텍스트를 추출할 수 없습니다."
|
|
|
|
status_indicator.set_detecting(False)
|
|
status_indicator.set_status("대기 중")
|
|
page.update()
|
|
|
|
except Exception as e:
|
|
print(f"오디오 파일 처리 오류: {e}")
|
|
realtime_indicator.value = f"오류: {str(e)}"
|
|
status_indicator.set_detecting(False)
|
|
status_indicator.set_status("오류")
|
|
page.update()
|
|
|
|
# 음원 감시 관련 함수
|
|
def start_monitoring():
|
|
if is_test_mode:
|
|
# 테스트 모드에서는 파일 선택을 유도
|
|
file_picker.pick_files(
|
|
allow_multiple=False,
|
|
allowed_extensions=["mp3", "wav"]
|
|
)
|
|
return
|
|
|
|
try:
|
|
selected_source = audio_source_selector.get_selected_source()
|
|
if not selected_source:
|
|
page.snack_bar = ft.SnackBar(ft.Text("음원 소스를 선택해주세요"))
|
|
page.snack_bar.open = True
|
|
page.update()
|
|
return
|
|
|
|
status_indicator.set_status("녹음 중")
|
|
# 일반 및 실시간 콜백 함수 전달
|
|
audio_source.start_recording(selected_source, on_audio_data, on_realtime_audio_data)
|
|
page.update()
|
|
except Exception as e:
|
|
print(f"녹음 시작 오류: {e}")
|
|
page.snack_bar = ft.SnackBar(ft.Text(f"녹음 시작 오류: {e}"))
|
|
page.snack_bar.open = True
|
|
page.update()
|
|
|
|
def stop_monitoring():
|
|
try:
|
|
audio_source.stop_recording()
|
|
|
|
# 실시간 인식 중지
|
|
if hasattr(speech_recognizer, 'stop_realtime_recognition'):
|
|
speech_recognizer.stop_realtime_recognition()
|
|
|
|
status_indicator.set_status("대기 중")
|
|
realtime_indicator.value = ""
|
|
page.update()
|
|
except Exception as e:
|
|
print(f"녹음 중지 오류: {e}")
|
|
|
|
def on_audio_data(audio_data):
|
|
"""완성된 오디오 데이터 처리 콜백"""
|
|
try:
|
|
if len(audio_data) == 0:
|
|
return
|
|
|
|
status_indicator.set_detecting(True)
|
|
page.update()
|
|
|
|
text = speech_recognizer.recognize(audio_data)
|
|
if text:
|
|
speakers = conversation_analyzer.identify_speakers(text)
|
|
context = conversation_analyzer.analyze_conversation(text, speakers)
|
|
|
|
conversation_view.update_conversation(text, speakers)
|
|
previous_conversations_list.add_conversation(context)
|
|
|
|
# 데이터베이스에서 관련 정보 검색
|
|
train_info = db_manager.get_train_info(context)
|
|
fault_info = db_manager.get_fault_info(context)
|
|
|
|
train_info_panel.update_info(train_info, fault_info)
|
|
|
|
status_indicator.set_detecting(False)
|
|
page.update()
|
|
except Exception as e:
|
|
print(f"오디오 데이터 처리 오류: {e}")
|
|
status_indicator.set_detecting(False)
|
|
page.update()
|
|
|
|
def on_realtime_audio_data(audio_data):
|
|
"""실시간 오디오 데이터 처리 콜백"""
|
|
try:
|
|
if len(audio_data) == 0:
|
|
return
|
|
|
|
status_indicator.set_detecting(True)
|
|
|
|
# 실시간 인식 결과 처리 콜백
|
|
def on_realtime_result(text):
|
|
if text:
|
|
realtime_indicator.value = f"실시간 인식 중: {text[:100]}..."
|
|
page.update()
|
|
|
|
# 실시간 처리 지원 확인
|
|
if hasattr(speech_recognizer, 'add_audio_data') and hasattr(speech_recognizer, 'start_realtime_recognition'):
|
|
# 큐에 오디오 데이터 추가
|
|
speech_recognizer.add_audio_data(audio_data)
|
|
|
|
# 처리가 시작되지 않았으면 시작
|
|
if hasattr(speech_recognizer, 'is_processing') and not speech_recognizer.is_processing:
|
|
speech_recognizer.start_realtime_recognition(on_realtime_result)
|
|
else:
|
|
# 실시간 변환을 지원하지 않는 경우 일반 변환 사용
|
|
text = speech_recognizer.recognize(audio_data)
|
|
if text:
|
|
realtime_indicator.value = f"감지: {text[:100]}..."
|
|
|
|
page.update()
|
|
except Exception as e:
|
|
print(f"실시간 오디오 처리 오류: {e}")
|
|
|
|
def show_settings_dialog():
|
|
"""설정 대화상자 표시"""
|
|
try:
|
|
# API 키 입력 필드
|
|
openai_api_key_field = ft.TextField(
|
|
label="OpenAI API 키",
|
|
value=speech_recognizer.api_key,
|
|
password=True,
|
|
width=400
|
|
)
|
|
|
|
# Hugging Face API 키 입력 필드
|
|
hf_api_key_field = ft.TextField(
|
|
label="Hugging Face API 키",
|
|
value=speech_recognizer.hf_api_key,
|
|
password=True,
|
|
width=400
|
|
)
|
|
|
|
# 모델 선택 관련 설정
|
|
# 모델 제공자 선택
|
|
model_provider_radio = ft.RadioGroup(
|
|
content=ft.Column([
|
|
ft.Radio(value="huggingface", label="Hugging Face (한국어 음성인식 권장)"),
|
|
ft.Radio(value="openai", label="OpenAI Whisper (백업)"),
|
|
ft.Radio(value="vosk", label="VOSK 오프라인 (완전 오프라인)"),
|
|
]),
|
|
value=speech_recognizer.model_provider
|
|
)
|
|
|
|
# 음성인식 모델 선택 드롭다운
|
|
model_options = [
|
|
ft.dropdown.Option("vosk-model-small-kr", "VOSK 모델 - 완전 오프라인 한국어 소형 모델"),
|
|
ft.dropdown.Option("kresnik/wav2vec2-large-xlsr-korean", "한국어 특화 음성인식 모델"),
|
|
ft.dropdown.Option("openai/whisper-small", "openai 한국어 특화 Whisper 소형 모델"),
|
|
ft.dropdown.Option("openai/whisper-medium", "openai 한국어 특화 Whisper 중형 모델"),
|
|
ft.dropdown.Option("openai/whisper-large-v3", "openai 한국어 특화 Whisper 대형 모델"),
|
|
ft.dropdown.Option("facebook/wav2vec2-base-960h", "영어 음성인식 기본 모델")
|
|
]
|
|
|
|
model_dropdown = ft.Dropdown(
|
|
label="음성인식 모델 선택",
|
|
width=400,
|
|
options=model_options,
|
|
value=speech_recognizer.model_name,
|
|
)
|
|
|
|
# 오디오 설정 슬라이더
|
|
silence_threshold_slider = ft.Slider(
|
|
min=0.01,
|
|
max=0.1,
|
|
divisions=9,
|
|
value=float(config.get("audio", "silence_threshold", fallback="0.02")),
|
|
label="{value}",
|
|
on_change=lambda e: update_slider_label(e)
|
|
)
|
|
|
|
silence_threshold_text = ft.Text(f"소리 감지 임계값: {silence_threshold_slider.value}")
|
|
|
|
silence_duration_slider = ft.Slider(
|
|
min=10,
|
|
max=120,
|
|
divisions=11,
|
|
value=float(config.get("audio", "silence_duration", fallback="60")),
|
|
label="{value}",
|
|
on_change=lambda e: update_duration_label(e)
|
|
)
|
|
|
|
silence_duration_text = ft.Text(f"대화 구분 시간(초): {silence_duration_slider.value}")
|
|
|
|
buffer_duration_slider = ft.Slider(
|
|
min=1,
|
|
max=10,
|
|
divisions=9,
|
|
value=float(config.get("audio", "buffer_duration", fallback="3")),
|
|
label="{value}",
|
|
on_change=lambda e: update_buffer_label(e)
|
|
)
|
|
|
|
buffer_duration_text = ft.Text(f"버퍼 길이(초): {buffer_duration_slider.value}")
|
|
|
|
# 슬라이더 레이블 업데이트 함수
|
|
def update_slider_label(e):
|
|
silence_threshold_text.value = f"소리 감지 임계값: {e.control.value:.2f}"
|
|
page.update()
|
|
|
|
def update_duration_label(e):
|
|
silence_duration_text.value = f"대화 구분 시간(초): {e.control.value:.0f}"
|
|
page.update()
|
|
|
|
def update_buffer_label(e):
|
|
buffer_duration_text.value = f"버퍼 길이(초): {e.control.value:.0f}"
|
|
page.update()
|
|
|
|
# 테마 선택 라디오 버튼
|
|
theme_radio = ft.RadioGroup(
|
|
content=ft.Column([
|
|
ft.Radio(value="light", label="밝은 테마"),
|
|
ft.Radio(value="dark", label="어두운 테마"),
|
|
]),
|
|
value=theme
|
|
)
|
|
|
|
# 설정 저장 함수
|
|
def save_settings(e):
|
|
try:
|
|
# API 키 저장
|
|
if openai_api_key_field.value:
|
|
speech_recognizer.set_api_key(openai_api_key_field.value)
|
|
|
|
# Hugging Face API 키 저장
|
|
if hf_api_key_field.value:
|
|
speech_recognizer.set_huggingface_api_key(hf_api_key_field.value)
|
|
|
|
# 모델 설정 저장
|
|
if model_provider_radio.value and model_dropdown.value:
|
|
speech_recognizer.set_model(
|
|
model_provider_radio.value,
|
|
model_dropdown.value
|
|
)
|
|
|
|
# 오디오 설정 저장
|
|
if hasattr(audio_source, 'update_settings'):
|
|
audio_source.update_settings(
|
|
silence_threshold=silence_threshold_slider.value,
|
|
silence_duration=silence_duration_slider.value,
|
|
buffer_duration=buffer_duration_slider.value
|
|
)
|
|
else:
|
|
# 직접 속성 설정
|
|
audio_source.silence_threshold = silence_threshold_slider.value
|
|
audio_source.silence_duration = silence_duration_slider.value
|
|
audio_source.buffer_duration = buffer_duration_slider.value
|
|
|
|
# 설정 저장
|
|
config.set("audio", "silence_threshold", str(silence_threshold_slider.value))
|
|
config.set("audio", "silence_duration", str(silence_duration_slider.value))
|
|
config.set("audio", "buffer_duration", str(buffer_duration_slider.value))
|
|
|
|
# 테마 설정 저장
|
|
config.set("app", "theme", theme_radio.value)
|
|
with open("config.ini", "w") as config_file:
|
|
config.write(config_file)
|
|
|
|
# 테마 적용
|
|
page.theme_mode = ft.ThemeMode.LIGHT if theme_radio.value.lower() == "light" else ft.ThemeMode.DARK
|
|
|
|
# 대화상자 닫기
|
|
dialog.open = False
|
|
page.update()
|
|
|
|
# 설정 적용 알림
|
|
page.snack_bar = ft.SnackBar(ft.Text("설정이 저장되었습니다"))
|
|
page.snack_bar.open = True
|
|
page.update()
|
|
except Exception as e:
|
|
print(f"설정 저장 오류: {e}")
|
|
page.snack_bar = ft.SnackBar(ft.Text(f"설정 저장 오류: {e}"))
|
|
page.snack_bar.open = True
|
|
page.update()
|
|
|
|
# 설정 대화상자
|
|
dialog = ft.AlertDialog(
|
|
title=ft.Text("설정"),
|
|
content=ft.Column([
|
|
ft.Text("API 설정", weight=ft.FontWeight.BOLD),
|
|
ft.Text("OpenAI API (백업용)", size=14),
|
|
openai_api_key_field,
|
|
ft.Divider(height=10),
|
|
ft.Text("Hugging Face API (권장)", size=14),
|
|
hf_api_key_field,
|
|
|
|
ft.Divider(height=20),
|
|
ft.Text("음성인식 모델 설정", weight=ft.FontWeight.BOLD),
|
|
ft.Text("API 제공자 선택", size=14),
|
|
model_provider_radio,
|
|
ft.Text("음성인식 모델 선택", size=14),
|
|
model_dropdown,
|
|
|
|
ft.Divider(height=20),
|
|
ft.Text("오디오 설정", weight=ft.FontWeight.BOLD),
|
|
silence_threshold_text,
|
|
silence_threshold_slider,
|
|
silence_duration_text,
|
|
silence_duration_slider,
|
|
buffer_duration_text,
|
|
buffer_duration_slider,
|
|
|
|
ft.Divider(height=20),
|
|
ft.Text("앱 설정", weight=ft.FontWeight.BOLD),
|
|
ft.Text("테마", size=14),
|
|
theme_radio,
|
|
], scroll=ft.ScrollMode.AUTO, height=600),
|
|
actions=[
|
|
ft.TextButton("취소", on_click=lambda e: setattr(dialog, "open", False)),
|
|
ft.TextButton("저장", on_click=save_settings),
|
|
],
|
|
actions_alignment=ft.MainAxisAlignment.END,
|
|
)
|
|
|
|
# 대화상자 표시
|
|
page.dialog = dialog
|
|
dialog.open = True
|
|
page.update()
|
|
except Exception as e:
|
|
print(f"설정 대화상자 표시 오류: {e}")
|
|
|
|
if __name__ == "__main__":
|
|
ft.app(target=main) |