mirror of
https://github.com/LmeSzinc/StarRailCopilot.git
synced 2024-11-25 01:49:41 +00:00
Add: Rogue route framework
This commit is contained in:
LmeSzinc
parent
f2e99a5db6
commit
76805d6753
@ -1,31 +1,30 @@
|
||||
import os
|
||||
import re
|
||||
from dataclasses import dataclass, fields
|
||||
from typing import Iterator
|
||||
|
||||
from module.base.code_generator import CodeGenerator
|
||||
import numpy as np
|
||||
from tqdm import tqdm
|
||||
|
||||
|
||||
@dataclass
|
||||
class RouteData:
|
||||
name: str
|
||||
route: str
|
||||
plane: str
|
||||
floor: str = 'F1'
|
||||
position: tuple = None
|
||||
from module.base.code_generator import CodeGenerator, MarkdownGenerator
|
||||
from module.base.decorator import cached_property
|
||||
from module.base.utils import SelectedGrids, load_image
|
||||
from module.config.utils import iter_folder
|
||||
from tasks.map.route.model import RouteModel
|
||||
from tasks.rogue.route.model import RogueRouteListModel, RogueRouteModel, RogueWaypointListModel, RogueWaypointModel
|
||||
|
||||
|
||||
class RouteExtract:
|
||||
def __init__(self, folder):
|
||||
self.folder = folder
|
||||
|
||||
def iter_files(self):
|
||||
def iter_files(self) -> Iterator[str]:
|
||||
for path, folders, files in os.walk(self.folder):
|
||||
path = path.replace('\\', '/')
|
||||
for file in files:
|
||||
if file.endswith('.py'):
|
||||
yield f'{path}/{file}'
|
||||
|
||||
def extract_route(self, file):
|
||||
def extract_route(self, file) -> Iterator[RouteModel]:
|
||||
print(f'Extract {file}')
|
||||
with open(file, 'r', encoding='utf-8') as f:
|
||||
content = f.read()
|
||||
@ -37,10 +36,10 @@ class RouteExtract:
|
||||
"""
|
||||
regex = re.compile(
|
||||
r'def (?P<func>[a-zA-Z0-9_]*?)\(self\):.*?'
|
||||
r'self\.map_init\((.*?)\)'
|
||||
, re.DOTALL)
|
||||
r'self\.map_init\((.*?)\)',
|
||||
re.DOTALL)
|
||||
file = file.replace(self.folder, '').replace('.py', '').replace('/', '_').strip('_')
|
||||
module = f"{self.folder.strip('./').replace('/', '.')}.{file.replace('_', '.')}"
|
||||
module = f"{self.folder.strip('./').replace('/', '.')}.{file}"
|
||||
|
||||
for result in regex.findall(content):
|
||||
func, data = result
|
||||
@ -62,30 +61,384 @@ class RouteExtract:
|
||||
else:
|
||||
position = None
|
||||
|
||||
yield RouteData(
|
||||
name=f'{file}__{func}',
|
||||
name = f'{file}__{func}'
|
||||
yield RouteModel(
|
||||
name=name,
|
||||
route=f'{module}:{func}',
|
||||
plane=plane,
|
||||
floor=floor,
|
||||
position=position,
|
||||
)
|
||||
|
||||
def iter_route(self):
|
||||
"""
|
||||
Yields:
|
||||
RouteData
|
||||
"""
|
||||
for f in self.iter_files():
|
||||
for row in self.extract_route(f):
|
||||
yield row
|
||||
|
||||
def write(self, file):
|
||||
gen = CodeGenerator()
|
||||
gen.Import("""
|
||||
from tasks.map.route.base import RouteData
|
||||
from tasks.map.route.model import RouteModel
|
||||
""")
|
||||
gen.CommentAutoGenerage('dev_tools.route_extract')
|
||||
|
||||
for f in self.iter_files():
|
||||
for row in self.extract_route(f):
|
||||
with gen.Object(key=row.name, object_class='RouteData'):
|
||||
for key in fields(row):
|
||||
value = getattr(row, key.name)
|
||||
gen.ObjectAttr(key.name, value)
|
||||
for row in self.iter_route():
|
||||
with gen.Object(key=row.name, object_class='RouteModel'):
|
||||
for key, value in row.__iter__():
|
||||
gen.ObjectAttr(key, value)
|
||||
gen.write(file)
|
||||
|
||||
|
||||
def model_to_json(model, file):
|
||||
content = model.model_dump_json(indent=2)
|
||||
with open(file, 'w', encoding='utf-8') as f:
|
||||
f.write(content)
|
||||
|
||||
|
||||
class RouteDetect:
|
||||
GEN_END = '===== End of generated waypoints ====='
|
||||
|
||||
def __init__(self, folder):
|
||||
self.folder = os.path.abspath(folder)
|
||||
print(self.folder)
|
||||
self.waypoints = SelectedGrids(list(self.iter_image()))
|
||||
|
||||
@cached_property
|
||||
def detector(self):
|
||||
from tasks.map.minimap.brute_force import MinimapWrapper
|
||||
return MinimapWrapper()
|
||||
|
||||
def get_minimap(self, route: RogueWaypointModel):
|
||||
return self.detector.all_minimap[route.plane_floor]
|
||||
|
||||
def iter_image(self) -> Iterator[RogueWaypointModel]:
|
||||
regex_posi = re.compile(r'_?X(\d+)Y(\d+)')
|
||||
for domain_folder in iter_folder(self.folder, is_dir=True):
|
||||
domain = os.path.basename(domain_folder)
|
||||
for route_folder in iter_folder(domain_folder, is_dir=True):
|
||||
route = os.path.basename(route_folder)
|
||||
try:
|
||||
for image_file in iter_folder(os.path.join(route_folder, 'route'), ext='.png'):
|
||||
waypoint = os.path.basename(image_file[:-4])
|
||||
|
||||
parts = route.split('_', maxsplit=3)
|
||||
if len(parts) == 4:
|
||||
world, plane, floor, position = parts
|
||||
res = regex_posi.search(position)
|
||||
if res:
|
||||
position = int(res.group(1)), int(res.group(2))
|
||||
else:
|
||||
position = (0, 0)
|
||||
elif len(parts) == 3:
|
||||
world, plane, floor = parts
|
||||
position = (0, 0)
|
||||
elif len(parts) == 2:
|
||||
world, plane = parts
|
||||
floor = 'F1'
|
||||
position = (0, 0)
|
||||
else:
|
||||
continue
|
||||
|
||||
file = f'{self.folder}/{domain}/{route}/route/{waypoint}.png'
|
||||
res = regex_posi.search(waypoint)
|
||||
if res:
|
||||
position = int(res.group(1)), int(res.group(2))
|
||||
# waypoint = regex_posi.sub('', waypoint)
|
||||
elif waypoint != 'spawn':
|
||||
position = (0, 0)
|
||||
model = RogueWaypointModel(
|
||||
domain=domain,
|
||||
route=route,
|
||||
waypoint=waypoint,
|
||||
index=0,
|
||||
file=file,
|
||||
plane=f'{world}_{plane}',
|
||||
floor=floor,
|
||||
position=position,
|
||||
direction=0.,
|
||||
rotation=0,
|
||||
)
|
||||
yield model
|
||||
# deep_set(out, keys=[image.route, image.waypoint], value=image)
|
||||
except FileNotFoundError:
|
||||
pass
|
||||
|
||||
def predict(self):
|
||||
regex_posi = re.compile(r'_?X(\d+)Y(\d+)')
|
||||
for waypoint in tqdm(self.waypoints.grids):
|
||||
waypoint: RogueWaypointModel = waypoint
|
||||
minimap = self.get_minimap(waypoint)
|
||||
im = load_image(waypoint.file)
|
||||
|
||||
prev = waypoint.position
|
||||
minimap.init_position(waypoint.position, show_log=False)
|
||||
minimap.update(im, show_log=False)
|
||||
waypoint.position = minimap.position
|
||||
waypoint.direction = minimap.direction
|
||||
waypoint.rotation = minimap.rotation
|
||||
if prev != (0, 0) and np.linalg.norm(np.subtract(waypoint.position, prev)) > 1.5:
|
||||
if waypoint.is_spawn:
|
||||
print(f'Position changed: {self.folder}/{waypoint.domain}/{waypoint.route}'
|
||||
f' -> {waypoint.plane}_{waypoint.floor}_{waypoint.positionXY}')
|
||||
else:
|
||||
name = regex_posi.sub('', waypoint.waypoint)
|
||||
print(f'Position changed: {waypoint.file}'
|
||||
f' -> {name}_{waypoint.positionXY}')
|
||||
|
||||
self.waypoints.create_index('route')
|
||||
# Sort by distance
|
||||
for waypoints in self.waypoints.indexes.values():
|
||||
waypoints = self.sort_waypoints(waypoints.grids)
|
||||
for index, waypoint in enumerate(waypoints):
|
||||
waypoint.index = index
|
||||
self.waypoints = self.waypoints.sort('route', 'index')
|
||||
|
||||
@staticmethod
|
||||
def sort_waypoints(waypoints: list[RogueWaypointModel]) -> list[RogueWaypointModel]:
|
||||
waypoints = sorted(waypoints, key=lambda point: point.waypoint, reverse=True)
|
||||
middle = [point for point in waypoints if not point.is_spawn and not point.is_exit]
|
||||
if not middle:
|
||||
return waypoints
|
||||
|
||||
try:
|
||||
spawn: RogueWaypointModel = [point for point in waypoints if point.is_spawn][0]
|
||||
except IndexError:
|
||||
return waypoints
|
||||
|
||||
prev = spawn.position
|
||||
if prev == (0, 0):
|
||||
return waypoints
|
||||
|
||||
sorted_middle = []
|
||||
while len(middle):
|
||||
distance = np.array([point.position for point in middle]) - prev
|
||||
distance = np.linalg.norm(distance, axis=1)
|
||||
index = np.argmin(distance)
|
||||
sorted_middle.append(middle[index])
|
||||
middle.pop(index)
|
||||
|
||||
end = [point for point in waypoints if point.is_exit]
|
||||
waypoints = [spawn] + sorted_middle + end
|
||||
return waypoints
|
||||
|
||||
def write(self):
|
||||
waypoints = RogueWaypointListModel(self.waypoints.grids)
|
||||
model_to_json(waypoints, f'{self.folder}/data.json')
|
||||
|
||||
def gen_route(self, waypoints: SelectedGrids):
|
||||
gen = CodeGenerator()
|
||||
|
||||
spawn: RogueWaypointModel = waypoints.select(is_spawn=True).first_or_none()
|
||||
exit_: RogueWaypointModel = waypoints.select(is_exit=True).first_or_none()
|
||||
if spawn is None or exit_ is None:
|
||||
return
|
||||
|
||||
class WaypointRepr:
|
||||
def __init__(self, position):
|
||||
if isinstance(position, RogueWaypointModel):
|
||||
position = position.position
|
||||
self.position = tuple(position)
|
||||
|
||||
def __repr__(self):
|
||||
return f'Waypoint({self.position})'
|
||||
|
||||
__str__ = __repr__
|
||||
|
||||
def filter_waypoints(name: str):
|
||||
return lambda x: x.waypoint.startswith(name)
|
||||
|
||||
def clear_elite():
|
||||
with gen.Object('self.clear_elite'):
|
||||
for w in waypoints.filter(filter_waypoints('enemy')):
|
||||
gen.ObjectAttr(value=WaypointRepr(w))
|
||||
|
||||
def clear_item():
|
||||
with gen.Object('self.clear_item'):
|
||||
for w in waypoints.filter(filter_waypoints('item')):
|
||||
gen.ObjectAttr(value=WaypointRepr(w))
|
||||
|
||||
def clear_event():
|
||||
with gen.Object('self.clear_event'):
|
||||
for w in waypoints.filter(filter_waypoints('event')):
|
||||
gen.ObjectAttr(value=WaypointRepr(w))
|
||||
|
||||
def domain_reward():
|
||||
with gen.Object('self.domain_reward'):
|
||||
for w in waypoints.filter(filter_waypoints('reward')):
|
||||
gen.ObjectAttr(value=WaypointRepr(w))
|
||||
|
||||
def domain_herta():
|
||||
with gen.Object('self.domain_herta'):
|
||||
for w in waypoints.filter(filter_waypoints('herta')):
|
||||
gen.ObjectAttr(value=WaypointRepr(w))
|
||||
|
||||
with gen.tab():
|
||||
with gen.Def(name=spawn.route, args='self'):
|
||||
table = MarkdownGenerator(['Waypoint', 'Position', 'Direction', 'Rotation'])
|
||||
for waypoint in waypoints:
|
||||
table.add_row([
|
||||
waypoint.waypoint,
|
||||
f'{WaypointRepr(waypoint)},',
|
||||
waypoint.direction,
|
||||
waypoint.rotation
|
||||
])
|
||||
gen.add('"""')
|
||||
for row in table.generate():
|
||||
gen.add(row)
|
||||
gen.add('"""')
|
||||
position = tuple(spawn.position)
|
||||
gen.add(f'self.map_init(plane={spawn.plane}, floor="{spawn.floor}", position={position})')
|
||||
if spawn.is_DomainBoss or spawn.is_DomainElite or spawn.is_DomainRespite:
|
||||
# Domain has only 1 exit
|
||||
pass
|
||||
else:
|
||||
gen.add(f'self.register_domain_exit({WaypointRepr(exit_)}, end_rotation={exit_.rotation})')
|
||||
|
||||
# Domain specific
|
||||
if spawn.is_DomainBoss or spawn.is_DomainElite:
|
||||
gen.Empty()
|
||||
clear_elite()
|
||||
domain_reward()
|
||||
if spawn.is_DomainRespite:
|
||||
gen.Empty()
|
||||
clear_item()
|
||||
domain_herta()
|
||||
if spawn.is_DomainOccurrence or spawn.is_DomainTransaction:
|
||||
gen.Empty()
|
||||
clear_item()
|
||||
clear_event()
|
||||
if spawn.is_DomainBoss or spawn.is_DomainElite or spawn.is_DomainRespite:
|
||||
# Domain has only 1 exit
|
||||
with gen.Object('self.domain_single_exit'):
|
||||
gen.ObjectAttr(value=WaypointRepr(exit_))
|
||||
# Single exit does not need end_rotation
|
||||
# gen.ObjectAttr(key='end_rotation', value=exit_.rotation)
|
||||
|
||||
gen.Comment(self.GEN_END)
|
||||
|
||||
return gen.generate()
|
||||
|
||||
def insert(self, folder, base='tasks.map.route.base'):
|
||||
# Create folder
|
||||
self.waypoints.create_index('domain')
|
||||
for index, waypoints in self.waypoints.indexes.items():
|
||||
domain = index[0]
|
||||
os.makedirs(f'{folder}/{domain}', exist_ok=True)
|
||||
|
||||
# Create file
|
||||
self.waypoints.create_index('domain', 'plane', 'floor')
|
||||
for index, waypoints in self.waypoints.indexes.items():
|
||||
domain, plane, floor = index
|
||||
file = f'{folder}/{domain}/{plane}_{floor}.py'
|
||||
if not os.path.exists(file):
|
||||
gen = CodeGenerator()
|
||||
gen.Import(f"""
|
||||
from {base} import RouteBase
|
||||
""")
|
||||
with gen.Class('Route', inherit='RouteBase'):
|
||||
pass
|
||||
# gen.Pass()
|
||||
gen.write(file)
|
||||
|
||||
for index, routes in self.waypoints.indexes.items():
|
||||
domain, plane, floor = index
|
||||
file = f'{folder}/{domain}/{plane}_{floor}.py'
|
||||
with open(file, 'r', encoding='utf-8') as f:
|
||||
content = f.read()
|
||||
# Add import
|
||||
if base != 'tasks.map.route.base':
|
||||
content = content.replace('tasks.map.route.base', base)
|
||||
p = '_'.join(plane.split('_', maxsplit=2)[:2])
|
||||
imp = [
|
||||
'from tasks.map.control.waypoint import Waypoint',
|
||||
f'from tasks.map.keywords.plane import {p}',
|
||||
f'from {base} import RouteBase',
|
||||
][::-1]
|
||||
res = re.search(r'^(.*?)class Route', content, re.DOTALL)
|
||||
if res:
|
||||
head = res.group(1)
|
||||
for row in imp:
|
||||
if row not in head:
|
||||
content = row + '\n' + content
|
||||
# Replace or add routes
|
||||
routes.create_index('route')
|
||||
for waypoints in routes.indexes.values():
|
||||
spawn = waypoints.select(is_spawn=True).first_or_none()
|
||||
if spawn is None:
|
||||
continue
|
||||
regex = re.compile(rf'def {spawn.route}.*?{self.GEN_END}', re.DOTALL)
|
||||
res = regex.search(content)
|
||||
if res:
|
||||
before = res.group(0).strip()
|
||||
after = self.gen_route(waypoints).strip()
|
||||
content = content.replace(before, after)
|
||||
else:
|
||||
content += '\n' + self.gen_route(waypoints)
|
||||
|
||||
# Sort routes
|
||||
regex = re.compile(
|
||||
r'(?=(\n def ([a-zA-Z0-9_]+)\(.*?\n def|\n def ([a-zA-Z0-9_]+)\(.*?$))', re.DOTALL)
|
||||
funcs = regex.findall(content)
|
||||
|
||||
known_routes = [route[0] for route in routes.indexes.keys()]
|
||||
routes = []
|
||||
for code, route1, route2 in funcs:
|
||||
if route1:
|
||||
route = route1
|
||||
elif route2:
|
||||
route = route2
|
||||
else:
|
||||
continue
|
||||
if route not in known_routes:
|
||||
continue
|
||||
code = code.removesuffix('\n def').removeprefix('\n')
|
||||
routes.append((route, code))
|
||||
|
||||
sorted_routes = sorted(routes, key=lambda x: x[0])
|
||||
routes = [route[1] for route in routes]
|
||||
sorted_routes = [route[1] for route in sorted_routes]
|
||||
new = ''
|
||||
for before, after in zip(routes, sorted_routes):
|
||||
left = content.index(before)
|
||||
right = left + len(before)
|
||||
new += content[:left]
|
||||
new += after
|
||||
content = content[right:]
|
||||
new += content
|
||||
content = new
|
||||
|
||||
# Format
|
||||
content = re.sub(r'[\n ]+ def', '\n\n def', content, re.DOTALL)
|
||||
content = content.rstrip('\n') + '\n'
|
||||
# Write
|
||||
with open(file, 'w', encoding='utf-8', newline='') as f:
|
||||
f.write(content)
|
||||
|
||||
|
||||
def rogue_extract(folder):
|
||||
print('rogue_extract')
|
||||
|
||||
def iter_route():
|
||||
for row in RouteExtract(f'{folder}').iter_route():
|
||||
domain = row.name.split('_', maxsplit=1)[0]
|
||||
row = RogueRouteModel(domain=domain, **row.model_dump())
|
||||
row.name = f'{row.domain}_{row.route.split(":")[1]}'
|
||||
row.route = row.route.replace('_', '.', 1)
|
||||
yield row
|
||||
|
||||
routes = RogueRouteListModel(list(iter_route()))
|
||||
model_to_json(routes, f'{folder}/route.json')
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
os.chdir(os.path.join(os.path.dirname(__file__), '../'))
|
||||
RouteExtract('./route/daily').write('./tasks/map/route/route/daily.py')
|
||||
|
||||
self = RouteDetect('../SrcRoute/rogue')
|
||||
self.predict()
|
||||
self.write()
|
||||
self.insert('./route/rogue', base='tasks.rogue.route.base')
|
||||
|
||||
@ -108,6 +108,13 @@ class MapPlane(Keyword):
|
||||
else:
|
||||
raise ScriptError(f'Plane {self} does not have floor {floor}')
|
||||
|
||||
@cached_property
|
||||
def rogue_domain(self) -> str:
|
||||
if self.name.startswith('Rogue_Domain'):
|
||||
return self.name.removeprefix('Rogue_Domain')
|
||||
else:
|
||||
return ''
|
||||
|
||||
|
||||
@dataclass(repr=False)
|
||||
class MapWorld(Keyword):
|
||||
|
||||
@ -1,19 +1,8 @@
|
||||
from dataclasses import dataclass
|
||||
|
||||
from tasks.map.control.control import MapControl
|
||||
from tasks.map.control.waypoint import Waypoint
|
||||
from tasks.map.keywords import MapPlane
|
||||
|
||||
|
||||
@dataclass
|
||||
class RouteData:
|
||||
name: str
|
||||
route: str
|
||||
plane: str
|
||||
floor: str = 'F1'
|
||||
position: tuple = None
|
||||
|
||||
|
||||
class RouteBase(MapControl):
|
||||
"""
|
||||
Base class of `Route`
|
||||
@ -59,3 +48,9 @@ class RouteBase(MapControl):
|
||||
self.minimap.set_plane(plane, floor=floor)
|
||||
if position is not None:
|
||||
self.minimap.init_position(position)
|
||||
|
||||
def before_route(self):
|
||||
pass
|
||||
|
||||
def after_route(self):
|
||||
pass
|
||||
|
||||
@ -5,7 +5,12 @@ from module.base.decorator import del_cached_property
|
||||
from module.exception import ScriptError
|
||||
from module.logger import logger
|
||||
from tasks.base.ui import UI
|
||||
from tasks.map.route.base import RouteBase, RouteData
|
||||
from tasks.map.route.base import RouteBase
|
||||
from tasks.map.route.model import RouteModel
|
||||
|
||||
|
||||
def empty_function(*arg, **kwargs):
|
||||
return False
|
||||
|
||||
|
||||
class RouteLoader(UI):
|
||||
@ -18,14 +23,14 @@ class RouteLoader(UI):
|
||||
self.route_module = ''
|
||||
self.route_func = ''
|
||||
|
||||
def route_run(self, route: RouteData | str):
|
||||
def route_run(self, route: RouteModel | str):
|
||||
"""
|
||||
Args:
|
||||
route: .py module path such as `route.daily.ForgottenHallStage1:route`
|
||||
which will load `./route/daily/ForgottenHallStage1.py` and run `Route.route()`
|
||||
"""
|
||||
logger.hr('Route run', level=1)
|
||||
if isinstance(route, RouteData):
|
||||
if isinstance(route, RouteModel):
|
||||
route = route.route
|
||||
logger.attr('Route', route)
|
||||
try:
|
||||
@ -58,7 +63,14 @@ class RouteLoader(UI):
|
||||
raise ScriptError
|
||||
self.route_module = module
|
||||
|
||||
# Get route func
|
||||
# before_route()
|
||||
try:
|
||||
before_func_obj = self.route_obj.__getattribute__('before_route')
|
||||
except AttributeError:
|
||||
before_func_obj = empty_function
|
||||
before_func_obj()
|
||||
|
||||
# Run route
|
||||
try:
|
||||
func_obj = self.route_obj.__getattribute__(func)
|
||||
except AttributeError as e:
|
||||
@ -66,6 +78,11 @@ class RouteLoader(UI):
|
||||
logger.critical(f'Route class in {route} ({path}) does not have method {func}')
|
||||
raise ScriptError
|
||||
self.route_func = func
|
||||
|
||||
# Run
|
||||
func_obj()
|
||||
|
||||
# after_route()
|
||||
try:
|
||||
after_route_obj = self.route_obj.__getattribute__('after_route')
|
||||
except AttributeError:
|
||||
after_route_obj = empty_function
|
||||
after_route_obj()
|
||||
|
||||
12
tasks/map/route/model.py
Normal file
12
tasks/map/route/model.py
Normal file
@ -0,0 +1,12 @@
|
||||
from pydantic import BaseModel, RootModel
|
||||
|
||||
|
||||
class RouteModel(BaseModel):
|
||||
name: str
|
||||
route: str
|
||||
plane: str
|
||||
floor: str
|
||||
position: tuple[float, float]
|
||||
|
||||
|
||||
RouteListModel = RootModel[list[RouteModel]]
|
||||
@ -1,37 +1,37 @@
|
||||
from tasks.map.route.base import RouteData
|
||||
from tasks.map.route.model import RouteModel
|
||||
|
||||
# This file was auto-generated, do not modify it manually. To generate:
|
||||
# ``` python -m dev_tools.route_extract ```
|
||||
|
||||
ForgottenHallStage1__route = RouteData(
|
||||
ForgottenHallStage1__route = RouteModel(
|
||||
name='ForgottenHallStage1__route',
|
||||
route='route.daily.ForgottenHallStage1:route',
|
||||
plane='Jarilo_BackwaterPass',
|
||||
floor='F1',
|
||||
position=(369.4, 643.4),
|
||||
)
|
||||
HimekoTrial__route_item_enemy = RouteData(
|
||||
HimekoTrial__route_item_enemy = RouteModel(
|
||||
name='HimekoTrial__route_item_enemy',
|
||||
route='route.daily.HimekoTrial:route_item_enemy',
|
||||
plane='Jarilo_BackwaterPass',
|
||||
floor='F1',
|
||||
position=(519.9, 361.5),
|
||||
)
|
||||
HimekoTrial__route_item = RouteData(
|
||||
HimekoTrial__route_item = RouteModel(
|
||||
name='HimekoTrial__route_item',
|
||||
route='route.daily.HimekoTrial:route_item',
|
||||
plane='Jarilo_BackwaterPass',
|
||||
floor='F1',
|
||||
position=(519.9, 361.5),
|
||||
)
|
||||
HimekoTrial__route_enemy = RouteData(
|
||||
HimekoTrial__route_enemy = RouteModel(
|
||||
name='HimekoTrial__route_enemy',
|
||||
route='route.daily.HimekoTrial:route_enemy',
|
||||
plane='Jarilo_BackwaterPass',
|
||||
floor='F1',
|
||||
position=(519.9, 361.5),
|
||||
)
|
||||
HimekoTrial__exit = RouteData(
|
||||
HimekoTrial__exit = RouteModel(
|
||||
name='HimekoTrial__exit',
|
||||
route='route.daily.HimekoTrial:exit',
|
||||
plane='Jarilo_BackwaterPass',
|
||||
|
||||
152
tasks/rogue/route/base.py
Normal file
152
tasks/rogue/route/base.py
Normal file
@ -0,0 +1,152 @@
|
||||
from module.logger import logger
|
||||
from tasks.map.control.waypoint import ensure_waypoints
|
||||
from tasks.map.route.base import RouteBase as RouteBase_
|
||||
from tasks.rogue.bleesing.blessing import RogueBlessingSelector
|
||||
from tasks.rogue.bleesing.bonus import RogueBonusSelector
|
||||
from tasks.rogue.bleesing.curio import RogueCurioSelector
|
||||
from tasks.rogue.bleesing.ui import RogueUI
|
||||
from tasks.rogue.route.exit import RogueExit
|
||||
|
||||
|
||||
class RouteBase(RouteBase_, RogueUI, RogueExit):
|
||||
registered_domain_exit = None
|
||||
|
||||
def combat_expected_end(self):
|
||||
if self.is_page_choose_blessing():
|
||||
logger.info('Combat ended at is_page_choose_blessing()')
|
||||
return True
|
||||
if self.is_page_choose_curio():
|
||||
logger.info('Combat ended at is_page_choose_curio()')
|
||||
return True
|
||||
if self.is_page_choose_bonus():
|
||||
logger.info('Combat ended at is_page_choose_bonus()')
|
||||
return True
|
||||
|
||||
return False
|
||||
|
||||
def combat_execute(self, expected_end=None):
|
||||
return super().combat_execute(expected_end=self.combat_expected_end)
|
||||
|
||||
def clear_blessing(self, skip_first_screenshot=True):
|
||||
"""
|
||||
Pages:
|
||||
in: combat_expected_end()
|
||||
out: is_in_main()
|
||||
"""
|
||||
logger.info(f'Clear blessing')
|
||||
while 1:
|
||||
if skip_first_screenshot:
|
||||
skip_first_screenshot = False
|
||||
else:
|
||||
self.device.screenshot()
|
||||
|
||||
# End
|
||||
if self.is_in_main():
|
||||
logger.info(f'clear_blessing() ended at page_main')
|
||||
break
|
||||
|
||||
if self.is_page_choose_blessing():
|
||||
logger.hr('Choose blessing', level=2)
|
||||
selector = RogueBlessingSelector(self)
|
||||
selector.recognize_and_select()
|
||||
if self.is_page_choose_curio():
|
||||
logger.hr('Choose curio', level=2)
|
||||
selector = RogueCurioSelector(self)
|
||||
selector.recognize_and_select()
|
||||
if self.is_page_choose_bonus():
|
||||
logger.hr('Choose bonus', level=2)
|
||||
selector = RogueBonusSelector(self)
|
||||
selector.recognize_and_select()
|
||||
|
||||
"""
|
||||
Additional rogue methods
|
||||
"""
|
||||
|
||||
def clear_enemy(self, *waypoints):
|
||||
logger.hr('Clear enemy', level=1)
|
||||
result = super().clear_enemy(*waypoints)
|
||||
|
||||
self.clear_blessing()
|
||||
return result
|
||||
|
||||
def clear_elite(self, *waypoints):
|
||||
logger.hr('Clear elite', level=1)
|
||||
waypoints = ensure_waypoints(waypoints)
|
||||
end_point = waypoints[-1]
|
||||
end_point.speed = 'run_2x'
|
||||
|
||||
# Use skill
|
||||
pass
|
||||
|
||||
result = super().clear_enemy(*waypoints)
|
||||
|
||||
self.clear_blessing()
|
||||
return result
|
||||
|
||||
def clear_event(self, *waypoints):
|
||||
"""
|
||||
Handle an event in DomainOccurrence, DomainEncounter, DomainTransaction
|
||||
"""
|
||||
logger.hr('Clear event', level=1)
|
||||
|
||||
result = self.goto(*waypoints)
|
||||
return result
|
||||
|
||||
def domain_reward(self, *waypoints):
|
||||
"""
|
||||
Get reward of the DomainElite and DomainBoss
|
||||
"""
|
||||
logger.hr('Clear reward', level=1)
|
||||
|
||||
# Skip if not going to get reward
|
||||
pass
|
||||
|
||||
result = self.goto(*waypoints)
|
||||
return result
|
||||
|
||||
def domain_herta(self, *waypoints):
|
||||
"""
|
||||
Most people don't buy herta shop, skip
|
||||
"""
|
||||
pass
|
||||
|
||||
def domain_single_exit(self, *waypoints):
|
||||
"""
|
||||
Goto a single exit, exit current domain
|
||||
end_rotation is not required
|
||||
"""
|
||||
logger.hr('Domain single exit', level=1)
|
||||
waypoints = ensure_waypoints(waypoints)
|
||||
result = self.goto(*waypoints)
|
||||
|
||||
self.domain_exit_interact()
|
||||
return result
|
||||
|
||||
def domain_exit(self, *waypoints, end_rotation=None):
|
||||
logger.hr('Domain exit', level=1)
|
||||
waypoints = ensure_waypoints(waypoints)
|
||||
end_point = waypoints[-1]
|
||||
end_point.end_rotation = end_rotation
|
||||
result = self.goto(*waypoints)
|
||||
|
||||
return result
|
||||
|
||||
"""
|
||||
Route
|
||||
"""
|
||||
|
||||
def register_domain_exit(self, *waypoints, end_rotation=None):
|
||||
"""
|
||||
Register an exit, call `domain_exit()` at route end
|
||||
"""
|
||||
self.registered_domain_exit = (waypoints, end_rotation)
|
||||
|
||||
def before_route(self):
|
||||
self.registered_domain_exit = None
|
||||
|
||||
def after_route(self):
|
||||
if self.registered_domain_exit is not None:
|
||||
waypoints, end_rotation = self.registered_domain_exit
|
||||
self.domain_exit(*waypoints, end_rotation=end_rotation)
|
||||
else:
|
||||
logger.info('No domain exit registered')
|
||||
151
tasks/rogue/route/loader.py
Normal file
151
tasks/rogue/route/loader.py
Normal file
@ -0,0 +1,151 @@
|
||||
from typing import Optional
|
||||
|
||||
from module.base.decorator import cached_property
|
||||
from module.logger import logger
|
||||
from tasks.base.main_page import MainPage
|
||||
from tasks.map.keywords import MapPlane
|
||||
from tasks.map.keywords.plane import (
|
||||
Herta_MasterControlZone,
|
||||
Herta_ParlorCar,
|
||||
Jarilo_AdministrativeDistrict,
|
||||
Luofu_AurumAlley,
|
||||
Luofu_ExaltingSanctum
|
||||
)
|
||||
from tasks.map.minimap.minimap import Minimap
|
||||
from tasks.map.route.loader import RouteLoader as RouteLoader_
|
||||
from tasks.rogue.route.base import RouteBase
|
||||
from tasks.rogue.route.model import RogueRouteListModel, RogueRouteModel
|
||||
|
||||
|
||||
def model_from_json(model, file: str):
|
||||
with open(file, 'r', encoding='utf-8') as f:
|
||||
content = f.read()
|
||||
data = model.model_validate_json(content)
|
||||
return data
|
||||
|
||||
|
||||
class RouteLoader(RouteLoader_, MainPage):
|
||||
@cached_property
|
||||
def all_minimap(self) -> dict[(str, str), Minimap]:
|
||||
"""
|
||||
Returns:
|
||||
dict: Key: {world}_{plane}_{floor}, e.g. Jarilo_SilvermaneGuardRestrictedZone_F1
|
||||
Value: Minimap object
|
||||
"""
|
||||
# No enemy spawn at the followings
|
||||
blacklist = [
|
||||
Herta_ParlorCar,
|
||||
Herta_MasterControlZone,
|
||||
Jarilo_AdministrativeDistrict,
|
||||
Luofu_ExaltingSanctum,
|
||||
Luofu_AurumAlley,
|
||||
]
|
||||
maps = {}
|
||||
for plane in MapPlane.instances.values():
|
||||
if plane in blacklist:
|
||||
continue
|
||||
if not plane.world:
|
||||
continue
|
||||
for floor in plane.floors:
|
||||
minimap = Minimap()
|
||||
minimap.set_plane(plane=plane, floor=floor)
|
||||
maps[f'{plane.name}_{floor}'] = minimap
|
||||
return maps
|
||||
|
||||
@cached_property
|
||||
def all_route(self) -> list[RogueRouteModel]:
|
||||
return model_from_json(RogueRouteListModel, './route/rogue/route.json').root
|
||||
|
||||
def get_minimap(self, route: RogueRouteModel):
|
||||
return self.all_minimap[route.plane_floor]
|
||||
|
||||
def position_find_known(self, image) -> Optional[RogueRouteModel]:
|
||||
"""
|
||||
Try to find from known route spawn point
|
||||
"""
|
||||
logger.info('position_find_known')
|
||||
plane = self.get_plane()
|
||||
if plane is None:
|
||||
logger.warning('Unknown rogue domain')
|
||||
return
|
||||
|
||||
visited = []
|
||||
for route in self.all_route:
|
||||
if plane.rogue_domain and plane.rogue_domain != route.domain:
|
||||
if plane.rogue_domain == 'Transaction' and route.is_DomainOccurrence:
|
||||
# Treat "Transaction" as "Occurrence"
|
||||
pass
|
||||
elif plane.rogue_domain == 'Encounter' and route.is_DomainOccurrence:
|
||||
# Treat "Encounter" as "Occurrence"
|
||||
pass
|
||||
else:
|
||||
continue
|
||||
minimap = self.get_minimap(route)
|
||||
minimap.init_position(route.position, show_log=False)
|
||||
try:
|
||||
minimap.update_position(image)
|
||||
except FileNotFoundError:
|
||||
continue
|
||||
visited.append((route, minimap.position_similarity))
|
||||
|
||||
if len(visited) < 3:
|
||||
logger.warning('Too few routes to search from, not enough to make a prediction')
|
||||
return
|
||||
|
||||
visited = sorted(visited, key=lambda x: x[1], reverse=True)
|
||||
logger.info(f'Best 3 prediction: {[(r.name, s) for r, s in visited[:3]]}')
|
||||
if visited[1][1] / visited[0][1] > 0.75:
|
||||
logger.warning('Similarity too close, not enough to make a prediction')
|
||||
return
|
||||
|
||||
logger.attr('RoutePredict', visited[0][0].name)
|
||||
return visited[0][0]
|
||||
|
||||
def position_find_bruteforce(self, image) -> Minimap:
|
||||
"""
|
||||
Fallback method to find from all planes and floors
|
||||
"""
|
||||
logger.warning('position_find_bruteforce, this may take a while')
|
||||
for name, minimap in self.all_minimap.items():
|
||||
minimap.init_position((0, 0), show_log=False)
|
||||
try:
|
||||
minimap.update_position(image)
|
||||
except FileNotFoundError:
|
||||
pass
|
||||
|
||||
def get_name(minimap_: Minimap) -> str:
|
||||
return f'{minimap_.plane.name}_{minimap_.floor}_X{int(minimap_.position[0])}Y{int(minimap_.position[1])}'
|
||||
|
||||
visited = sorted(self.all_minimap.values(), key=lambda x: x.position_similarity, reverse=True)
|
||||
logger.info(f'Best 5 prediction: {[(get_name(m), m.position_similarity) for m in visited[:5]]}')
|
||||
if visited[1].position_similarity / visited[0].position_similarity > 0.75:
|
||||
logger.warning('Similarity too close, prediction may goes wrong')
|
||||
|
||||
logger.attr('RoutePredict', get_name(visited[0]))
|
||||
return visited[0]
|
||||
|
||||
def route_run(self, route=None):
|
||||
"""
|
||||
Run a rogue domain
|
||||
|
||||
Pages:
|
||||
in: page_main
|
||||
out: page_main, at another domain
|
||||
"""
|
||||
route = self.position_find_known(self.device.image)
|
||||
if route is not None:
|
||||
super().route_run(route)
|
||||
else:
|
||||
self.position_find_bruteforce(self.device.image)
|
||||
logger.error('New route detected, please record it')
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
self = RouteLoader('src', task='Rogue')
|
||||
# self.image_file = r''
|
||||
# self.position_find_bruteforce(self.device.image)
|
||||
|
||||
self.device.screenshot()
|
||||
base = RouteBase(config=self.config, device=self.device, task='Rogue')
|
||||
base.clear_blessing()
|
||||
self.route_run()
|
||||
135
tasks/rogue/route/model.py
Normal file
135
tasks/rogue/route/model.py
Normal file
@ -0,0 +1,135 @@
|
||||
from pydantic import BaseModel, RootModel
|
||||
|
||||
from tasks.map.route.model import RouteModel
|
||||
|
||||
|
||||
class RogueWaypointModel(BaseModel):
|
||||
"""
|
||||
{
|
||||
"domain": "Combat",
|
||||
"route": "Herta_StorageZone_F1_X252Y84",
|
||||
"waypoint": "spawn",
|
||||
"index": 0,
|
||||
"file": "./screenshots/rogue/Combat/Herta_StorageZone_F1_X252Y84/route/spawn.png",
|
||||
"plane": "Herta_StorageZone",
|
||||
"floor": "F1",
|
||||
"position": [
|
||||
252.8,
|
||||
84.8
|
||||
],
|
||||
"direction": 300.1,
|
||||
"rotation": 299
|
||||
},
|
||||
"""
|
||||
domain: str
|
||||
route: str
|
||||
waypoint: str
|
||||
index: int
|
||||
|
||||
file: str
|
||||
|
||||
plane: str
|
||||
floor: str
|
||||
position: tuple[float, float]
|
||||
direction: float
|
||||
rotation: int
|
||||
|
||||
@property
|
||||
def plane_floor(self):
|
||||
return f'{self.plane}_{self.floor}'
|
||||
|
||||
@property
|
||||
def positionXY(self):
|
||||
x, y = int(self.position[0]), int(self.position[1])
|
||||
return f'X{x}Y{y}'
|
||||
|
||||
@property
|
||||
def is_spawn(self) -> bool:
|
||||
return self.waypoint.startswith('spawn')
|
||||
|
||||
@property
|
||||
def is_exit(self) -> bool:
|
||||
return self.waypoint.startswith('exit')
|
||||
|
||||
@property
|
||||
def is_DomainBoss(self):
|
||||
return self.domain == 'Boss'
|
||||
|
||||
@property
|
||||
def is_DomainCombat(self):
|
||||
return self.domain == 'Combat'
|
||||
|
||||
@property
|
||||
def is_DomainElite(self):
|
||||
return self.domain == 'Elite'
|
||||
|
||||
@property
|
||||
def is_DomainEncounter(self):
|
||||
return self.domain == 'Encounter'
|
||||
|
||||
@property
|
||||
def is_DomainOccurrence(self):
|
||||
return self.domain == 'Occurrence'
|
||||
|
||||
@property
|
||||
def is_DomainRespite(self):
|
||||
return self.domain == 'Respite'
|
||||
|
||||
@property
|
||||
def is_DomainTransaction(self):
|
||||
return self.domain == 'Transaction'
|
||||
|
||||
|
||||
RogueWaypointListModel = RootModel[list[RogueWaypointModel]]
|
||||
|
||||
|
||||
class RogueRouteModel(RouteModel):
|
||||
"""
|
||||
{
|
||||
"name": "Boss_Luofu_ArtisanshipCommission_F1_X506Y495",
|
||||
"domain": "Boss",
|
||||
"route": "route.rogue.Boss.Luofu_ArtisanshipCommission_F1:Luofu_ArtisanshipCommission_F1_X506Y495",
|
||||
"plane": "Luofu_ArtisanshipCommission",
|
||||
"floor": "F1",
|
||||
"position": [
|
||||
506.0,
|
||||
495.4
|
||||
]
|
||||
},
|
||||
"""
|
||||
domain: str
|
||||
|
||||
@property
|
||||
def plane_floor(self):
|
||||
return f'{self.plane}_{self.floor}'
|
||||
|
||||
@property
|
||||
def is_DomainBoss(self):
|
||||
return self.domain == 'Boss'
|
||||
|
||||
@property
|
||||
def is_DomainCombat(self):
|
||||
return self.domain == 'Combat'
|
||||
|
||||
@property
|
||||
def is_DomainElite(self):
|
||||
return self.domain == 'Elite'
|
||||
|
||||
@property
|
||||
def is_DomainEncounter(self):
|
||||
return self.domain == 'Encounter'
|
||||
|
||||
@property
|
||||
def is_DomainOccurrence(self):
|
||||
return self.domain == 'Occurrence'
|
||||
|
||||
@property
|
||||
def is_DomainRespite(self):
|
||||
return self.domain == 'Respite'
|
||||
|
||||
@property
|
||||
def is_DomainTransaction(self):
|
||||
return self.domain == 'Transaction'
|
||||
|
||||
|
||||
RogueRouteListModel = RootModel[list[RogueRouteModel]]
|
||||
Loading…
Reference in New Issue
Block a user