如何将移动计算移动到 Tkinter 中的单独线程?

问题描述 投票:0回答:1

在 tkinter 上跳棋的当前实现中,尽管它似乎已将棋步的计算移至单独的线程,但当计算机计算最佳棋步(具有平均计算深度)时,界面速度会变慢。如果不尝试将其放入单独的线程中,界面根本不会响应,但会重现动画 关键方法如下:

def mouse_down(self, event: Event):
        '''Событие нажатия мышки'''
        x, y = (event.x - BOARD_BORDER) // CELL_SIZE, (event.y - BOARD_BORDER) // CELL_SIZE
        # Если точка не внутри поля
        if not (self.__field.is_within(x, y)): return
        if MULTIPLAYER:
        ...
        else:
            if not (self.__player_turn): return

            if (PLAYER_SIDE == SideType.WHITE):
                player_checkers = WHITE_CHECKERS
            elif (PLAYER_SIDE == SideType.BLACK):
                player_checkers = BLACK_CHECKERS
            else:
                return

            # Если нажатие по шашке игрока, то выбрать её
            if (self.__field.type_at(x, y) in player_checkers):
                self.__selected_cell = Point(x, y)
                self.__draw()
            elif (self.__player_turn):
                move = Move(self.__selected_cell.x, self.__selected_cell.y, x, y)
                # Если нажатие по ячейке, на которую можно походить
                if (move in self.__get_moves_list(PLAYER_SIDE)):
                    self.__handle_player_turn(move)
                    # Если не ход игрока, то ход противника
                    if not (self.__player_turn):
                        self.test_field = self.__field
                        self.handle_enemy_turn_async()
                ...
    def handle_enemy_turn_async(game_instance):
        def run():
            game_instance.__handle_enemy_turn()

        threading.Thread(target=run).start()

    def __handle_enemy_turn(self):
        '''Обработка хода противника (компьютера)'''
        self.__player_turn = False
        self.calc = True
        optimal_moves_list = self.__predict_optimal_moves(SideType.opposite(PLAYER_SIDE))
        self.calc = False
        for move in optimal_moves_list:
            self.__handle_move(self.__field, move)
        self.__player_turn = True
        self.__check_for_game_over()

    def __predict_optimal_moves(self, side: SideType) -> list[Move]:
        '''Предсказать оптимальный ход'''
        best_result = 0
        optimal_moves = []
        self.test_field = Field.copy(self.__field)
        predicted_moves_list = self.__get_predicted_moves_list(side)
        if (predicted_moves_list):
            field_copy = Field.copy(self.__field)
            self.test_field = Field.copy(self.__field)
            field_copy_test = Field.copy(self.test_field)
            for moves in predicted_moves_list:
                for move in moves:
                    self.__handle_move(self.test_field, move, draw=False)
                try:
                    if (side == SideType.WHITE):
                        result = self.test_field.white_score / self.test_field.black_score
                    elif (side == SideType.BLACK):
                        result = self.test_field.black_score / self.test_field.white_score
                except ZeroDivisionError:
                    result = inf

                if (result > best_result):
                    best_result = result
                    optimal_moves.clear()
                    optimal_moves.append(moves)
                elif (result == best_result):
                    optimal_moves.append(moves)

                self.test_field = Field.copy(field_copy_test)

        optimal_move = []
        if (optimal_moves):
            # Фильтрация хода
            for move in choice(optimal_moves):
                if (side == SideType.WHITE and self.__field.type_at(move.from_x, move.from_y) in BLACK_CHECKERS):
                    break
                elif (side == SideType.BLACK and self.__field.type_at(move.from_x, move.from_y) in WHITE_CHECKERS):
                    break
                optimal_move.append(move)
        return optimal_move

    def __get_predicted_moves_list(self, side: SideType, current_prediction_depth: int = 0,
                                   all_moves_list: list[Move] = [], current_moves_list: list[Move] = [],
                                   required_moves_list: list[Move] = []) -> list[Move]:
        '''Предсказать все возможные ходы'''
        if (current_moves_list):
            all_moves_list.append(current_moves_list)
        else:
            all_moves_list.clear()

        if (required_moves_list):
            moves_list = required_moves_list
        else:
            now = False
            moves_list = self.__get_moves_list(side)

        if (moves_list and current_prediction_depth < MAX_DEPTH):
            field_copy = Field.copy(self.test_field)
            for move in moves_list:
                field_for_check = Field.copy(self.test_field)
                has_killed_checker = self.__handle_move(self.test_field, move, draw=False)
                required_moves_list = list(filter(
                    lambda required_move: move.to_x == required_move.from_x and move.to_y == required_move.from_y,
                    self.__get_required_moves_list(side, self.test_field)))
                # Если есть ещё ход этой же шашкой
                if (has_killed_checker and required_moves_list):
                    self.__get_predicted_moves_list(side, current_prediction_depth, all_moves_list,
                                                    current_moves_list + [move], required_moves_list)
                else:
                    self.__get_predicted_moves_list(SideType.opposite(side), current_prediction_depth + 1,
                                                    all_moves_list, current_moves_list + [move])

                self.test_field = Field.copy(field_copy)

        return all_moves_list
python multithreading tkinter multiprocessing minimax
1个回答
0
投票

在显示的片段中,我没有看到任何机制可以阻止“敌人回合”的“多次”计算立即开始。也许您只是同时开始几个这样的计算,并且速度减慢是由于 CPU 使用率造成的? 您可以添加单个线程作为实例属性,并重用它 -

... import queue ... def _ensure_calc_thread(self): if not getattr(self, "_calc_thread"): self._calc_thread = threading.Thread(target=self._calc_thread_driver) self._calc_queue = queue.Queue() self._calc_thread.start() def self._calc_thread_driver(self): while True: # this line waits until there is something on the queue. move_now = self._calc_queue.get() # maybe put a special marker in the queue to stop the thread # at the end of the game - just add an "if" statement here! self.__handle_enemy_turn() def handle_enemy_turn_async(game_instance): game_instance.__ensure_calc_thread() game_instance._calc_queue.put("go") # <- signals the "driver"to start calculating. # even if this is called more than once, the next call to "__handle_enemy_turn" will # just take place after the first one finihes running, and sets `__player_turn`. def __handle_enemy_turn(self): '''Обработка хода противника (компьютера)''' self.__player_turn = False ... self.__player_turn = True self.__check_for_game_over()

© www.soinside.com 2019 - 2024. All rights reserved.