Refactor: Auto-switch between domain exit implements

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('>')])
 

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",

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):

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:
+            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