Add: Map movement control

tasks/map/control/control.py

@@ -3,13 +3,13 @@ from functools import cached_property
 from module.base.timer import Timer
 from module.logger import logger
 from tasks.map.assets.assets_map_control import ROTATION_SWIPE_AREA
-from tasks.map.control.joystick import MapControlJoystick
+from tasks.map.control.joystick import JoystickContact, MapControlJoystick
+from tasks.map.control.waypoint import Waypoint, WaypointRun, WaypointStraightRun, ensure_waypoint
 from tasks.map.minimap.minimap import Minimap
+from tasks.map.resource.const import diff_to_180_180
 
 
 class MapControl(MapControlJoystick):
-    _rotation_swipe_interval = Timer(1.2, count=2)
-
     @cached_property
     def minimap(self) -> Minimap:
         return Minimap()
@@ -28,8 +28,10 @@ class MapControl(MapControlJoystick):
         """
         if self.minimap.is_rotation_near(target, threshold=threshold):
             return False
-        if not self._rotation_swipe_interval.reached():
-            return False
+
+        # if abs(self.minimap.rotation_diff(target)) > 60:
+        #     self.device.image_save()
+        #     exit(1)
 
         logger.info(f'Rotation set: {target}')
         diff = self.minimap.rotation_diff(target) * self.minimap.ROTATION_SWIPE_MULTIPLY
@@ -37,7 +39,6 @@ class MapControl(MapControlJoystick):
         diff = max(diff, -self.minimap.ROTATION_SWIPE_MAX_DISTANCE)
 
         self.device.swipe_vector((-diff, 0), box=ROTATION_SWIPE_AREA.area, duration=(0.2, 0.5))
-        self._rotation_swipe_interval.reset()
         return True
 
     def rotation_set(self, target, threshold=15, skip_first_screenshot=False):
@@ -52,6 +53,7 @@ class MapControl(MapControlJoystick):
         Returns:
             bool: If swiped rotation
         """
+        interval = Timer(1, count=2)
         while 1:
             if skip_first_screenshot:
                 skip_first_screenshot = False
@@ -65,5 +67,199 @@ class MapControl(MapControlJoystick):
                 logger.info(f'Rotation is now at: {target}')
                 break
 
+            if interval.reached():
                 if self.handle_rotation_set(target, threshold=threshold):
+                    interval.reset()
                     continue
