Callet/main.py at main · jangwonjun/Callet · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
from flask import Flask, jsonify, render_template, request
import paho.mqtt.client as mqtt
import ssl
import time
import pyttsx3
import speech_recognition as sr
import threading
import re
from transformers import PreTrainedTokenizerFast, GPT2LMHeadModel
import torch
import numpy as np

app = Flask(__name__)

# ===== MQTT 설정 =====
MQTT_BROKER = "voicecardwallet.r-e.kr"  # 실제 MQTT 브로커 주소로 변경하세요
MQTT_PORT = 8883


MQTT_TOPIC = "esp32/commands"  # ESP32가 구독하는 토픽

mqtt_client = mqtt.Client()

mqtt_client.tls_set(
    ca_certs="/etc/ssl/certs/ca-certificates.crt",  # 대부분 이걸로 충분함
    tls_version=ssl.PROTOCOL_TLSv1_2
)
mqtt_client.connect(MQTT_BROKER, MQTT_PORT, 60)

def on_connect(client, userdata, flags, rc):
    if rc == 0:
        print("MQTT 브로커 연결 성공")
    else:
        print(f"MQTT 연결 실패, 코드: {rc}")

mqtt_client.on_connect = on_connect
mqtt_client.connect(MQTT_BROKER, MQTT_PORT, 60)
mqtt_client.loop_start()

# ===== TTS 엔진 및 락 =====
tts_engine = pyttsx3.init()
tts_lock = threading.Lock()

# ===== 슬롯 데이터 =====
slots = {}

# ===== KoGPT2 모델 로드 =====
tokenizer = PreTrainedTokenizerFast.from_pretrained("skt/kogpt2-base-v2")
model = GPT2LMHeadModel.from_pretrained("skt/kogpt2-base-v2")
model.eval()

# ===== 동의어 그룹 =====
synonym_groups = [
    ["주민등록증", "민증", "등록증"],
    ["롯데카드", "롯데"],
    ["삼성카드", "삼성"],
]

# ===== 음성 합성 함수 =====
def speak(text):
    with tts_lock:
        tts_engine.say(text)
        tts_engine.runAndWait()

# ===== 음성 인식 함수 =====
def listen_command():
    r = sr.Recognizer()
    try:
        with sr.Microphone() as source:
            print("🎤 음성 입력 대기 중...")
            audio = r.listen(source, timeout=5, phrase_time_limit=7)
    except Exception as e:
        print(f"마이크 에러: {e}")
        return "인식 실패"

    try:
        text = r.recognize_google(audio, language='ko-KR')
        print(f"🎿 인식된 텍스트: {text}")
        return text
    except sr.UnknownValueError:
        return "인식 실패"
    except sr.RequestError as e:
        return f"API 오류: {e}"
# ===== MQTT 명령 전송 함수 =====
def send_esp32_command(command):
    try:
        mqtt_client.publish(MQTT_TOPIC, command)
        print(f"MQTT 명령 전송: {command}")
    except Exception as e:
        print(f"MQTT 명령 전송 오류: {e}")

# ===== 음성 인식 함수 =====
def listen_command():
    r = sr.Recognizer()
    try:
        with sr.Microphone() as source:
            print("🎤 음성 입력 대기 중...")
            audio = r.listen(source, timeout=5, phrase_time_limit=7)
    except Exception as e:
        print(f"마이크 에러: {e}")
        return "인식 실패"

    try:
        text = r.recognize_google(audio, language='ko-KR')
        print(f"🎿 인식된 텍스트: {text}")
        return text
    except sr.UnknownValueError:
        return "인식 실패"
    except sr.RequestError as e:
        return f"API 오류: {e}"