Spaces:
Build error
Build error
| import flet as ft | |
| import chess | |
| from domain.chess_piece_model import ChessPiece, PieceType, PieceID | |
| from domain.translations import spanish_translations | |
| from domain.localapi_dialog_processor import LocalApiDialogueProcessor | |
| import logging | |
| import asyncio | |
| from typing import Optional, List, Callable, Tuple | |
| from audio_utils import get_base64_audio | |
| from pydantic import BaseModel, Field | |
| from camera_utils import CameraImage | |
| logging.basicConfig(level=logging.INFO) | |
| logger = logging.getLogger(__name__) | |
| class EmpathyConfig: | |
| INITIAL_EMPATHY = 0.5 | |
| INITIAL_EMPATHY_PAWN = 0.1 | |
| INITIAL_EMPATHY_OTHER_PIECES = 0.8 | |
| DECREASE_AMOUNT = 0.02 | |
| MIN_EMPATHY = 0.05 | |
| HIGH_EMPATHY_THRESHOLD = 0.9 | |
| FREE_MOVE_EMPATHY = 0.5 | |
| STEP_AMOUNT = 0.2 | |
| class ChessModel: | |
| def __init__(self): | |
| self.board = chess.Board() | |
| self.piece_mapping = self._initialize_piece_mapping() | |
| self.last_move = None | |
| def is_king_empathy_zero_or_below(self) -> bool: | |
| for piece in self.piece_mapping.values(): | |
| if piece.piece_type == PieceType.KING and piece.empathy <= 0.0: | |
| return True | |
| return False | |
| def are_all_pieces_empathy_above(self, threshold: float) -> bool: | |
| print(f"Checking if all pieces have empathy above {threshold}") | |
| result = all(piece.empathy >= threshold for piece in self.piece_mapping.values()) | |
| print(f"All pieces above threshold: {result}") | |
| return result | |
| def remove_piece(self, square: chess.Square): | |
| piece = self.piece_mapping.pop(square, None) | |
| if piece: | |
| self.board.remove_piece_at(square) | |
| self.last_move = chess.Move(square, square) | |
| print(f"Piece at {chess.SQUARE_NAMES[square]} has been eliminated") | |
| else: | |
| print(f"No piece at {chess.SQUARE_NAMES[square]} to remove") | |
| def _initialize_piece_mapping(self): | |
| mapping = {} | |
| white_pieces = [ | |
| PieceID.WhiteRook1, PieceID.WhiteKnight1, PieceID.WhiteBishop1, PieceID.WhiteQueen, PieceID.WhiteKing, | |
| PieceID.WhiteBishop2, PieceID.WhiteKnight2, PieceID.WhiteRook2, | |
| PieceID.WhitePawn1, PieceID.WhitePawn2, PieceID.WhitePawn3, PieceID.WhitePawn4, | |
| PieceID.WhitePawn5, PieceID.WhitePawn6, PieceID.WhitePawn7, PieceID.WhitePawn8 | |
| ] | |
| black_pieces = [ | |
| PieceID.WhitePawn1, PieceID.WhitePawn2, PieceID.WhitePawn3, PieceID.WhitePawn4, | |
| PieceID.WhitePawn5, PieceID.WhitePawn6, PieceID.WhitePawn7, PieceID.WhitePawn8, | |
| PieceID.WhiteRook1, PieceID.WhiteKnight1, PieceID.WhiteBishop1, PieceID.WhiteQueen, PieceID.WhiteKing, | |
| PieceID.WhiteBishop2, PieceID.WhiteKnight2, PieceID.WhiteRook2, | |
| ] | |
| for square in chess.SQUARES: | |
| piece = self.board.piece_at(square) | |
| if piece: | |
| piece_type = PieceType(piece.piece_type) | |
| color = piece.color | |
| if color == chess.WHITE: | |
| piece_id = white_pieces.pop(0) | |
| else: | |
| piece_id = black_pieces.pop(0) | |
| chess_piece = ChessPiece(piece_type, color, piece_id) | |
| if piece_type == PieceType.PAWN: | |
| chess_piece.empathy = EmpathyConfig.INITIAL_EMPATHY_PAWN | |
| else: | |
| chess_piece.empathy = EmpathyConfig.INITIAL_EMPATHY_OTHER_PIECES | |
| mapping[square] = chess_piece | |
| logger.info(f"Piece at {chess.SQUARE_NAMES[square]}: {chess_piece} (Empathy: {chess_piece.empathy:.2f})") | |
| return mapping | |
| def get_piece_at(self, square: chess.Square): | |
| chess_piece = self.piece_mapping.get(square) | |
| if chess_piece: | |
| return chess_piece.symbol, chess_piece | |
| return " ", None | |
| def make_move(self, move: chess.Move) -> bool: | |
| if move in self.board.legal_moves: | |
| from_square = move.from_square | |
| to_square = move.to_square | |
| moving_piece = self.piece_mapping.get(from_square) | |
| if moving_piece: | |
| logger.info(f"Moving {moving_piece} from {chess.SQUARE_NAMES[from_square]} to {chess.SQUARE_NAMES[to_square]} (Empathy: {moving_piece.empathy:.2f})") | |
| captured_piece = self.piece_mapping.get(to_square) | |
| if captured_piece: | |
| logger.info(f"Captured {captured_piece} at {chess.SQUARE_NAMES[to_square]} (Empathy: {captured_piece.empathy:.2f})") | |
| # Handle pawn promotion | |
| if move.promotion: | |
| promoted_piece_type = PieceType(move.promotion) | |
| promoted_piece = ChessPiece(promoted_piece_type, moving_piece.color) | |
| promoted_piece.empathy = moving_piece.empathy # Preserve empathy | |
| self.piece_mapping[to_square] = promoted_piece | |
| logger.info(f"Pawn promoted to {promoted_piece_type} at {chess.SQUARE_NAMES[to_square]}") | |
| # Remove the pawn from the original square | |
| del self.piece_mapping[from_square] | |
| else: | |
| # For non-promotion moves, update the piece's position | |
| self.piece_mapping[to_square] = self.piece_mapping.pop(from_square) | |
| # Handle castling | |
| if self.board.is_castling(move): | |
| if to_square > from_square: # Kingside castling | |
| rook_from = chess.H1 if self.board.turn == chess.WHITE else chess.H8 | |
| rook_to = chess.F1 if self.board.turn == chess.WHITE else chess.F8 | |
| else: # Queenside castling | |
| rook_from = chess.A1 if self.board.turn == chess.WHITE else chess.A8 | |
| rook_to = chess.D1 if self.board.turn == chess.WHITE else chess.D8 | |
| # Move the rook | |
| self.piece_mapping[rook_to] = self.piece_mapping.pop(rook_from) | |
| logger.info(f"Castling: Rook moved from {chess.SQUARE_NAMES[rook_from]} to {chess.SQUARE_NAMES[rook_to]}") | |
| self.board.push(move) | |
| self.last_move = move | |
| self.decrease_all_pieces_empathy() | |
| return True | |
| return False | |
| def decrease_all_pieces_empathy(self, decrease_amount=EmpathyConfig.DECREASE_AMOUNT): | |
| for square, piece in self.piece_mapping.items(): | |
| if piece.empathy > EmpathyConfig.MIN_EMPATHY and piece.empathy < EmpathyConfig.HIGH_EMPATHY_THRESHOLD: | |
| piece.empathy = max(EmpathyConfig.MIN_EMPATHY, piece.empathy - decrease_amount) | |
| logger.info(f"Decreased empathy for {piece} at {chess.SQUARE_NAMES[square]} to {piece.empathy:.2f}") | |
| def is_white_to_move(self) -> bool: | |
| return self.board.turn == chess.WHITE | |
| def is_promotion_move(self, move: chess.Move) -> bool: | |
| piece = self.board.piece_at(move.from_square) | |
| return (piece is not None and piece.piece_type == chess.PAWN and | |
| ((piece.color == chess.WHITE and chess.square_rank(move.to_square) == 7) or | |
| (piece.color == chess.BLACK and chess.square_rank(move.to_square) == 0))) | |
| def is_checkmate(self) -> bool: | |
| return self.board.is_checkmate() | |
| def reset_game(self): | |
| self.board.reset() | |
| self.piece_mapping = self._initialize_piece_mapping() | |
| self.last_move = None | |
| logger.info("Game reset. New piece empathy values:") | |
| for square, piece in self.piece_mapping.items(): | |
| logger.info(f"Piece at {chess.SQUARE_NAMES[square]}: {piece} (Empathy: {piece.empathy:.2f})") | |
| class StreamConfig(BaseModel): | |
| max_width: int = Field(default=320, gt=0) | |
| fps_target: int = Field(default=40, gt=0) | |
| jpeg_quality: int = Field(default=40, ge=0, le=100) | |
| class RecordingCountdown(ft.ProgressBar): | |
| def __init__(self, seconds): | |
| super().__init__() | |
| self.seconds = seconds | |
| self.value = 1.0 | |
| def did_mount(self): | |
| self.running = True | |
| self.page.run_task(self.update_timer) | |
| def will_unmount(self): | |
| self.running = False | |
| async def update_timer(self): | |
| start_time = asyncio.get_event_loop().time() | |
| while self.seconds > 0 and self.running: | |
| elapsed_time = asyncio.get_event_loop().time() - start_time | |
| remaining_time = max(0, self.seconds - elapsed_time) | |
| self.value = remaining_time / self.seconds | |
| self.update() | |
| if remaining_time <= 0: | |
| break | |
| await asyncio.sleep(0.1) | |
| class ChessPresenter: | |
| def __init__(self, view: 'ChessView'): | |
| self.view = view | |
| self.model = ChessModel() | |
| self.dialogue_processor = LocalApiDialogueProcessor("https://danisz-chesscommanderservice.hf.space") # "https://danisz-chesscommanderservice.hf.space" | |
| self.selected_square = None | |
| self.pending_move = None | |
| self.valid_moves = [] | |
| self.audio_file_path = None | |
| self.video_file_path = None | |
| self.current_sprite = None | |
| self.last_move_sprite = None | |
| self.last_move_response = None | |
| self.elimination_mode = False | |
| self.recording_timer = None | |
| self.countdown_timer = None | |
| def toggle_video_recording(self): | |
| if self.view.is_camera_active(): | |
| self.stop_video_recording() | |
| self.view.show_processing_feedback() | |
| else: | |
| self.start_video_recording() | |
| def start_video_recording(self): | |
| self.view.start_camera_stream() | |
| self.view.show_video_container() | |
| self.view.update_video_button(is_recording=True) | |
| def stop_video_recording(self): | |
| self.view.stop_camera_stream() | |
| self.view.update_video_button(is_recording=False) | |
| self.view.hide_video_container() | |
| def process_audio_move_argument(self, audio_base64: str): | |
| piece_symbol, chess_piece = self.model.get_piece_at(self.pending_move.from_square) | |
| if chess_piece is not None: | |
| logger.info(f"Processing audio move for {chess_piece} (Empathy: {chess_piece.empathy:.2f})") | |
| dialogue_response = self.dialogue_processor.process_audio_input( | |
| chess_piece.piece_id, | |
| audio_base64, | |
| self.view.current_language, | |
| chess_piece.empathy | |
| ) | |
| logger.info(f"Dialogue Processor Response: {dialogue_response.response}") | |
| logger.info(f"Audio Transcription: {dialogue_response.transcription}") | |
| self.view.show_response(dialogue_response.response, piece_symbol, dialogue_response.transcription) | |
| self.view.update_piece_voice_attributes(dialogue_response.piece_speak_velocity) | |
| self.view.update_player_voice_attributes(dialogue_response.transcription, dialogue_response.player_speak_velocity) | |
| self.view.hide_processing_feedback() | |
| self.view.hide_move_argument_dialog() | |
| self.handle_move_decision(dialogue_response.decision, chess_piece) | |
| if dialogue_response.audio_data: | |
| mime_type = 'audio/wav' | |
| audio_url = f"data:{mime_type};base64,{dialogue_response.audio_data}" | |
| self.view.play_audio(audio_url) | |
| else: | |
| logger.error("Error: No valid piece selected for the move.") | |
| self.pending_move = None | |
| def process_text_move_argument(self, text_input: str): | |
| piece_symbol, chess_piece = self.model.get_piece_at(self.pending_move.from_square) | |
| if chess_piece is not None: | |
| dialogue_response = self.dialogue_processor.process_text_input( | |
| chess_piece.piece_id, | |
| text_input, | |
| self.view.current_language, | |
| chess_piece.empathy | |
| ) | |
| logger.info(f"Dialogue Processor Response: {dialogue_response.response}") | |
| logger.info(f"Transcription: {dialogue_response.transcription}") | |
| self.view.show_response(dialogue_response.response, piece_symbol, dialogue_response.transcription) | |
| self.view.update_player_voice_attributes(None) | |
| self.view.hide_processing_feedback() | |
| self.view.hide_move_argument_dialog() | |
| self.handle_move_decision(dialogue_response.decision, chess_piece) | |
| if dialogue_response.audio_data: | |
| mime_type = 'audio/wav' | |
| audio_url = f"data:{mime_type};base64,{dialogue_response.audio_data}" | |
| self.view.play_audio(audio_url) | |
| else: | |
| logger.error("Error: No valid piece selected for the move.") | |
| self.pending_move = None | |
| def adjust_empathy(self, chess_piece: ChessPiece, increase: bool): | |
| if increase: | |
| chess_piece.empathy = min(chess_piece.empathy + EmpathyConfig.STEP_AMOUNT, 1.0) | |
| else: | |
| chess_piece.empathy = max(chess_piece.empathy - EmpathyConfig.STEP_AMOUNT, 0.0) | |
| if self.model.is_king_empathy_zero_or_below(): | |
| loser = "White" if self.model.is_white_to_move() else "Black" | |
| logger.info(f"{loser} loses because the king's empathy is zero or below.") | |
| self.view.show_king_empathy_loss_dialog(loser) | |
| def handle_move_decision(self, decision: bool, chess_piece: ChessPiece): | |
| if decision: | |
| self.make_move(self.pending_move) | |
| self.adjust_empathy(chess_piece, True) | |
| else: | |
| logger.info("Move rejected based on dialogue processor decision.") | |
| self.adjust_empathy(chess_piece, False) | |
| self.check_and_remove_piece(self.pending_move.from_square) | |
| # Change the turn from white to black or vice versa | |
| self.model.board.turn = not self.model.board.turn | |
| self.view.switch_turn() | |
| # Decrease empathy for all pieces after each turn | |
| self.model.decrease_all_pieces_empathy() | |
| def check_and_remove_piece(self, square: chess.Square): | |
| piece = self.model.piece_mapping.get(square) | |
| if piece and piece.empathy <= 0: | |
| self.model.remove_piece(square) | |
| self.update_board() | |
| def make_move(self, move: chess.Move): | |
| if self.model.make_move(move): | |
| from_square = move.from_square | |
| to_square = move.to_square | |
| moving_piece = self.model.piece_mapping.get(to_square) | |
| if moving_piece: | |
| if moving_piece.empathy >= EmpathyConfig.HIGH_EMPATHY_THRESHOLD: | |
| moving_piece.empathy = EmpathyConfig.FREE_MOVE_EMPATHY | |
| logger.info(f"Piece at {chess.SQUARE_NAMES[to_square]} made a free move. Empathy reset to {EmpathyConfig.FREE_MOVE_EMPATHY}") | |
| was_promotion = move.promotion is not None | |
| self.update_board(from_square, to_square, was_promotion) | |
| if self.model.is_checkmate(): | |
| winner = "White" if not self.model.is_white_to_move() else "Black" | |
| logger.info(f"Checkmate! {winner} wins.") | |
| self.view.show_checkmate_dialog(winner) | |
| else: | |
| self.view.switch_turn() | |
| self.on_turn_begin() | |
| else: | |
| logger.warning(f"Invalid move: {move}") | |
| def on_text_submit(self, text: str): | |
| self.process_text_move_argument(text) | |
| def on_view_initialized(self): | |
| logger.info("Chess game initialized. Current board state:") | |
| self.update_board() | |
| for square, piece in self.model.piece_mapping.items(): | |
| logger.info(f"Piece at {chess.SQUARE_NAMES[square]}: {piece} (Empathy: {piece.empathy:.2f})") | |
| self.on_turn_begin() # Make sure this is called at the start of the game | |
| def update_board(self, from_square=None, to_square=None, was_promotion=False): | |
| board_state = self._get_board_state() | |
| last_move_square = self.model.last_move.to_square if self.model.last_move else None | |
| self.view.draw_board(board_state, self.selected_square, self.valid_moves, last_move_square, from_square, to_square, was_promotion) | |
| def _get_board_state(self): | |
| return [[self.model.get_piece_at(chess.square(col, 7 - row))[0] for col in range(8)] for row in range(8)] | |
| def on_square_click(self, row, col): | |
| square = chess.square(col, 7 - row) | |
| piece, piece_enum = self.model.get_piece_at(square) | |
| is_valid_piece = (piece != " " and | |
| ((self.model.is_white_to_move() and piece.isupper()) or | |
| (not self.model.is_white_to_move() and piece.islower()))) | |
| print(f"Click: Piece={piece}, self.model.is_white_to_move()={self.model.is_white_to_move()}") | |
| if is_valid_piece: | |
| self.selected_square = square | |
| self.valid_moves = [move.to_square for move in self.model.board.legal_moves if move.from_square == square] | |
| logger.info(f"Selected piece at {chess.SQUARE_NAMES[square]}: {piece}") | |
| self.current_sprite = piece | |
| self.view.show_empathy_bar(piece_enum.empathy) | |
| elif self.selected_square is not None: | |
| move = chess.Move(self.selected_square, square) | |
| if self.model.is_promotion_move(move): | |
| move = chess.Move(self.selected_square, square, promotion=chess.QUEEN) | |
| if self.model.board.is_legal(move): | |
| _, moving_piece = self.model.get_piece_at(self.selected_square) | |
| if moving_piece and moving_piece.empathy >= 0.9: | |
| self.make_move(move) | |
| else: | |
| self.pending_move = move | |
| self.view.show_move_argument_dialog() | |
| else: | |
| logger.info(f"Invalid move: {move}") | |
| self.selected_square = None | |
| self.valid_moves = [] | |
| self.update_board() | |
| def restart_game(self): | |
| self.model.reset_game() | |
| self.selected_square = None | |
| self.pending_move = None | |
| self.valid_moves = [] | |
| self.update_board() | |
| self.view.hide_checkmate_dialog() | |
| self.view.reset_turn() | |
| self.view.hide_sprite_and_dialogue() | |
| self.on_turn_begin() # Make sure this is called when the game is restarted | |
| def on_turn_begin(self): | |
| print("on_turn_begin: Method called") | |
| logger.info("Checking piece empathy levels...") | |
| self.view.hide_player_transcription() | |
| all_high_empathy = self.model.are_all_pieces_empathy_above(EmpathyConfig.HIGH_EMPATHY_THRESHOLD) | |
| print(f"All pieces have high empathy: {all_high_empathy}") | |
| if all_high_empathy: | |
| self.view.show_eliminate_piece_popup() | |
| else: | |
| print("Not all pieces have high empathy. Continuing normal gameplay.") | |
| # Log the empathy of all pieces | |
| for square, piece in self.model.piece_mapping.items(): | |
| print(f"Piece at {chess.SQUARE_NAMES[square]}: {piece} (Empathy: {piece.empathy:.2f})") | |
| def start_piece_elimination(self): | |
| self.elimination_mode = True | |
| self.update_board() # Use the presenter's method to update the board | |
| def eliminate_rival_piece(self, square: chess.Square): | |
| if not self.elimination_mode: | |
| return | |
| piece = self.model.board.piece_at(square) | |
| if piece and piece.color != self.model.board.turn: | |
| # Ensure that the piece to be eliminated is not a king | |
| if piece.piece_type == chess.KING: | |
| logger.info("Cannot eliminate the king piece.") | |
| self.view.show_error_message("You cannot eliminate the king piece.") | |
| return | |
| self.model.remove_piece(square) | |
| self.update_board() | |
| self.elimination_mode = False | |
| # Switch turn after elimination | |
| self.model.board.turn = not self.model.board.turn | |
| self.view.switch_turn() | |
| self.model.decrease_all_pieces_empathy() | |
| # Reset selection and valid moves | |
| self.selected_square = None | |
| self.valid_moves = [] | |
| # Call on_turn_begin to set up for the next turn | |
| self.on_turn_begin() | |
| else: | |
| logger.info("Invalid piece selection for elimination.") | |
| self.view.show_error_message("Invalid piece selection for elimination.") | |
| class ChessView(ft.Container): | |
| def __init__(self, input_page): | |
| super().__init__() | |
| self.my_page = input_page | |
| self.presenter = ChessPresenter(self) | |
| self.sprite_visible = False | |
| self.elimination_mode = False | |
| self.processing_indicator = ft.ProgressRing(visible=False) | |
| self.processing_text = ft.Text("Processing...", visible=False) | |
| self.current_language = "en" | |
| self.en_button = ft.ElevatedButton("EN", on_click=self.set_english) | |
| self.es_button = ft.ElevatedButton("ES", on_click=self.set_spanish) | |
| self.piece_speak_velocity_bar = self.speak_velocity_bar() | |
| self.player_speak_velocity_bar = self.speak_velocity_bar() | |
| self.empathy_bar_visible = True | |
| self.language_selector = ft.Row( | |
| [self.en_button, self.es_button], | |
| alignment=ft.MainAxisAlignment.CENTER | |
| ) | |
| self.eliminate_piece_dialog = ft.AlertDialog( | |
| title=ft.Text("Eliminate a Rival Piece"), | |
| content=ft.Text("Do you want to eliminate a rival piece?"), | |
| actions=[ | |
| ft.ElevatedButton("Accept", on_click=self.accept_elimination), | |
| ft.ElevatedButton("Cancel", on_click=self.hide_eliminate_piece_popup), | |
| ], | |
| actions_alignment=ft.MainAxisAlignment.END, | |
| ) | |
| self.board_size = 352 | |
| self.square_size = self.board_size // 8 | |
| self.selected_color = ft.colors.YELLOW_100 | |
| self.valid_move_color = ft.colors.AMBER_200 | |
| self.last_move_color = ft.colors.LIGHT_GREEN_200 | |
| self.empathy_bar = ft.ProgressBar( | |
| width=self.board_size, | |
| height=10, | |
| color=ft.colors.BLUE, | |
| bgcolor=ft.colors.GREY_300, | |
| value=0.5, | |
| ) | |
| self.empathy_text = ft.Text("Empathy", size=14, color=ft.colors.WHITE) | |
| self.empathy_container = ft.Container( | |
| content=ft.Column([ | |
| self.empathy_text, | |
| self.empathy_bar | |
| ], spacing=5, alignment=ft.MainAxisAlignment.CENTER), | |
| width=self.board_size, | |
| padding=10, | |
| bgcolor=ft.colors.BLACK, | |
| border_radius=ft.border_radius.all(10), | |
| visible=False, # Start hidden | |
| ) | |
| self.countdown = RecordingCountdown(10) | |
| self.piece_images = { | |
| 'P': '/images/sprites/pawn.png', | |
| 'R': '/images/sprites/rook.png', | |
| 'N': '/images/sprites/knight.png', | |
| 'B': '/images/sprites/bishop.png', | |
| 'Q': '/images/sprites/queen.png', | |
| 'K': '/images/sprites/king.png', | |
| } | |
| self.chessboardpiece_images = { | |
| 'P': '/images/imgs-80px/white_pawn.gif', | |
| 'R': '/images/imgs-80px/white_rook.gif', | |
| 'N': '/images/imgs-80px/white_knight.gif', | |
| 'B': '/images/imgs-80px/white_bishop.gif', | |
| 'Q': '/images/imgs-80px/white_queen.gif', | |
| 'K': '/images/imgs-80px/white_king.gif', | |
| 'p': '/images/imgs-80px/black_pawn.gif', | |
| 'r': '/images/imgs-80px/black_rook.gif', | |
| 'n': '/images/imgs-80px/black_knight.gif', | |
| 'b': '/images/imgs-80px/black_bishop.gif', | |
| 'q': '/images/imgs-80px/black_queen.gif', | |
| 'k': '/images/imgs-80px/black_king.gif', | |
| } | |
| self.audio_recorder = ft.AudioRecorder( | |
| audio_encoder=ft.AudioEncoder.OPUS, | |
| on_state_changed=self.on_recorder_state_changed | |
| ) | |
| self.move_sound = ft.Audio(src="/sounds/move.wav") | |
| self.capture_sound = ft.Audio(src="/sounds/capture.wav") | |
| # Add audio playback component | |
| self.audio_player = ft.Audio( | |
| src="X", # Dummy non-empty string for Flutter dart internals | |
| autoplay=False, | |
| on_loaded=lambda _: print("Loaded"), | |
| on_duration_changed=lambda e: print("Duration changed:", e.data), | |
| on_position_changed=lambda e: print("Position changed:", e.data), | |
| on_state_changed=lambda e: clean_audio_player(e), | |
| on_seek_complete=lambda _: print("\nSeek complete"), | |
| ) | |
| def clean_audio_player(e): | |
| print("State changed:", e.data) | |
| if e.data == "completed": | |
| self.audio_player.autoplay = False | |
| self.audio_player.update() | |
| self.camera_image = CameraImage(width=320, height=240) | |
| self.video_container = ft.Container( | |
| content=self.camera_image, | |
| visible=False, | |
| ) | |
| # Adjust the dialogue size | |
| self.dialogue_height = 800 # Increased height | |
| self.dialogue_width = 840 # Increased width | |
| # Adjust the sprite size | |
| self.sprite_placeholder = ft.Image( | |
| src="", | |
| width=800, | |
| height=800, | |
| fit=ft.ImageFit.CONTAIN | |
| ) | |
| # Adjust the response text and dialogue | |
| self.response_text = ft.Text("", size=18, color=ft.colors.WHITE) | |
| self.piece_dialogue = ft.Container( | |
| content=self.response_text, | |
| bgcolor=ft.colors.BLACK, # Semi-transparent background | |
| padding=10, | |
| width=360, | |
| height=100, | |
| border_radius=ft.border_radius.all(10), | |
| border=ft.border.all(color=ft.colors.YELLOW, width=2), | |
| ) | |
| # Define response_title_container | |
| self.response_title_container = ft.Container( | |
| content=ft.Text("Response", color=ft.colors.WHITE, size=14), | |
| bgcolor=ft.colors.BLUE, | |
| padding=5, | |
| border_radius=ft.border_radius.all(10), | |
| ) | |
| # Adjust the response container | |
| self.response_container = ft.Stack( | |
| controls=[ | |
| self.piece_dialogue, | |
| ft.Container(self.response_title_container, alignment=ft.alignment.bottom_left, padding=ft.Padding(left=20, top=-10, right=0, bottom=0)), | |
| ], | |
| width=320, # Increased width | |
| ) | |
| # Adjust the sprite_and_dialogue_container | |
| self.response_layout = ft.Container( | |
| content=None, # We'll set this in show_sprite_and_dialogue | |
| alignment=ft.alignment.bottom_center, | |
| bottom=20, # Distance from bottom of screen | |
| left=0, | |
| right=0, | |
| ) | |
| self.chessboard = ft.Container( | |
| width=self.board_size, | |
| height=self.board_size, | |
| content=ft.Column(spacing=0), | |
| bgcolor=None, | |
| ) | |
| # Add these lines to create the transcription elements | |
| self.transcription_text = ft.Text("", size=14, color=ft.colors.WHITE) | |
| self.transcription_container = ft.Container( | |
| content=self.transcription_text, | |
| bgcolor=ft.colors.BLACK54, | |
| padding=10, | |
| border_radius=ft.border_radius.all(10), | |
| visible=False | |
| ) | |
| self.built_control = self.build_ui() | |
| self.setup_ui(self.built_control) | |
| self.current_turn = chess.WHITE | |
| def initialize_view(self): | |
| self.update_turn_indicator() | |
| self.presenter.on_view_initialized() | |
| def play_audio(self, file_path: str): | |
| print("Playing audio") | |
| self.audio_player.src = file_path | |
| self.audio_player.autoplay = True | |
| self.audio_player.update() | |
| def set_english(self, e): | |
| self.current_language = "en" | |
| self.update_language() | |
| def set_spanish(self, e): | |
| self.current_language = "es" | |
| self.update_language() | |
| def update_language(self): | |
| self.update_ui_texts() | |
| self.update() | |
| def update_ui_texts(self): | |
| self.checkmate_dialog.title.value = self.get_localized_string("Checkmate!") | |
| self.submit_button.text = self.get_localized_string("Submit") | |
| self.move_argument_dialog.title.value = self.get_localized_string("Describe your move") | |
| self.text_input.label = self.get_localized_string("Enter your move description") | |
| self.record_button.tooltip = self.get_localized_string("Record move argument") | |
| self.move_argument_text.value = self.get_localized_string("Type your move or use the microphone to record") | |
| self.processing_text.value = self.get_localized_string("Processing...") | |
| self.update_turn_indicator() | |
| def get_localized_string(self, key): | |
| if self.current_language == "es": | |
| return spanish_translations.get(key, key) | |
| return key | |
| def show_processing_feedback(self): | |
| self.move_argument_text.visible = False | |
| self.text_input.visible = False | |
| self.submit_button.visible = False | |
| self.record_button.visible = False | |
| self.video_record_button.visible = False | |
| self.processing_indicator.visible = True | |
| self.processing_text.visible = True | |
| self.update() | |
| def hide_processing_feedback(self): | |
| self.move_argument_text.visible = True | |
| self.text_input.visible = True | |
| self.submit_button.visible = True | |
| self.record_button.visible = True | |
| self.video_record_button.visible = True | |
| self.processing_indicator.visible = False | |
| self.processing_text.visible = False | |
| self.update() | |
| def setup_ui(self, control: ft.Control): | |
| self.width = 360 | |
| self.height = 800 | |
| self.bgcolor = ft.colors.RED # Debugging | |
| self.alignment = ft.alignment.center | |
| self.expand = False | |
| background_image_container = ft.Container( | |
| image_src='/images/chess_background.jpg', | |
| image_fit=ft.ImageFit.COVER, | |
| alignment=ft.alignment.center, | |
| bgcolor=ft.colors.BLUE, # Debugging | |
| expand=True | |
| ) | |
| self.content = ft.Stack( | |
| controls=[ | |
| background_image_container, | |
| control | |
| ], | |
| alignment=ft.alignment.center | |
| ) | |
| def did_mount(self): | |
| self.presenter.on_view_initialized() | |
| def show_sprite(self, piece: str): | |
| piece_image_path = self.piece_images.get(piece.upper(), "") | |
| self.sprite_placeholder.src = piece_image_path | |
| self.hide_sprite_and_dialogue() | |
| self.update() | |
| def draw_board(self, board, selected_square, valid_moves, last_move_square, from_square=None, to_square=None, was_promotion=False): | |
| rows = [] | |
| for row in range(8): | |
| board_row = ft.Row(spacing=0) | |
| for col in range(8): | |
| square = chess.square(col, 7 - row) | |
| if selected_square is not None and square == selected_square: | |
| color = self.selected_color | |
| elif square in valid_moves: | |
| color = self.valid_move_color | |
| elif square == last_move_square: | |
| color = self.last_move_color | |
| else: | |
| color = ft.colors.ORANGE_100 if (row + col) % 2 == 0 else ft.colors.ORANGE_300 | |
| piece, chess_piece = self.presenter.model.get_piece_at(square) | |
| if was_promotion and square == from_square: | |
| piece = " " | |
| chess_piece = None | |
| high_empathy = chess_piece is not None and chess_piece.empathy >= 0.9 | |
| square_container = self.create_square(row, col, color, piece, chess_piece, high_empathy) | |
| board_row.controls.append(square_container) | |
| rows.append(board_row) | |
| self.chessboard.content = ft.Column(rows, spacing=0) | |
| self.update() | |
| def create_square(self, row, col, color, piece, chess_piece, high_empathy=False): | |
| piece_image_path = "" | |
| if high_empathy: | |
| color = ft.colors.WHITE | |
| if chess_piece: | |
| if chess_piece.piece_type == PieceType.QUEEN and piece.lower() == 'q': | |
| piece_image_path = self.chessboardpiece_images.get('Q' if chess_piece.color == chess.WHITE else 'q', "") | |
| else: | |
| piece_image_path = self.chessboardpiece_images.get(piece, "") | |
| piece_image = None | |
| if piece_image_path and not (chess_piece and chess_piece.is_disappeared()): | |
| # Add a scaling factor for the knight | |
| scale = 1.2 if piece.lower() == 'n' else 1.0 | |
| piece_image = ft.Image( | |
| src=piece_image_path, | |
| width=self.square_size * scale, | |
| height=self.square_size * scale, | |
| fit=ft.ImageFit.CONTAIN | |
| ) | |
| def on_square_click(e): | |
| self.on_square_click(row, col) | |
| return ft.Container( | |
| width=self.square_size, | |
| height=self.square_size, | |
| bgcolor=color, | |
| border=ft.border.all(1, ft.colors.BLACK), | |
| border_radius=ft.border_radius.all(0), | |
| alignment=ft.alignment.center, | |
| data=f"{row},{col}", | |
| on_click=on_square_click, | |
| content=piece_image, | |
| ) | |
| def show_eliminate_piece_popup(self): | |
| self.elimination_mode = True | |
| self.eliminate_piece_dialog.open = True | |
| self.update() | |
| def hide_eliminate_piece_popup(self, _=None): | |
| self.elimination_mode = False | |
| self.eliminate_piece_dialog.open = False | |
| self.update() | |
| def exit_elimination_mode(self): | |
| self.elimination_mode = False | |
| self.update() | |
| def show_king_empathy_loss_dialog(self, loser: str): | |
| self.my_page.run_task(self._show_king_empathy_loss_dialog, loser) | |
| async def _show_king_empathy_loss_dialog(self, loser: str): | |
| await asyncio.sleep(1) | |
| loser_text = self.get_localized_string(loser) | |
| lose_reason = self.get_localized_string('loses because the king\'s empathy is zero or below!') | |
| message = f"{loser_text} {lose_reason}" | |
| self.king_empathy_loss_dialog.content.value = message | |
| self.king_empathy_loss_dialog.open = True | |
| self.update() | |
| def update_video_button(self, is_recording: bool): | |
| self.video_record_button.icon = ft.icons.STOP if is_recording else ft.icons.VIDEO_CALL | |
| self.video_record_button.tooltip = "Stop recording" if is_recording else "Record video move argument" | |
| self.update() | |
| def show_response(self, response, piece: str, transcription: str): | |
| self.response_text.value = response | |
| self.show_sprite_and_dialogue(piece) | |
| if transcription: | |
| self.transcription_text.value = transcription | |
| self.transcription_container.visible = True | |
| self.update() | |
| def hide_move_argument_dialog(self): | |
| self.move_argument_dialog.open = False | |
| self.update() | |
| def show_move_argument_dialog(self): | |
| self.move_argument_dialog.open = True | |
| self.text_input.focus() | |
| self.update() | |
| def show_checkmate_dialog(self, winner: str): | |
| winner_text = self.get_localized_string(winner) | |
| message = f"{winner_text} {self.get_localized_string('wins by checkmate!')}" | |
| self.checkmate_dialog.content.value = message | |
| self.checkmate_dialog.open = True | |
| self.update() | |
| def hide_checkmate_dialog(self): | |
| self.checkmate_dialog.open = False | |
| self.update() | |
| def restart_game(self, _=None): | |
| self.checkmate_dialog.open = False | |
| self.king_empathy_loss_dialog.open = False | |
| self.update() | |
| self.presenter.restart_game() | |
| def start_countdown(self): | |
| self.countdown.seconds = 10 | |
| self.countdown.value = 1.0 | |
| self.countdown.running = True | |
| self.countdown.did_mount() | |
| self.update() | |
| def stop_countdown(self): | |
| self.countdown.running = False | |
| self.countdown.will_unmount() | |
| self.update() | |
| async def toggle_recording(self, e): | |
| if await self.audio_recorder.is_recording_async(): | |
| await self.stop_recording() | |
| else: | |
| await self.start_recording() | |
| async def start_recording(self): | |
| import tempfile | |
| self.record_button.disabled = True | |
| self.update() | |
| temp_recording_file = tempfile.NamedTemporaryFile(delete=False, suffix=".wav") | |
| print(f"Starting recording to {temp_recording_file.name}") | |
| started = await self.audio_recorder.invoke_method_async( | |
| "start_recording", | |
| {"outputPath": temp_recording_file.name}, | |
| wait_for_result=True, | |
| wait_timeout=10 | |
| ) | |
| if started == "true": | |
| print("Recording started successfully") | |
| self.record_button.icon = ft.icons.STOP | |
| self.record_button.tooltip = "Stop recording" | |
| self.record_button.disabled = False | |
| self.start_countdown() | |
| else: | |
| print("Failed to start recording") | |
| self.record_button.disabled = False | |
| self.update() | |
| async def stop_recording(self): | |
| self.stop_countdown() | |
| self.show_processing_feedback() | |
| self.update() | |
| audio_uri = await self.audio_recorder.stop_recording_async() | |
| print(f"Recording stopped. URI: {audio_uri}") | |
| audio_base64 = await self.get_base64_audio(audio_uri) | |
| print("Base64 representation (first 50 chars):", audio_base64[:50]) | |
| self.record_button.icon = ft.icons.MIC | |
| self.record_button.tooltip = "Record move argument" | |
| self.presenter.process_audio_move_argument(audio_base64) | |
| self.update() | |
| async def get_base64_audio(self, audio_uri: str) -> str: | |
| import sys | |
| if sys.platform == "emscripten": # Pyodide environment | |
| from pyodide.code import run_js | |
| # Get the base64 representation of the blob | |
| js_code = f""" | |
| async function getBase64(uri) {{ | |
| const response = await fetch(uri); | |
| const blob = await response.blob(); | |
| return new Promise((resolve, reject) => {{ | |
| const reader = new FileReader(); | |
| reader.onloadend = () => resolve(reader.result); | |
| reader.onerror = reject; | |
| reader.readAsDataURL(blob); | |
| }}); | |
| }} | |
| getBase64("{audio_uri}") | |
| """ | |
| base64_data = await run_js(js_code) | |
| return base64_data | |
| else: | |
| # For non-Pyodide environments, assume audio_uri is already a base64 string | |
| return await get_base64_audio(self.my_page, audio_uri) | |
| def on_recorder_state_changed(self, e): | |
| if e.data == "stopped": | |
| pass # We don't need to do anything here anymore | |
| def on_text_submit(self, e: ft.ControlEvent): | |
| self.presenter.on_text_submit(self.text_input.value) | |
| self.text_input.value = "" | |
| self.update() | |
| def switch_turn(self): | |
| self.current_turn = not self.current_turn | |
| self.update_turn_indicator() | |
| def update_turn_indicator(self): | |
| turn_text = "White's turn" if self.current_turn == chess.WHITE else "Black's turn" | |
| self.turn_indicator.value = self.get_localized_string(turn_text) | |
| self.update() | |
| def reset_turn(self): | |
| self.current_turn = chess.WHITE | |
| self.update_turn_indicator() | |
| def on_square_click(self, row, col): | |
| square = chess.square(col, 7 - row) | |
| piece, chess_piece = self.presenter.model.get_piece_at(square) | |
| if self.elimination_mode: | |
| self.presenter.eliminate_rival_piece(square) | |
| return | |
| if chess_piece and chess_piece.color == self.current_turn: | |
| # Show empathy bar when a valid piece is clicked | |
| print(f"Valid piece selected: {chess_piece}, empathy: {chess_piece.empathy}") | |
| self.show_empathy_bar(chess_piece.empathy) | |
| else: | |
| print("No valid piece selected") | |
| self.hide_empathy_bar() | |
| # Delegate handling the square click to the presenter | |
| self.presenter.on_square_click(row, col) | |
| def show_empathy_bar(self, empathy): | |
| print(f"Showing empathy bar with value: {empathy}") | |
| # Ensure the empathy value is between 0 and 1 | |
| self.empathy_bar.value = max(0.0, min(empathy, 1.0)) | |
| # Make the empathy container visible | |
| self.empathy_container.visible = True | |
| self.update() | |
| def hide_empathy_bar(self): | |
| print("Hiding empathy bar") | |
| self.empathy_container.visible = False | |
| self.update() | |
| def show_sprite_and_dialogue(self, piece: str): | |
| piece_image_path = self.piece_images.get(piece.upper(), "") | |
| self.sprite_placeholder.src = piece_image_path | |
| # Create a stack with the sprite and dialogue overlapped | |
| sprite_and_dialogue = ft.Stack([ | |
| self.sprite_placeholder, | |
| ft.Container( | |
| content=self.piece_dialogue, | |
| alignment=ft.alignment.bottom_center, | |
| bottom=20, # Adjust this value to position the dialogue | |
| left=0, | |
| right=0, | |
| ) | |
| ]) | |
| # Update the response_layout with the new content | |
| self.response_layout.content = sprite_and_dialogue | |
| self.response_layout.visible = True | |
| self.update() | |
| def hide_sprite_and_dialogue(self): | |
| self.response_layout.visible = False | |
| self.update() | |
| def hide_player_transcription(self): | |
| self.transcription_container.visible = False | |
| self.update() | |
| def update_piece_voice_attributes(self, piece_speak_velocity: str): | |
| self.set_velocity(self.piece_speak_velocity_bar, piece_speak_velocity) | |
| self.update() | |
| def update_player_voice_attributes(self, transcription: Optional[str] = None, player_speak_velocity: Optional[str] = None): | |
| if transcription and player_speak_velocity: | |
| self.transcription_text.value = transcription | |
| self.set_velocity(self.player_speak_velocity_bar, player_speak_velocity) | |
| self.transcription_container.visible = True | |
| else: | |
| # Don't clear the transcription text or hide the container | |
| self.set_velocity(self.player_speak_velocity_bar, "low") # Reset to low | |
| self.update() | |
| def accept_elimination(self, _=None): | |
| self.hide_eliminate_piece_popup() | |
| self.presenter.start_piece_elimination() | |
| def build_ui(self): | |
| self.checkmate_dialog = ft.AlertDialog( | |
| modal=True, | |
| title=ft.Text("Checkmate!"), | |
| content=ft.Text(""), | |
| actions=[ | |
| ft.TextButton("Restart", on_click=self.restart_game), | |
| ], | |
| actions_alignment=ft.MainAxisAlignment.END, | |
| ) | |
| self.king_empathy_loss_dialog = ft.AlertDialog( | |
| modal=True, | |
| title=ft.Text("Game Over"), | |
| content=ft.Text(""), | |
| actions=[ | |
| ft.TextButton("Restart", on_click=self.restart_game), | |
| ], | |
| actions_alignment=ft.MainAxisAlignment.END, | |
| ) | |
| self.record_button = ft.IconButton( | |
| icon=ft.icons.MIC, | |
| on_click=self.toggle_recording, | |
| tooltip="Record move argument" | |
| ) | |
| self.video_record_button = ft.IconButton( | |
| icon=ft.icons.VIDEO_CALL, | |
| on_click=lambda _: self.presenter.toggle_video_recording(), | |
| tooltip="Record video move argument" | |
| ) | |
| self.text_input = ft.TextField( | |
| label="Enter your move description", | |
| on_submit=self.on_text_submit, | |
| width=200 | |
| ) | |
| self.submit_button = ft.ElevatedButton("Submit", on_click=self.on_text_submit) | |
| self.move_argument_text = ft.Text("Type your move or use the microphone to record") | |
| self.move_argument_dialog = ft.AlertDialog( | |
| title=ft.Text("Describe your move"), | |
| content=ft.Column([ | |
| self.countdown, | |
| self.move_argument_text, | |
| ft.Row([self.text_input]), | |
| self.submit_button, | |
| ft.Row([ | |
| self.record_button, | |
| self.video_record_button, | |
| ]), | |
| ft.Row([ | |
| self.processing_indicator, | |
| self.processing_text | |
| ], alignment=ft.MainAxisAlignment.CENTER), | |
| self.video_container, # Use the video_container instead of video_placeholder | |
| ], tight=True), | |
| ) | |
| self.turn_indicator = ft.Text("White's turn", size=20, color=ft.colors.WHITE) | |
| # Create separate containers for turn indicator and empathy bar | |
| turn_container = ft.Container( | |
| content=self.turn_indicator, | |
| margin=ft.margin.only(bottom=5) | |
| ) | |
| # Place turn container, empathy container, and chessboard in a column | |
| board_column = ft.Column([ | |
| turn_container, | |
| self.empathy_container, | |
| ft.Container(ft.Stack([self.chessboard, self.move_argument_dialog]), bgcolor='blue') | |
| ], alignment=ft.MainAxisAlignment.CENTER, spacing=10) # Added spacing between elements | |
| main_layout = ft.Container( | |
| content=board_column, | |
| margin=ft.margin.only(top=20) | |
| ) | |
| full_layout = ft.Stack([ | |
| ft.Column(controls=[ | |
| ft.Container(ft.Column(controls=[main_layout])), | |
| self.checkmate_dialog, | |
| self.eliminate_piece_dialog, | |
| self.king_empathy_loss_dialog, | |
| self.transcription_container # Add this line | |
| ]), | |
| self.response_layout, # Now part of the main stack | |
| ]) | |
| return ft.Container( | |
| content=full_layout, | |
| width=360, | |
| height=800, | |
| on_click=self.on_screen_click | |
| ) | |
| def speak_velocity_bar(self, width=150, height=15): | |
| segments = [ | |
| ft.Container(width=width/3, height=height, bgcolor=ft.colors.GREEN), | |
| ft.Container(width=width/3, height=height, bgcolor=ft.colors.GREY_300), | |
| ft.Container(width=width/3, height=height, bgcolor=ft.colors.GREY_300) | |
| ] | |
| return ft.Row(segments, spacing=2) | |
| def set_velocity(self, bar, velocity: str): | |
| if velocity: | |
| colors = { | |
| "low": [ft.colors.GREEN, ft.colors.GREY_300, ft.colors.GREY_300], | |
| "medium": [ft.colors.GREEN, ft.colors.YELLOW, ft.colors.GREY_300], | |
| "high": [ft.colors.GREEN, ft.colors.YELLOW, ft.colors.RED] | |
| } | |
| selected_colors = colors.get(velocity.lower(), colors["low"]) | |
| for i, color in enumerate(selected_colors): | |
| bar.controls[i].bgcolor = color | |
| self.update() | |
| def is_camera_active(self): | |
| return self.camera_image.stream_active | |
| def start_camera_stream(self): | |
| self.my_page.run_task(self.camera_image.start_camera_stream_async) | |
| def stop_camera_stream(self): | |
| self.my_page.run_task(self.camera_image.stop_camera_stream_async) | |
| def show_video_container(self): | |
| self.video_container.visible = True | |
| self.update() | |
| def hide_video_container(self): | |
| self.video_container.visible = False | |
| self.update() | |
| def on_screen_click(self, e): | |
| if self.empathy_bar_visible: | |
| print("Screen clicked, hiding empathy bar") | |
| self.hide_empathy_bar() | |
| self.hide_sprite_and_dialogue() | |
| # Clear the self.transcription_text.value = "" | |
| self.update() | |
| async def main(page: ft.Page): | |
| import sys | |
| print("SYS PLATFORM!:") | |
| print(sys.platform) | |
| if sys.platform == "emscripten": # check if run in Pyodide environment | |
| print("WE ARE RUNNING PYODIDE!") | |
| from pyodide.code import run_js | |
| # Run arbitrary JavaScript code | |
| result = run_js('Math.random() * 100') | |
| # The result is returned to Python | |
| print("Random number from JS:", result) | |
| page.title = "Multiplayer Chess" | |
| page.window_width = 360 | |
| page.window_height = 800 | |
| page.window_resizable = True | |
| page.padding = 0 | |
| page.horizontal_alignment = ft.CrossAxisAlignment.CENTER | |
| page.vertical_alignment = ft.MainAxisAlignment.CENTER | |
| chess_view = ChessView(page) | |
| page.overlay.append(chess_view.audio_recorder) | |
| page.add(chess_view) | |
| chess_view.initialize_view() # Call this after adding to the page | |
| page.overlay.extend([chess_view.move_sound, chess_view.capture_sound, chess_view.audio_player]) | |
| page.update() # Needed to update overlay | |
| if __name__ == "__main__": | |
| app = ft.app(target=main) |