+
+    def _goto(
+            self,
+            contact: JoystickContact,
+            waypoint: Waypoint,
+            end_point_opt=True,
+            skip_first_screenshot=False
+    ):
+        """
+        Point to point walk.
+
+        Args:
+            contact:
+                JoystickContact, must be wrapped with:
+                `with JoystickContact(self) as contact:`
+            waypoint:
+                Position to goto, (x, y)
+            end_point_opt:
+                True to enable endpoint optimizations,
+                character will smoothly approach target position
+            skip_first_screenshot:
+        """
+        logger.hr('Goto', level=2)
+        logger.info(f'Goto {waypoint}')
+        self.device.stuck_record_clear()
+        self.device.click_record_clear()
+
+        end_point_opt = end_point_opt and waypoint.end_point_opt
+        allow_2x_run = waypoint.speed in ['2x_run']
+        allow_straight_run = waypoint.speed in ['2x_run', 'straight_run']
+        allow_run = waypoint.speed in ['2x_run', 'straight_run', 'run']
+        allow_rotation_set = True
+        last_rotation = 0
+
+        direction_interval = Timer(0.5, count=1)
+        rotation_interval = Timer(0.3, count=1)
+        while 1:
+            if skip_first_screenshot:
+                skip_first_screenshot = False
+            else:
+                self.device.screenshot()
+
+            # Update
+            self.minimap.update(self.device.image)
+
+            # Arrive
+            if self.minimap.is_position_near(waypoint.position, threshold=waypoint.get_threshold(end_point_opt)):
+                logger.info(f'Arrive {waypoint}')
+                break
+
+            # Switch run case
+            diff = self.minimap.position_diff(waypoint.position)
+            if end_point_opt:
+                if allow_2x_run and diff < 20:
+                    logger.info(f'Approaching target, diff={round(diff, 1)}, disallow 2x_run')
+                    allow_2x_run = False
+                if allow_straight_run and diff < 15:
+                    logger.info(f'Approaching target, diff={round(diff, 1)}, disallow straight_run')
+                    direction_interval = Timer(0.2)
+                    self._map_2x_run_timer.reset()
+                    allow_straight_run = False
+                if allow_run and diff < 7:
+                    logger.info(f'Approaching target, diff={round(diff, 1)}, disallow run')
+                    direction_interval = Timer(0.2)
+                    allow_run = False
+
+            # Control
+            direction = self.minimap.position2direction(waypoint.position)
+            if allow_2x_run:
+                # Run with 2x_run button
+                # - Set rotation once
+                # - Continuous fine-tuning direction
+                # - Enable 2x_run
+                if allow_rotation_set:
+                    # Cache rotation cause rotation detection has a higher error rate
+                    last_rotation = self.minimap.rotation
+                    if self.minimap.is_rotation_near(direction, threshold=10):
+                        logger.info(f'Already at target rotation, '
+                                    f'current={last_rotation}, target={direction}, disallow rotation_set')
+                        allow_rotation_set = False
+                if allow_rotation_set and rotation_interval.reached():
+                    if self.handle_rotation_set(direction, threshold=10):
+                        rotation_interval.reset()
+                        direction_interval.reset()
+                if direction_interval.reached():
+                    contact.set(direction=diff_to_180_180(direction - last_rotation), run=True)
+                    direction_interval.reset()
+                self.handle_map_2x_run(run=True)
+            elif allow_straight_run:
+                # Run with 2x_run button
+                # - Set rotation once
+                # - Continuous fine-tuning direction
+                # - Disable 2x_run
+                if allow_rotation_set:
+                    # Cache rotation cause rotation detection has a higher error rate
+                    last_rotation = self.minimap.rotation
+                    if self.minimap.is_rotation_near(direction, threshold=10):
+                        logger.info(f'Already at target rotation, '
+                                    f'current={last_rotation}, target={direction}, disallow rotation_set')
+                        allow_rotation_set = False
+                if allow_rotation_set and rotation_interval.reached():
+                    if self.handle_rotation_set(direction, threshold=10):
+                        rotation_interval.reset()
+                        direction_interval.reset()
+                if direction_interval.reached():
+                    contact.set(direction=diff_to_180_180(direction - last_rotation), run=True)
+                    direction_interval.reset()
+                self.handle_map_2x_run(run=False)
+            elif allow_run:
+                # Run
+                # - No rotation set
+                # - Continuous fine-tuning direction
+                # - Disable 2x_run
+                if allow_rotation_set:
+                    last_rotation = self.minimap.rotation
+                    allow_rotation_set = False
+                if direction_interval.reached():
+                    contact.set(direction=diff_to_180_180(direction - last_rotation), run=True)
+                self.handle_map_2x_run(run=False)
+            else:
+                # Walk
+                # - Continuous fine-tuning direction
+                # - Disable 2x_run
+                if allow_rotation_set:
+                    last_rotation = self.minimap.rotation
+                    allow_rotation_set = False
+                if direction_interval.reached():
+                    contact.set(direction=diff_to_180_180(direction - last_rotation), run=False)
+                    direction_interval.reset()
+                self.handle_map_2x_run(run=False)
+
+    def goto(
+            self,
+            waypoints,
+            skip_first_screenshot=True
+    ):
+        """
+        Go along a list of position, or goto target position
+
+        Args:
+            waypoints:
+                position (x, y) to goto, or a list of position to go along.
+                Waypoint object to goto, or a list of Waypoint objects to go along.
+
+            skip_first_screenshot:
+        """
+        logger.hr('Goto', level=1)
+        if not isinstance(waypoints, list):
+            waypoints = [waypoints]
+        waypoints = [ensure_waypoint(point) for point in waypoints]
+        logger.info(f'Go along {len(waypoints)} waypoints')
+        end_list = [False for _ in waypoints]
+        end_list[-1] = True
+
+        with JoystickContact(self) as contact:
+            for point, end in zip(waypoints, end_list):
+                point: Waypoint
+                self._goto(
+                    contact=contact,
+                    waypoint=point,
+                    end_point_opt=end,
+                    skip_first_screenshot=skip_first_screenshot
+                )
+                skip_first_screenshot = True
+
+        end_point = waypoints[-1]
+        if end_point.end_point_rotation is not None:
+            self.rotation_set(end_point.end_point_rotation, threshold=end_point.end_point_rotation_threshold)
+
+
+if __name__ == '__main__':
+    # Control test in Himeko trail
+    # Must manually enter Himeko trail first and dismiss popup
+    self = MapControl('alas')
+    self.minimap.set_plane('Jarilo_BackwaterPass', floor='F1')
+    self.device.screenshot()
+    self.minimap.init_position((519, 359))
+    # Visit 3 items
+    self.goto([
+        WaypointRun((577.6, 363.4)),
+    ])
+    self.goto([
+        WaypointStraightRun((577.5, 369.4), end_point_rotation=200),
+    ])
+    self.goto([
+        WaypointRun((581.5, 387.3)),
+        WaypointRun((577.4, 411.5)),
+    ])
+    # Goto boss
+    self.goto([
+        WaypointStraightRun((607.6, 425.3)),
+    ])

