From 2591a3aee0e73bfd55d4955ee179230603961b5f Mon Sep 17 00:00:00 2001 From: LmeSzinc <37934724+LmeSzinc@users.noreply.github.com> Date: Tue, 14 Nov 2023 01:07:23 +0800 Subject: [PATCH] Refactor: Auto-switch between domain exit implements --- module/config/config_manual.py | 6 +++ tasks/map/control/waypoint.py | 3 ++ tasks/rogue/route/base.py | 49 ++++++++++++++++- tasks/rogue/route/exit.py | 97 +++++++++++++++++++++++++++++++--- 4 files changed, 148 insertions(+), 7 deletions(-) diff --git a/module/config/config_manual.py b/module/config/config_manual.py index 36b6d7d13..9943fe054 100644 --- a/module/config/config_manual.py +++ b/module/config/config_manual.py @@ -121,6 +121,12 @@ class ManualConfig: OS_ACTION_POINT_PRESERVE = 0 OS_CL1_YELLOW_COINS_PRESERVE = 100000 + """ + tasks.rogue + """ + # 2023.11.13 Migrate domain exit implementation, True to stop before domain exit + DOMAIN_EXIT_MIGRATE_DEV = False + ADDING = ''.join([chr(int(f)) for f in ManualConfig.OS_EXPLORE_CENTER.split('>')]) diff --git a/tasks/map/control/waypoint.py b/tasks/map/control/waypoint.py index ef13394f0..8110eb19f 100644 --- a/tasks/map/control/waypoint.py +++ b/tasks/map/control/waypoint.py @@ -51,6 +51,9 @@ class Waypoint: __repr__ = __str__ + def __bool__(self): + return True + def run_2x(self) -> "Waypoint": """ Product a Waypoint object with overridden "speed", diff --git a/tasks/rogue/route/base.py b/tasks/rogue/route/base.py index 2325fb06f..f31b1c702 100644 --- a/tasks/rogue/route/base.py +++ b/tasks/rogue/route/base.py @@ -267,6 +267,7 @@ class RouteBase(RouteBase_, RogueExit, RogueEvent, RogueReward): logger.hr('Domain single exit', level=1) waypoints = ensure_waypoints(waypoints) end_point = waypoints[-1] + end_point.min_speed = 'run' end_point.interact_radius = 5 end_point.expected_end.append(self._domain_exit_expected_end) @@ -274,6 +275,32 @@ class RouteBase(RouteBase_, RogueExit, RogueEvent, RogueReward): self._domain_exit_wait_next() return result + def _domain_exit_old(self): + """ + An old implementation that go along specific direction without retries + """ + logger.info(f'Using old predict_door()') + direction = self.predict_door_old() + direction_limit = 55 + if direction is not None: + if abs(direction) > direction_limit: + logger.warning(f'Unexpected direction to go: {direction}, limited in {direction_limit}') + if direction > 0: + direction = direction_limit + elif direction < 0: + direction = -direction_limit + + point = Waypoint( + position=(0, 0), + min_speed='run', + lock_direction=direction, + interact_radius=10000, + expected_end=[self._domain_exit_expected_end], + ) + self.goto(point) + self._domain_exit_wait_next() + return True + def domain_exit( self, *waypoints, @@ -281,22 +308,36 @@ class RouteBase(RouteBase_, RogueExit, RogueEvent, RogueReward): left_door: Waypoint = None, right_door: Waypoint = None ): + """ + Goto domain exit, choose one door, goto door + """ logger.hr('Domain exit', level=1) + # Goto the front of the two doors waypoints = ensure_waypoints(waypoints) end_point = waypoints[-1] end_point.endpoint_threshold = 1.5 self.goto(*waypoints) + # Rotate camera to insight two doors logger.hr('End rotation', level=2) self.rotation_set(end_rotation, threshold=10) + # Choose a door logger.hr('Find domain exit', level=2) + logger.info(f'Migrate={self.config.DOMAIN_EXIT_MIGRATE_DEV}, left_door={left_door}, right_door={right_door}') + if not self.config.DOMAIN_EXIT_MIGRATE_DEV and (not left_door and not right_door): + return self._domain_exit_old() + + logger.info(f'Using new predict_door()') door = self.predict_door() - if left_door is None or right_door is None: + if self.config.DOMAIN_EXIT_MIGRATE_DEV and self.exit_has_double_door and (not left_door or not right_door): logger.critical(f'Domain exit is not defined in: {self.route_func}') exit(1) + # Goto door if door == 'left_door': + if not left_door: + return self._domain_exit_old() if self.domain_single_exit(left_door): return True else: @@ -306,6 +347,8 @@ class RouteBase(RouteBase_, RogueExit, RogueEvent, RogueReward): else: return False elif door == 'right_door': + if not right_door: + return self._domain_exit_old() if self.domain_single_exit(right_door): return True else: @@ -316,6 +359,10 @@ class RouteBase(RouteBase_, RogueExit, RogueEvent, RogueReward): return False else: logger.error('Cannot goto either exit doors, try both') + if not left_door: + return self._domain_exit_old() + if not right_door: + return self._domain_exit_old() if self.domain_single_exit(left_door): return True elif self.domain_single_exit(right_door): diff --git a/tasks/rogue/route/exit.py b/tasks/rogue/route/exit.py index 7decefb2f..7f385dedd 100644 --- a/tasks/rogue/route/exit.py +++ b/tasks/rogue/route/exit.py @@ -139,6 +139,83 @@ class RogueExit(CombatInteract): logger.info(f'PlanarDoor: {planar_door}, direction: {direction}') return direction + def predict_door_by_name_old(self, image) -> float | None: + # Paint current name black + x1, y1, x2, y2 = OCR_MAP_NAME.area + image[y1:y2, x1:x2] = (0, 0, 0) + + ocr = OcrDomainExit(OCR_DOMAIN_EXIT) + results = ocr.matched_ocr(image, keyword_classes=MapPlane) + centers = [area_center(result.area) for result in results] + logger.info(f'DomainDoor: {centers}') + directions = [self.screen2direction(center) for center in centers] + + count = len(centers) + if count == 0: + logger.warning('No domain exit found') + return None + if count == 1: + logger.info(f'Goto next domain: {results[0]}') + return directions[0] + + # Doors >= 2 + for expect in [ + KEYWORDS_MAP_PLANE.Rogue_DomainBoss, + KEYWORDS_MAP_PLANE.Rogue_DomainElite, + KEYWORDS_MAP_PLANE.Rogue_DomainRespite, + ]: + for domain, direction in zip(results, directions): + if domain == expect: + logger.warning('Found multiple doors but has unique domain in it') + logger.info(f'Goto next domain: {domain}') + return direction + + logger.attr('DomainStrategy', self.config.RogueWorld_DomainStrategy) + if self.config.RogueWorld_DomainStrategy == 'occurrence': + for expect in [ + KEYWORDS_MAP_PLANE.Rogue_DomainTransaction, + KEYWORDS_MAP_PLANE.Rogue_DomainOccurrence, + KEYWORDS_MAP_PLANE.Rogue_DomainEncounter, + KEYWORDS_MAP_PLANE.Rogue_DomainCombat, + ]: + for domain, direction in zip(results, directions): + if domain == expect: + logger.info(f'Goto next domain: {domain}') + return direction + elif self.config.RogueWorld_DomainStrategy == 'combat': + for expect in [ + KEYWORDS_MAP_PLANE.Rogue_DomainCombat, + KEYWORDS_MAP_PLANE.Rogue_DomainEncounter, + KEYWORDS_MAP_PLANE.Rogue_DomainOccurrence, + KEYWORDS_MAP_PLANE.Rogue_DomainTransaction, + ]: + for domain, direction in zip(results, directions): + if domain == expect: + logger.info(f'Goto next domain: {domain}') + return direction + else: + logger.error(f'Unknown domain strategy: {self.config.RogueWorld_DomainStrategy}') + + logger.error('No domain was selected, return the first instead') + logger.info(f'Goto next domain: {results[0]}') + return directions[0] + + def predict_door_old(self, skip_first_screenshot=True) -> float | None: + timeout = Timer(3, count=6).start() + while 1: + if skip_first_screenshot: + skip_first_screenshot = False + else: + self.device.screenshot() + + if timeout.reached(): + logger.error('Predict door timeout') + return None + + direction = self.predict_door_by_name_old(self.device.image) + if direction is not None: + return direction + def predict_door_by_name(self, image) -> tuple[MapPlane | None, MapPlane | None]: """ Args: @@ -223,16 +300,18 @@ class RogueExit(CombatInteract): logger.error(f'Unknown domain strategy: {self.config.RogueWorld_DomainStrategy}') logger.error('No domain was selected, return the first instead') - if left_door is not None: + if left_door: logger.info(f'Goto next domain: left_door={left_door}') return 'left_door' - elif right_door is not None: + elif right_door: logger.info(f'Goto next domain: right_door={right_door}') return 'right_door' else: logger.error(f'No domain door') return None + exit_has_double_door = False + def predict_door(self, skip_first_screenshot=True) -> str | None: """ Args: @@ -242,19 +321,25 @@ class RogueExit(CombatInteract): str: 'left_door' or 'right_door' or None """ timeout = Timer(3, count=6).start() + self.exit_has_double_door = False while 1: if skip_first_screenshot: skip_first_screenshot = False else: self.device.screenshot() + # End if timeout.reached(): logger.error('Predict door timeout') return None left_door, right_door = self.predict_door_by_name(self.device.image) logger.info(f'DomainExit: left_door={left_door}, right_door={right_door}') - if left_door is not None or right_door is not None: - door = self.choose_door(left_door, right_door) - if door is not None: - return door + if not left_door and not right_door: + continue + + # End + self.exit_has_double_door = left_door and right_door + door = self.choose_door(left_door, right_door) + if door is not None: + return door