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)