Add: Walk with result expecting

This commit is contained in:
LmeSzinc 2023-09-22 00:23:00 +08:00
parent 5358c5ff2e
commit cd01e1c402
7 changed files with 251 additions and 90 deletions

View File

@ -136,6 +136,10 @@ class Combat(CombatInteract, CombatPrepare, CombatState, CombatTeam, CombatSuppo
# End
if self.appear(COMBAT_AGAIN):
logger.info(f'Combat execute ended at {COMBAT_AGAIN}')
break
if self.is_in_main():
logger.info(f'Combat execute ended at page_main')
break
# Daemon

View File

@ -199,4 +199,4 @@ class ForgottenHallUI(DungeonUI):
if self.match_template_color(DUNGEON_ENTER_CHECKED):
logger.info("Forgotten hall dungeon entered")
break
joystick.handle_map_2x_run()
joystick.handle_map_run_2x()

View File

@ -2,14 +2,16 @@ from functools import cached_property
from module.base.timer import Timer
from module.logger import logger
from tasks.combat.combat import Combat
from tasks.map.assets.assets_map_control import ROTATION_SWIPE_AREA
from tasks.map.control.joystick import JoystickContact, MapControlJoystick
from tasks.map.control.waypoint import Waypoint, ensure_waypoint
from tasks.map.control.joystick import JoystickContact
from tasks.map.control.waypoint import Waypoint, ensure_waypoints
from tasks.map.interact.aim import AimDetectorMixin
from tasks.map.minimap.minimap import Minimap
from tasks.map.resource.const import diff_to_180_180
class MapControl(MapControlJoystick):
class MapControl(Combat, AimDetectorMixin):
@cached_property
def minimap(self) -> Minimap:
return Minimap()
@ -78,7 +80,7 @@ class MapControl(MapControlJoystick):
waypoint: Waypoint,
end_opt=True,
skip_first_screenshot=False
):
) -> list[str]:
"""
Point to point walk.
@ -92,6 +94,9 @@ class MapControl(MapControlJoystick):
True to enable endpoint optimizations,
character will smoothly approach target position
skip_first_screenshot:
Returns:
list[str]: A list of walk result
"""
logger.hr('Goto', level=2)
logger.info(f'Goto {waypoint}')
@ -99,51 +104,119 @@ class MapControl(MapControlJoystick):
self.device.click_record_clear()
end_opt = end_opt and waypoint.end_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_run_2x = waypoint.speed in ['run_2x']
allow_straight_run = waypoint.speed in ['run_2x', 'straight_run']
allow_run = waypoint.speed in ['run_2x', 'straight_run', 'run']
allow_walk = True
allow_rotation_set = True
last_rotation = 0
result = []
direction_interval = Timer(0.5, count=1)
rotation_interval = Timer(0.3, count=1)
aim_interval = Timer(0.3, count=1)
attacked_enemy = Timer(1.2, count=4)
attacked_item = Timer(0.6, count=2)
while 1:
if skip_first_screenshot:
skip_first_screenshot = False
else:
self.device.screenshot()
# End
for expected in waypoint.expected_end:
if callable(expected):
if expected():
logger.info(f'Walk result add: {expected.__name__}')
result.append(expected.__name__)
return result
if self.is_combat_executing():
logger.info('Walk result add: enemy')
result.append('enemy')
contact.up()
self.combat_execute()
if waypoint.early_stop:
return result
# The following detection require page_main
if not self.is_in_main():
attacked_enemy.clear()
attacked_item.clear()
continue
# Update
self.minimap.update(self.device.image)
if aim_interval.reached_and_reset():
self.aim.predict(self.device.image)
diff = self.minimap.position_diff(waypoint.position)
direction = self.minimap.position2direction(waypoint.position)
rotation_diff = self.minimap.direction_diff(direction)
logger.info(f'Position diff: {diff}, rotation: {rotation_diff}')
# Interact
if self.aim.aimed_enemy:
if 'enemy' in waypoint.expected_end:
if self.handle_map_A():
allow_run_2x = allow_straight_run = allow_run = allow_walk = False
attacked_enemy.reset()
direction_interval.reset()
rotation_interval.reset()
if attacked_enemy.started():
attacked_enemy.reset()
if self.aim.aimed_item:
if 'item' in waypoint.expected_end:
if self.handle_map_A():
allow_run_2x = allow_straight_run = allow_run = allow_walk = False
attacked_item.reset()
direction_interval.reset()
rotation_interval.reset()
if attacked_item.started():
attacked_item.reset()
else:
if attacked_item.started() and attacked_item.reached():
logger.info('Walk result add: item')
result.append('item')
if waypoint.early_stop:
return result
# Arrive
if not attacked_enemy.started() and not attacked_item.started():
if self.minimap.is_position_near(waypoint.position, threshold=waypoint.get_threshold(end_opt)):
logger.info(f'Arrive {waypoint}')
break
if not waypoint.expected_end or waypoint.match_results(result):
logger.info(f'Arrive waypoint: {waypoint}')
return result
else:
if waypoint.unexpected_confirm.reached():
logger.info(f'Arrive waypoint with unexpected result: {waypoint}')
return result
else:
waypoint.unexpected_confirm.reset()
# Switch run case
diff = self.minimap.position_diff(waypoint.position)
if end_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:
if allow_run_2x and diff < 20:
logger.info(f'Approaching target, diff={round(diff, 1)}, disallow run_2x')
allow_run_2x = False
if allow_straight_run and diff < 15 and not allow_rotation_set:
logger.info(f'Approaching target, diff={round(diff, 1)}, disallow straight_run')
direction_interval = Timer(0.2)
self.map_2x_run_timer.reset()
aim_interval = Timer(0.1)
self.map_run_2x_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)
aim_interval = Timer(0.2)
allow_run = False
# Control
direction = self.minimap.position2direction(waypoint.position)
if allow_2x_run:
# Run with 2x_run button
if allow_run_2x:
# Run with run_2x button
# - Set rotation once
# - Continuous fine-tuning direction
# - Enable 2x_run
# - Enable run_2x
if allow_rotation_set:
# Cache rotation cause rotation detection has a higher error rate
last_rotation = self.minimap.rotation
@ -158,12 +231,12 @@ class MapControl(MapControlJoystick):
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)
self.handle_map_run_2x(run=True)
elif allow_straight_run:
# Run straight forward
# - Set rotation once
# - Continuous fine-tuning direction
# - Disable 2x_run
# - Disable run_2x
if allow_rotation_set:
# Cache rotation cause rotation detection has a higher error rate
last_rotation = self.minimap.rotation
@ -178,67 +251,102 @@ class MapControl(MapControlJoystick):
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)
self.handle_map_run_2x(run=False)
elif allow_run:
# Run
# - No rotation set
# - Continuous fine-tuning direction
# - Disable 2x_run
# - Disable run_2x
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:
direction_interval.reset()
self.handle_map_run_2x(run=False)
elif allow_walk:
# Walk
# - Continuous fine-tuning direction
# - Disable 2x_run
# - Disable run_2x
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)
self.handle_map_run_2x(run=False)
else:
contact.up()
def goto(
self,
*waypoints,
skip_first_screenshot=True
):
def goto(self, *waypoints):
"""
Go along a list of position, or goto target position
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:
waypoints: position (x, y), a list of position to go along,
or a list of Waypoint objects to go along.
"""
logger.hr('Goto', level=1)
waypoints = [ensure_waypoint(point) for point in waypoints]
self.map_A_timer.clear()
self.map_E_timer.clear()
self.map_run_2x_timer.clear()
waypoints = ensure_waypoints(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(
for waypoint, end in zip(waypoints, end_list):
waypoint: Waypoint
result = self._goto(
contact=contact,
waypoint=point,
waypoint=waypoint,
end_opt=end,
skip_first_screenshot=skip_first_screenshot
skip_first_screenshot=True,
)
skip_first_screenshot = True
expected = waypoint.expected_to_str(waypoint.expected_end)
logger.info(f'Arrive waypoint, expected: {expected}, result: {result}')
matched = waypoint.match_results(result)
if not waypoint.expected_end or matched:
logger.info(f'Arrive waypoint with expected result: {matched}')
else:
logger.warning(f'Arrive waypoint with unexpected result: {result}')
end_point = waypoints[-1]
if end_point.end_rotation is not None:
logger.hr('End rotation', level=1)
logger.hr('End rotation', level=2)
self.rotation_set(end_point.end_rotation, threshold=end_point.end_rotation_threshold)
def clear_item(self, *waypoints):
"""
Go along a list of position and clear destructive object at last.
Args:
waypoints: position (x, y), a list of position to go along.
or a list of Waypoint objects to go along.
"""
logger.hr('Clear item', level=1)
waypoints = ensure_waypoints(waypoints)
end_point = waypoints[-1]
end_point.expected_end.append('item')
self.goto(*waypoints)
def clear_enemy(self, *waypoints):
"""
Go along a list of position and enemy at last.
Args:
waypoints: position (x, y), a list of position to go along.
or a list of Waypoint objects to go along.
"""
logger.hr('Clear item', level=1)
waypoints = ensure_waypoints(waypoints)
end_point = waypoints[-1]
end_point.expected_end.append('enemy')
self.goto(*waypoints)
if __name__ == '__main__':
# Control test in Himeko trial
@ -248,17 +356,14 @@ if __name__ == '__main__':
self.device.screenshot()
self.minimap.init_position((519, 359))
# Visit 3 items
self.goto(
Waypoint((577.6, 363.4)),
self.clear_item(
Waypoint.run_2x((587.6, 366.9)),
)
self.goto(
Waypoint((577.5, 369.4), end_rotation=200),
)
self.goto(
Waypoint((581.5, 387.3)),
Waypoint((577.4, 411.5)),
self.clear_item((575.5, 377.4))
self.clear_item(
# Go through arched door
Waypoint.run((581.5, 383.3), threshold=3),
Waypoint.run((575.7, 417.2)),
)
# Goto boss
self.goto(
Waypoint((607.6, 425.3)),
)
self.clear_enemy((613.5, 427.3))

View File

@ -16,7 +16,7 @@ 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)
RADIUS_WALK = (25, 40)
# Minimum radius 103px
RADIUS_RUN = (105, 115)
@ -114,7 +114,7 @@ class JoystickContact:
direction (int, float): Direction to goto (0~360)
run: True for character running, False for walking
"""
logger.info(f'JoystickContact set to {direction}')
logger.info(f'JoystickContact set to {direction}, run={run}')
point = JoystickContact.direction2screen(direction, run=run)
builder = self.builder
@ -136,7 +136,7 @@ class JoystickContact:
# 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.main.map_run_2x_timer.set_current(0.7)
self.main.joystick_lost_timer.reset()
self.prev_point = point
@ -145,7 +145,7 @@ class JoystickContact:
class MapControlJoystick(UI):
map_A_timer = Timer(1)
map_E_timer = Timer(1)
map_2x_run_timer = Timer(1)
map_run_2x_timer = Timer(1)
joystick_lost_timer = Timer(1, count=2)
@ -235,7 +235,7 @@ class MapControlJoystick(UI):
return False
def handle_map_2x_run(self, run=True):
def handle_map_run_2x(self, run=True):
"""
Keep character running.
Note that RUN button can only be clicked when character is moving.
@ -245,13 +245,13 @@ class MapControlJoystick(UI):
"""
is_running = self.image_color_count(RUN_BUTTON, color=(208, 183, 138), threshold=221, count=100)
if run and not is_running and self.map_2x_run_timer.reached():
if run and not is_running and self.map_run_2x_timer.reached():
self.device.click(RUN_BUTTON)
self.map_2x_run_timer.reset()
self.map_run_2x_timer.reset()
return True
if not run and is_running and self.map_2x_run_timer.reached():
if not run and is_running and self.map_run_2x_timer.reached():
self.device.click(RUN_BUTTON)
self.map_2x_run_timer.reset()
self.map_run_2x_timer.reset()
return True
return False

View File

@ -1,4 +1,6 @@
from dataclasses import dataclass
from dataclasses import dataclass, field
from module.base.timer import Timer
@dataclass
@ -11,7 +13,7 @@ class Waypoint:
# 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'
# Max move speed, 'run_2x', 'straight_run', 'run', 'walk'
# See MapControl._goto() for details of each speed level
speed: str = 'straight_run'
@ -25,11 +27,50 @@ class Waypoint:
end_rotation: int = None
end_rotation_threshold: int = 15
"""
Walk
"""
# A list of expected events, e.g. ['enemy', 'item']
# - "enemy", finished any combat
# - "item", destroyed any destructive objects
# - "interact", have map interact option (interact is not handled)
# - callable, A function that returns bool, True represents stop
# Or empty list [] for just walking
expected_end: list = field(default_factory=lambda: [])
# If triggered any expected event, consider arrive and stop walking
early_stop: bool = True
# Confirm timer if arrived but didn't trigger any expected event
unexpected_confirm: Timer = field(default_factory=lambda: Timer(2, count=6))
def __str__(self):
return f'Waypoint({self.position})'
__repr__ = __str__
@classmethod
def run_2x(cls, *args, **kwargs) -> "Waypoint":
"""
Product a Waypoint object with overridden "speed",
see Waypoint class for args.
"""
kwargs['speed'] = 'run_2x'
return cls(*args, **kwargs)
@classmethod
def straight_run(cls, *args, **kwargs) -> "Waypoint":
kwargs['speed'] = 'straight_run'
return cls(*args, **kwargs)
@classmethod
def run(cls, *args, **kwargs) -> "Waypoint":
kwargs['speed'] = 'run'
return cls(*args, **kwargs)
@classmethod
def walk(cls, *args, **kwargs) -> "Waypoint":
kwargs['speed'] = 'walk'
return cls(*args, **kwargs)
def get_threshold(self, end):
"""
Args:
@ -45,6 +86,27 @@ class Waypoint:
else:
return self.waypoint_threshold
@staticmethod
def expected_to_str(results: list) -> list[str]:
return [result.__name__ if callable(result) else str(result) for result in results]
def match_results(self, results) -> list[str]:
"""
Args:
results:
Returns:
list[str]: A list if matched results
"""
if not results and not self.expected_end:
return []
results = set(self.expected_to_str(results))
expected_end = set(self.expected_to_str(self.expected_end))
same = results.intersection(expected_end)
return list(same)
def ensure_waypoint(point) -> Waypoint:
"""
@ -54,26 +116,13 @@ def ensure_waypoint(point) -> Waypoint:
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'
def ensure_waypoints(points) -> list[Waypoint]:
if not isinstance(points, (list, tuple)):
points = [points]
return [ensure_waypoint(point) for point in points]

View File

@ -390,7 +390,7 @@ if __name__ == '__main__':
# Set plane, assume starting from Jarilo_AdministrativeDistrict
self.set_plane('Jarilo_AdministrativeDistrict', floor='F1')
ui = UI('alas')
ui = UI('src')
ui.device.disable_stuck_detection()
# Set starter point. Starter point will be calculated if it's missing but may contain errors.
# With starter point set, position is only searched around starter point and new position becomes new starter point.

View File

@ -101,7 +101,9 @@ class ResourceConst:
Returns:
float: Distance to current position
"""
return np.linalg.norm(np.subtract(target, self.position))
diff = np.linalg.norm(np.subtract(target, self.position))
diff = round(diff, 3)
return diff
def is_position_near(self, target, threshold=5):
return self.position_diff(target) <= threshold
@ -118,6 +120,7 @@ class ResourceConst:
theta = np.rad2deg(np.arccos(-diff[1] / np.linalg.norm(diff)))
if diff[0] < 0:
theta = 360 - theta
theta = round(theta, 3)
return theta
def direction_diff(self, target):