Spaces:
Build error
Build error
| 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) | |
| 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}") | |
| 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) |