|
import socket |
|
import os |
|
import torch |
|
from threading import Timer |
|
import pyttsx3 |
|
import speech_recognition as sr |
|
from detectfaces import fer |
|
from models.PosterV2_7cls import pyramid_trans_expr2 |
|
from main import RecorderMeter1, RecorderMeter |
|
import time |
|
|
|
script_dir = os.path.dirname(os.path.abspath(__file__)) |
|
|
|
|
|
model_path = os.path.join(script_dir,"models","checkpoints","raf-db-model_best.pth") |
|
|
|
|
|
if torch.backends.mps.is_available(): |
|
device = "mps" |
|
elif torch.cuda.is_available(): |
|
device = "cuda" |
|
else: |
|
device = "cpu" |
|
|
|
|
|
model = pyramid_trans_expr2(img_size=224, num_classes=7) |
|
|
|
|
|
model = torch.nn.DataParallel(model) |
|
|
|
|
|
model = model.to(device) |
|
fer(model_path=model_path, device=device, model=model) |
|
|
|
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) |
|
s.bind(('', 5001)) |
|
s.listen(5) |
|
print("Bot is Running") |
|
|
|
def handle_client(clientsocket): |
|
r = sr.Recognizer() |
|
m = sr.Microphone() |
|
|
|
try: |
|
while True: |
|
prompt = '' |
|
print("Speak now:") |
|
sent = False |
|
with m as source: |
|
audio = r.listen(source) |
|
|
|
try: |
|
prompt = r.recognize_google(audio) |
|
print("Tadbot Thinks you said:", prompt) |
|
sent = True |
|
except sr.UnknownValueError: |
|
print("Tadbot could not understand audio. Try Again") |
|
except sr.RequestError as e: |
|
print(f"Could not request results from Google Speech Recognition service: {e}") |
|
|
|
if sent: |
|
print("please Wait!") |
|
try: |
|
clientsocket.send(bytes(prompt, 'utf-8')) |
|
response = clientsocket.recv(1024).decode("utf-8") |
|
engine = pyttsx3.init('espeak') |
|
voices = engine.getProperty('voices') |
|
engine.setProperty('voice', voices[26].id) |
|
engine.setProperty('rate', 145) |
|
engine.say(response) |
|
engine.runAndWait() |
|
print("TADBot:", response) |
|
except (socket.error, ConnectionResetError) as e: |
|
print(f"Connection error: {e}") |
|
break |
|
time.sleep(60) |
|
|
|
finally: |
|
clientsocket.close() |
|
print("Connection Closed") |
|
|
|
|
|
while True: |
|
try: |
|
clientsocket, address = s.accept() |
|
print(f"Accepted connection from {address}") |
|
handle_client(clientsocket) |
|
except KeyboardInterrupt: |
|
print("Server shutting down...") |
|
break |
|
except Exception as e: |
|
print(f"An error occurred: {e}") |
|
|
|
s.close() |