mirror of
https://github.com/LmeSzinc/StarRailCopilot.git
synced 2024-12-12 07:29:03 +00:00
383 lines
15 KiB
Python
383 lines
15 KiB
Python
from typing import Optional
|
|
|
|
import numpy as np
|
|
|
|
from module.base.decorator import cached_property
|
|
from module.base.timer import Timer
|
|
from module.exception import GameStuckError, HandledError
|
|
from module.logger import logger, save_error_log
|
|
from tasks.base.assets.assets_base_main_page import ROGUE_LEAVE_FOR_NOW
|
|
from tasks.base.assets.assets_base_page import MAP_EXIT
|
|
from tasks.character.switch import CharacterSwitch
|
|
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.resource.resource import SPECIAL_PLANES
|
|
from tasks.map.route.loader import RouteLoader as RouteLoader_
|
|
from tasks.rogue.assets.assets_rogue_ui import BLESSING_CONFIRM
|
|
from tasks.rogue.assets.assets_rogue_weekly import ROGUE_REPORT
|
|
from tasks.rogue.blessing.ui import RogueUI
|
|
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 MinimapWrapper:
|
|
@cached_property
|
|
def all_minimap(self) -> dict[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, floor in SPECIAL_PLANES:
|
|
minimap = Minimap()
|
|
minimap.set_plane(plane=plane, floor=floor)
|
|
maps[f'{plane}_{floor}'] = minimap
|
|
|
|
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
|
|
|
|
logger.attr('MinimapLoaded', len(maps))
|
|
return maps
|
|
|
|
@cached_property
|
|
def all_route(self) -> list[RogueRouteModel]:
|
|
routes = model_from_json(RogueRouteListModel, './route/rogue/route.json').root
|
|
logger.attr('RouteLoaded', len(routes))
|
|
return routes
|
|
|
|
def get_minimap(self, route: RogueRouteModel):
|
|
return self.all_minimap[route.plane_floor]
|
|
|
|
|
|
class RouteLoader(RouteBase, MinimapWrapper, RouteLoader_, CharacterSwitch):
|
|
def position_find_known(self, image, force_return=False) -> Optional[RogueRouteModel]:
|
|
"""
|
|
Try to find from known route spawn point
|
|
"""
|
|
logger.info('position_find_known')
|
|
plane = self.update_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.is_rogue_occurrence and route.is_DomainOccurrence:
|
|
# Treat as "Occurrence"
|
|
pass
|
|
elif plane.is_rogue_elite and route.is_DomainElite:
|
|
# Treat as "Elite"
|
|
pass
|
|
else:
|
|
continue
|
|
minimap = self.get_minimap(route)
|
|
minimap.init_position(route.position, show_log=False)
|
|
try:
|
|
minimap.update_position(image)
|
|
except FileNotFoundError as e:
|
|
logger.warning(e)
|
|
continue
|
|
visited.append((route, minimap.position_similarity, minimap.position))
|
|
|
|
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 predictions: {[(r.name, s, p) for r, s, p in visited[:3]]}')
|
|
nearby = [
|
|
(r, s, p) for r, s, p in visited if np.linalg.norm(np.subtract(r.position, p)) < 5
|
|
]
|
|
logger.info(f'Best 3 nearby predictions: {[(r.name, s, p) for r, s, p in nearby[:3]]}')
|
|
# Check special
|
|
for r, s, p in nearby:
|
|
if self._position_match_special(r, s, p):
|
|
logger.info(f'Route match special: {r.name}')
|
|
return r
|
|
# Check nearby
|
|
if len(nearby) == 1:
|
|
if nearby[0][1] > 0.05:
|
|
logger.attr('RoutePredict', nearby[0][0].name)
|
|
return nearby[0][0]
|
|
elif len(nearby) >= 2:
|
|
if nearby[0][1] / nearby[1][1] > 0.55:
|
|
logger.attr('RoutePredict', nearby[0][0].name)
|
|
return nearby[0][0]
|
|
|
|
# logger.info(f'Best 3 prediction: {[(r.name, s, p) for r, s, p in visited[:3]]}')
|
|
# if visited[0][1] / visited[1][1] > 0.75:
|
|
# logger.attr('RoutePredict', visited[0][0].name)
|
|
# return visited[0][0]
|
|
|
|
if force_return:
|
|
if len(nearby) >= 1:
|
|
route = nearby[0][0]
|
|
else:
|
|
route = visited[0][0]
|
|
logger.attr('RoutePredict', route.name)
|
|
return route
|
|
else:
|
|
logger.warning('Similarity too close, not enough to make a prediction')
|
|
return None
|
|
|
|
def _position_match_special(
|
|
self,
|
|
route: RogueRouteModel,
|
|
similarity: float,
|
|
position: tuple[float, float]
|
|
) -> bool:
|
|
# 2023-11-13 23:50:00.470 | INFO | Best 3 predictions: [
|
|
# ('Occurrence_Luofu_Cloudford_F1_X241Y947', 0.119, (272.6, 948.5)),
|
|
# ('Occurrence_Jarilo_GreatMine_F1_X277Y605', 0.107, (264.4, 645.0)),
|
|
# ('Occurrence_Luofu_ArtisanshipCommission_F1_X521Y63', 0.106, (569.8, 34.8))]
|
|
# 2023-11-13 23:50:00.472 | INFO | Best 3 nearby predictions: [
|
|
# ('Occurrence_Herta_SupplyZone_F2Rogue_X397Y223', 0.102, (393.2, 222.8)),
|
|
# ('Occurrence_Herta_StorageZone_F2_X365Y167', 0.094, (363.0, 166.8)),
|
|
# ('Occurrence_Herta_StorageZone_F2_X363Y166', 0.094, (363.0, 166.8))]
|
|
# if route.name == 'Occurrence_Herta_StorageZone_F2_X363Y166' and similarity > 0.05:
|
|
# return True
|
|
|
|
# Before Combat_Herta_SupplyZone_F2_X45Y369
|
|
if route.name in [
|
|
'Combat_Herta_SupplyZone_F2_X543Y255', # 0.462, (543.3, 255.4)
|
|
'Combat_Luofu_DivinationCommission_F1_X737Y237',
|
|
# ('Occurrence_Luofu_Cloudford_F1_X241Y947', 0.307, (236.5, 949.6)),
|
|
# ('Occurrence_Luofu_Cloudford_F1_X244Y951', 0.307, (236.5, 949.6)),
|
|
# ('Occurrence_Jarilo_SilvermaneGuardRestrictedZone_F1_X509Y541', 0.154, (507.8, 515.2))
|
|
'Occurrence_Luofu_Cloudford_F1_X241Y947',
|
|
'Occurrence_Luofu_Cloudford_F1_X244Y951',
|
|
] and similarity > 0.25:
|
|
return True
|
|
# Before Combat_Luofu_Cloudford_F1_X281Y873
|
|
if route.name in [
|
|
# ('Combat_Jarilo_BackwaterPass_F1_X507Y733', 0.26, (503.2, 736.9)),
|
|
# ('Combat_Herta_SupplyZone_F2_X45Y369', 0.168, (46.5, 370.0))
|
|
'Jarilo_BackwaterPass_F1_X507Y733',
|
|
'Jarilo_BackwaterPass_F1_X555Y643',
|
|
'Occurrence_Jarilo_BackwaterPass_F1_X553Y643',
|
|
'Combat_Jarilo_GreatMine_F1_X545Y513',
|
|
'Combat_Herta_SupplyZone_F2_X45Y369',
|
|
] and similarity > 0.20:
|
|
return True
|
|
# Before Occurrence_Luofu_DivinationCommission_F2_X425Y791
|
|
if route.name in [
|
|
'Occurrence_Jarilo_RivetTown_F1_X157Y435',
|
|
# ('Occurrence_Luofu_DivinationCommission_F2_X149Y659', 0.237, (148.9, 658.8)),
|
|
# ('Occurrence_Luofu_DivinationCommission_F2_X425Y791', 0.11, (425.2, 793.8))
|
|
'Occurrence_Luofu_DivinationCommission_F2_X149Y659',
|
|
# ('Combat_Luofu_DivinationCommission_F1_X97Y457', 0.222, (97.8, 456.9)),
|
|
# ('Combat_Luofu_ScalegorgeWaterscape_F1_X415Y261', 0.112, (371.8, 289.4)),
|
|
# ('Combat_Herta_SupplyZone_F2_X45Y369', 0.104, (11.7, 367.6))
|
|
'Combat_Luofu_DivinationCommission_F1_X97Y457',
|
|
# ('Occurrence_Jarilo_BackwaterPass_F1_X613Y755', 0.206, (611.3, 759.0)),
|
|
# ('Occurrence_Jarilo_BackwaterPass_F1_X611Y761', 0.206, (611.3, 759.0)),
|
|
# ('Occurrence_Luofu_DivinationCommission_F2_X425Y791', 0.105, (429.7, 791.6))
|
|
'Occurrence_Jarilo_BackwaterPass_F1_X613Y755',
|
|
'Occurrence_Jarilo_BackwaterPass_F1_X611Y761',
|
|
] and similarity > 0.15:
|
|
return True
|
|
if route.name in [
|
|
'Combat_Herta_StorageZone_F1_X273Y92',
|
|
'Occurrence_Herta_StorageZone_F1_X273Y93',
|
|
'Occurrence_Jarilo_RivetTown_F1_X289Y97',
|
|
'Occurrence_Luofu_DivinationCommission_F2_X425Y791',
|
|
'Occurrence_Luofu_ArtisanshipCommission_F1_X169Y491',
|
|
] and similarity > 0.1:
|
|
return True
|
|
# Luofu_Cloudford_F1_X283Y865 and its equivalents
|
|
# INFO 21:27:00.816 │ Best 3 nearby predictions: [
|
|
# ('Combat_Herta_SupplyZone_F2_X45Y369', 0.184, (41.0, 369.1)),
|
|
# ('Combat_Luofu_Cloudford_F1_X281Y873', 0.149, (281.8, 869.6)),
|
|
# ('Combat_Luofu_Cloudford_F1_X283Y865', 0.149, (281.8, 869.6))]
|
|
# INFO | Best 3 predictions: [('Combat_Herta_SupplyZone_F2_X45Y369', 0.149, (43.4, 369.3)),
|
|
# ('Combat_Luofu_Cloudford_F1_X241Y947', 0.138, (198.6, 956.8)),
|
|
# ('Combat_Luofu_Cloudford_F1Rogue_X59Y405', 0.134, (81.0, 397.4))]
|
|
if route.name in [
|
|
'Combat_Luofu_Cloudford_F1_X283Y865',
|
|
'Occurrence_Luofu_Cloudford_F1_X283Y865',
|
|
'Combat_Luofu_Cloudford_F1_X281Y873',
|
|
'Occurrence_Luofu_Cloudford_F1_X281Y873',
|
|
] and similarity > 0.05:
|
|
return True
|
|
return False
|
|
|
|
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():
|
|
if minimap.is_special_plane:
|
|
continue
|
|
|
|
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[:50]]}')
|
|
if visited[1].position_similarity / visited[0].position_similarity > 0.75:
|
|
logger.warning('Similarity too close, predictions may go wrong')
|
|
|
|
logger.attr('RoutePredict', get_name(visited[0]))
|
|
return visited[0]
|
|
|
|
def position_find(self, skip_first_screenshot=True):
|
|
timeout = Timer(1, count=3).start()
|
|
while 1:
|
|
if skip_first_screenshot:
|
|
skip_first_screenshot = False
|
|
else:
|
|
self.device.screenshot()
|
|
|
|
if timeout.reached():
|
|
self.position_find_bruteforce(self.device.image)
|
|
logger.warning('Find position timeout, force return route')
|
|
return self.position_find_known(self.device.screenshot(), force_return=True)
|
|
|
|
route = self.position_find_known(self.device.image)
|
|
if route is not None:
|
|
return route
|
|
|
|
def rogue_leave(self, skip_first_screenshot=True):
|
|
logger.hr('Rogue leave', level=1)
|
|
while 1:
|
|
if skip_first_screenshot:
|
|
skip_first_screenshot = False
|
|
else:
|
|
self.device.screenshot()
|
|
|
|
# End
|
|
if self.is_page_rogue_main():
|
|
logger.info('Rogue left')
|
|
break
|
|
|
|
# Re-enter
|
|
if self.handle_combat_interact():
|
|
continue
|
|
# From ui_leave_special
|
|
if self.is_in_map_exit(interval=2):
|
|
self.device.click(MAP_EXIT)
|
|
continue
|
|
if self.handle_popup_confirm():
|
|
continue
|
|
if self.appear_then_click(ROGUE_LEAVE_FOR_NOW, interval=2):
|
|
continue
|
|
# Blessing
|
|
if self.handle_blessing():
|
|
continue
|
|
# _domain_exit_wait_next()
|
|
if self.match_template_color(ROGUE_REPORT, interval=2):
|
|
logger.info(f'{ROGUE_REPORT} -> {BLESSING_CONFIRM}')
|
|
self.device.click(BLESSING_CONFIRM)
|
|
continue
|
|
if self.handle_reward():
|
|
continue
|
|
if self.handle_get_character():
|
|
continue
|
|
|
|
def route_run(self, route=None):
|
|
"""
|
|
Run a rogue domain
|
|
|
|
Returns:
|
|
bool: True if success, False if route unknown
|
|
|
|
Pages:
|
|
in: page_main
|
|
out: page_main, at another domain
|
|
or page_rogue if rogue cleared
|
|
"""
|
|
# To have a newer image, since previous loadings took some time
|
|
route = self.position_find(skip_first_screenshot=False)
|
|
self.screenshot_tracking_add()
|
|
|
|
try:
|
|
super().route_run(route)
|
|
return True
|
|
except GameStuckError as e:
|
|
logger.error(e)
|
|
save_error_log(config=self.config, device=self.device)
|
|
self.rogue_leave()
|
|
raise HandledError('Rogue run failed')
|
|
|
|
def rogue_run(self, skip_first_screenshot=True):
|
|
"""
|
|
Do a complete rogue run, no error handle yet.
|
|
|
|
Pages:
|
|
in: page_rogue, is_page_rogue_launch()
|
|
out: page_rogue, is_page_rogue_main()
|
|
"""
|
|
base = RouteBase(config=self.config, device=self.device, task=self.config.task.command)
|
|
count = 1
|
|
self.character_is_ranged = None
|
|
while 1:
|
|
if skip_first_screenshot:
|
|
skip_first_screenshot = False
|
|
else:
|
|
self.device.screenshot()
|
|
|
|
logger.hr(f'Route run: {count}', level=1)
|
|
base.clear_blessing()
|
|
self.character_switch_to_ranged(update=True)
|
|
|
|
self.route_run()
|
|
# if not success:
|
|
# self.device.image_save()
|
|
# continue
|
|
|
|
# End
|
|
if self.is_page_rogue_main():
|
|
break
|
|
|
|
count += 1
|
|
|
|
|
|
if __name__ == '__main__':
|
|
self = RouteLoader('src', task='Rogue')
|
|
# self.image_file = r''
|
|
self.device.screenshot()
|
|
self.position_find()
|
|
self.position_find_bruteforce(self.device.image)
|
|
|
|
# self.device.screenshot()
|
|
# self.rogue_run()
|