tasks/map/control/joystick.py

@@ -1,15 +1,135 @@
+import math
 from functools import cached_property
 
 from module.base.timer import Timer
+from module.device.method.maatouch import MaatouchBuilder
+from module.device.method.minitouch import CommandBuilder, insert_swipe, random_normal_distribution
+from module.exception import ScriptError
 from module.logger import logger
 from tasks.base.ui import UI
 from tasks.map.assets.assets_map_control import *
 
 
+class JoystickContact:
+    CENTER = (JOYSTICK.area[0] + JOYSTICK.area[2]) / 2, (JOYSTICK.area[1] + JOYSTICK.area[3]) / 2
+    # Minimum radius 49px
+    RADIUS_WALK = (55, 65)
+    # Minimum radius 103px
+    RADIUS_RUN = (105, 115)
+
+    def __init__(self, main):
+        """
+        Args:
+            main (MapControlJoystick):
+        """
+        self.main = main
+        self.prev_point = None
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        """
+        Lift finger when:
+        - Walk event ends, JoystickContact ends
+        - Any error is raised
+        Can not lift finger when:
+        - Process is force terminated
+        """
+        builder = self.builder
+        if self.is_downed:
+            builder.up().commit()
+            builder.send()
+            logger.info('JoystickContact ends')
+        else:
+            logger.info('JoystickContact ends but it was never downed')
+
+    @property
+    def is_downed(self):
+        return self.prev_point is not None
+
+    @cached_property
+    def builder(self):
+        """
+        Initialize a command builder
+        """
+        method = self.main.config.Emulator_ControlMethod
+        if method == 'MaaTouch':
+            # Get the very first builder to initialize MaaTouch
+            _ = self.main.device.maatouch_builder
+            builder = MaatouchBuilder(self.main.device, contact=1)
+        elif method == 'minitouch':
+            # Get the very first builder to initialize minitouch
+            _ = self.main.device.minitouch_builder
+            builder = CommandBuilder(self.main.device, contact=1)
+        else:
+            raise ScriptError(f'Control method {method} does not support multi-finger, '
+                              f'please use MaaTouch or minitouch instead')
+
+        # def empty_func():
+        #     pass
+        #
+        # # No clear()
+        # builder.clear = empty_func
+        # No delay
+        builder.DEFAULT_DELAY = 0.
+
+        return builder
+
+    @classmethod
+    def direction2screen(cls, direction, run=True):
+        """
+        Args:
+            direction (int, float): Direction to goto (0~360)
+            run: True for character running, False for walking
+
+        Returns:
+            tuple[int, int]: Position on screen to control joystick
+        """
+        direction += random_normal_distribution(-5, 5, n=5)
+        radius = cls.RADIUS_RUN if run else cls.RADIUS_WALK
+        radius = random_normal_distribution(*radius, n=5)
+
+        direction = math.radians(direction)
+        point = (
+            cls.CENTER[0] + radius * math.sin(direction),
+            cls.CENTER[1] - radius * math.cos(direction),
+        )
+        point = (int(round(point[0])), int(round(point[1])))
+        return point
+
+    def set(self, direction, run=True):
+        """
+        Set joystick to given position
+
+        Args:
+            direction (int, float): Direction to goto (0~360)
+            run: True for character running, False for walking
+        """
+        logger.info(f'JoystickContact set to {direction}')
+        point = JoystickContact.direction2screen(direction, run=run)
+        builder = self.builder
+
+        if self.is_downed:
+            points = insert_swipe(p0=self.prev_point, p3=point, speed=20)
+            for point in points[1:]:
+                builder.move(*point).commit().wait(10)
+            builder.send()
+        else:
+            builder.down(*point).commit()
+            builder.send()
+            # Character starts moving, RUN button is still unavailable in a short time.
+            # Assume available in 0.3s
+            # We still have reties if 0.3s is incorrect.
+            self.main._map_2x_run_timer.set_current(0.7)
+
+        self.prev_point = point
+
+
 class MapControlJoystick(UI):
     _map_A_timer = Timer(1)
     _map_E_timer = Timer(1)
