Spaces:
Build error
Build error
File size: 6,344 Bytes
26440d6 | 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 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 | import requests
import json
import base64
from typing import Optional, Dict, List
import os
from pydantic import BaseModel, Field
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class ApiError(Exception):
"""Custom exception class for API errors."""
pass
class DialogueRequestBase(BaseModel):
chess_piece_id: int = Field(..., alias="IdPieza")
language: str = Field("en", alias="Language")
empathy: float = Field(..., alias="Empathy")
def get_endpoint(self) -> str:
pass
class TextDialogueRequest(DialogueRequestBase):
text_input: str = Field(..., alias="Input")
def get_endpoint(self) -> str:
return "process_text"
class AudioDialogueRequest(DialogueRequestBase):
audio_file_path: str = Field(..., alias="AudioData")
def get_endpoint(self) -> str:
return "process_audio"
class WebcamDialogueRequest(DialogueRequestBase):
audio_file_path: str = Field(..., alias="AudioData")
images: List[str] = Field(..., alias="Images")
def get_endpoint(self) -> str:
return "process_images"
class LLMGeneration(BaseModel):
TextResponse: str
Decision: bool
class ChessPieceResponse(BaseModel):
ChessPieceLLMGeneration: LLMGeneration
AudioFile: Optional[str] = None
PieceSpeechVelocity: Optional[str] = None
class PlayerVoiceAttributes(BaseModel):
Transcription: Optional[str] = None
Language: Optional[str] = None
Accent: Optional[str] = None
SpeechVelocity: str = "unknown"
class DialogueResponseData(BaseModel):
ChessPieceResponse: ChessPieceResponse
player_voice_attributes: Optional[PlayerVoiceAttributes] = None
class DialogueResponse(BaseModel):
response: str
decision: bool
transcription: Optional[str] = None
audio_data: Optional[str] = None
player_speak_velocity: Optional[str] = None
piece_speak_velocity: Optional[str] = None
class ApiClient:
def __init__(self, base_url: str):
self.base_url = base_url
def send_request(self, endpoint: str, data: DialogueRequestBase) -> DialogueResponse:
try:
url = f"{self.base_url}/{endpoint}"
response = requests.post(url, json=data.dict(by_alias=True), headers={"Content-Type": "application/json"})
response.raise_for_status()
# print(response.json())
result = DialogueResponseData(**response.json())
chess_response = result.ChessPieceResponse
player_attributes = result.player_voice_attributes
transcription = player_attributes.Transcription if player_attributes else None
player_speak_velocity = None
piece_speak_velocity = None
if player_attributes:
player_speak_velocity = player_attributes.SpeechVelocity if player_attributes else 'unknown'
piece_speak_velocity = result.ChessPieceResponse.PieceSpeechVelocity
print("VELOCITIES!")
print(f"Piece: {piece_speak_velocity}")
print(f"Player: {player_speak_velocity}")
response_text = chess_response.ChessPieceLLMGeneration.TextResponse
decision = chess_response.ChessPieceLLMGeneration.Decision
audio_data = chess_response.AudioFile
return DialogueResponse(
response=response_text,
decision=decision,
transcription=transcription,
audio_data=audio_data,
player_speak_velocity=player_speak_velocity,
piece_speak_velocity=piece_speak_velocity
)
except requests.RequestException as e:
raise ApiError(f"API request failed: {str(e)}")
except json.JSONDecodeError:
raise ApiError("Failed to parse API response")
class DialogueProcessor:
def __init__(self, api_client: ApiClient):
self.api_client = api_client
def process_request(self, request: DialogueRequestBase) -> DialogueResponse:
try:
result = self.api_client.send_request(request.get_endpoint(), request)
return result
except ApiError as e:
return DialogueResponse(response=str(e), decision=False, transcription=None, audio_data=None)
@staticmethod
def save_audio_file(audio_data: str, filename: str):
filename = f"./assets/{filename}" # Needed to play audio files
audio_bytes = base64.b64decode(audio_data)
# Remove the file if it already exists to prevent permission issues
if os.path.exists(filename):
print("audio path exists")
os.remove(filename)
with open(filename, "wb") as audio_file:
logger.info(f"Writing file!")
audio_file.write(audio_bytes)
print(f"Saved audio to {filename}")
logger.info(f"Saved audio to {filename}")
@staticmethod
def encode_image(image_path: str) -> str:
with open(image_path, "rb") as image_file:
return base64.b64encode(image_file.read()).decode('utf-8')
class LocalApiDialogueProcessor:
def __init__(self, base_url: str = "http://localhost:49521"):
api_client = ApiClient(base_url)
self.processor = DialogueProcessor(api_client)
def process_text_input(self, chess_piece_id: int, text_input: str, language: str, empathy: float) -> DialogueResponse:
request = TextDialogueRequest(IdPieza=chess_piece_id, Input=text_input, Language=language, Empathy=empathy)
return self.processor.process_request(request)
def process_audio_input(self, chess_piece_id: int, audio_file_path: str, language: str, empathy: float) -> DialogueResponse:
request = AudioDialogueRequest(IdPieza=chess_piece_id, AudioData=audio_file_path, Language=language, Empathy=empathy)
return self.processor.process_request(request)
def process_webcam_input(self, chess_piece_id: int, audio_file_path: str, image_paths: List[str], language: str, empathy: float) -> DialogueResponse:
encoded_images = [DialogueProcessor.encode_image(img_path) for img_path in image_paths]
request = WebcamDialogueRequest(IdPieza=chess_piece_id, AudioData=audio_file_path, Images=encoded_images, Language=language, Empathy=empathy)
return self.processor.process_request(request) |