-    _map_run_timer = Timer(1)
+    _map_2x_run_timer = Timer(1)
 
     @cached_property
     def joystick_center(self) -> tuple[float, float]:
@@ -64,7 +184,7 @@ class MapControlJoystick(UI):
 
         return False
 
-    def handle_map_run(self):
+    def handle_map_2x_run(self, run=True):
         """
         Keep character running.
         Note that RUN button can only be clicked when character is moving.
@@ -74,9 +194,13 @@ class MapControlJoystick(UI):
         """
         is_running = self.image_color_count(RUN_BUTTON, color=(208, 183, 138), threshold=221, count=100)
 
-        if not is_running and self._map_run_timer.reached():
+        if run and not is_running and self._map_2x_run_timer.reached():
             self.device.click(RUN_BUTTON)
-            self._map_run_timer.reset()
+            self._map_2x_run_timer.reset()
+            return True
+        if not run and is_running and self._map_2x_run_timer.reached():
+            self.device.click(RUN_BUTTON)
+            self._map_2x_run_timer.reset()
             return True
 
         return False

tasks/map/control/waypoint.py

@@ -0,0 +1,79 @@
+from dataclasses import dataclass
+
+
+@dataclass
+class Waypoint:
+    # Position to goto, (x, y)
+    position: tuple
+    # Position diff < threshold is considered as arrived
+    # `threshold` is used first if it is set
+    threshold: int = None
+    # If `threshold` is not set, `waypoint_threshold` and `endpoint_threshold` are used
+    waypoint_threshold: int = 10
+    endpoint_threshold: int = 3
+    # Max move speed, '2x_run', 'straight_run', 'run', 'walk'
+    # See MapControl._goto() for details of each speed level
+    speed: str = '2x_run'
+
+    """
+    The following attributes are only be used if this waypoint is the end point of goto()
+    """
+    # True to enable endpoint optimizations, character will smoothly approach target position
+    # False to stop all controls at arrive
+    end_point_opt: bool = True
+    # Set rotation after arrive, 0~360
+    end_point_rotation: int = None
+    end_point_rotation_threshold: int = 15
+
+    def __str__(self):
+        return f'Waypoint({self.position})'
+
+    __repr__ = __str__
+
+    def get_threshold(self, end):
+        """
+        Args:
+            end: True if this is an end point
+
+        Returns:
+            int
+        """
+        if self.threshold is not None:
+            return self.threshold
+        if end:
+            return self.endpoint_threshold
+        else:
+            return self.waypoint_threshold
+
+
+def ensure_waypoint(point) -> Waypoint:
+    """
+    Args:
+        point: Position (x, y) or Waypoint object
+
+    Returns:
+        Waypoint:
+    """
+    if isinstance(point, Waypoint):
+        return point
+    return Waypoint(point)
+
+
+@dataclass(repr=False)
+class Waypoint2xRun(Waypoint):
+    speed: str = '2x_run'
+
+
+@dataclass(repr=False)
+class WaypointStraightRun(Waypoint):
+    speed: str = 'straight_run'
+
+
+@dataclass(repr=False)
+class WaypointRun(Waypoint):
+    speed: str = 'run'
+
+
+@dataclass(repr=False)
+class WaypointWalk(Waypoint):
+    speed: str = 'walk'

tasks/map/resource/const.py

@@ -128,10 +128,7 @@ class ResourceConst:
         Returns:
             float: Diff to current direction (-180~180)
         """
-        diff = (self.direction - target) % 360
-        if diff > 180:
-            diff -= 360
-        return diff
+        return diff_to_180_180(self.direction - target)
 
     def is_direction_near(self, target, threshold=15):
         return abs(self.direction_diff(target)) <= threshold
@@ -144,10 +141,32 @@ class ResourceConst:
         Returns:
             float: Diff to current rotation (-180~180)
         """
-        diff = (self.rotation - target) % 360
-        if diff > 180:
-            diff -= 360
-        return diff
+        return diff_to_180_180(self.rotation - target)
 
     def is_rotation_near(self, target, threshold=10):
         return abs(self.rotation_diff(target)) <= threshold
+
+
+def diff_to_180_180(diff):
+    """
+    Args:
+        diff: Degree diff
+
+    Returns:
+        float: Degree diff (-180~180)
+    """
+    diff = diff % 360
+    if diff > 180:
+        diff -= 360
+    return round(diff, 3)
+
+
+def diff_to_0_360(diff):
+    """
+    Args:
+        diff: Degree diff
+
+    Returns:
+        float: Degree diff (0~360)
+    """
+    return round(diff % 360, 3)