import datetime from module.config.utils import get_server_next_update, get_server_last_update from module.exercise.assets import * from module.exercise.combat import ExerciseCombat from module.logger import logger from module.ocr.ocr import Digit, Ocr, OcrYuv from module.ui.page import page_exercise from module.smart_mgmt.utils import parse_downtime class DatedDuration(Ocr): def __init__(self, buttons, lang='cnocr', letter=(255, 255, 255), threshold=128, alphabet='0123456789:IDS天日d', name=None): super().__init__(buttons, lang=lang, letter=letter, threshold=threshold, alphabet=alphabet, name=name) def after_process(self, result): result = super().after_process(result) result = result.replace('I', '1').replace('D', '0').replace('S', '5') return result def ocr(self, image, direct_ocr=False): """ Do OCR on a dated duration, such as `10d 01:30:30` or `7日01:30:30`. Args: image: direct_ocr: Returns: list, datetime.timedelta: timedelta object, or a list of it. """ result_list = super().ocr(image, direct_ocr=direct_ocr) if not isinstance(result_list, list): result_list = [result_list] result_list = [self.parse_time(result) for result in result_list] if len(self.buttons) == 1: result_list = result_list[0] return result_list @staticmethod def parse_time(string): """ Args: string (str): `10d 01:30:30` or `7日01:30:30` Returns: datetime.timedelta: """ import re result = re.search(r'(\d{1,2})\D?(\d{1,2}):?(\d{2}):?(\d{2})', string) if result: result = [int(s) for s in result.groups()] return datetime.timedelta(days=result[0], hours=result[1], minutes=result[2], seconds=result[3]) else: logger.warning(f'Invalid dated duration: {string}') return datetime.timedelta(days=0, hours=0, minutes=0, seconds=0) class DatedDurationYuv(DatedDuration, OcrYuv): pass OCR_EXERCISE_REMAIN = Digit(OCR_EXERCISE_REMAIN, letter=(173, 247, 74), threshold=128) OCR_PERIOD_REMAIN = DatedDuration(OCR_PERIOD_REMAIN, letter=(255, 255, 255), threshold=128) ADMIRAL_TRIAL_HOUR_INTERVAL = { # "aggressive": [336, 0] "sun18": [6, 0], "sun12": [12, 6], "sun0": [24, 12], "sat18": [30, 24], "sat12": [36, 30], "sat0": [48, 36], "fri18": [56, 48] } class Exercise(ExerciseCombat): opponent_change_count = 0 remain = 0 preserve = 0 DOWNTIME_END_HOUR_THRESHOLD = 15 def _new_opponent(self): logger.info('New opponent') self.appear_then_click(NEW_OPPONENT) self.opponent_change_count += 1 logger.attr("Change_opponent_count", self.opponent_change_count) self.config.set_record(Exercise_OpponentRefreshValue=self.opponent_change_count) self.ensure_no_info_bar(timeout=3) def _opponent_fleet_check_all(self): if self.config.Exercise_OpponentChooseMode != 'leftmost': super()._opponent_fleet_check_all() def _opponent_sort(self, method=None): if method is None: method = self.config.Exercise_OpponentChooseMode if method != 'leftmost': return super()._opponent_sort(method=method) else: return [0, 1, 2, 3] def _exercise_once(self): """Execute exercise once. This method handles exercise refresh and exercise failure. Returns: bool: True if success to defeat one opponent. False if failed to defeat any opponent and refresh exhausted. """ self._opponent_fleet_check_all() while 1: for opponent in self._opponent_sort(): logger.hr(f'Opponent {opponent}', level=2) success = self._combat(opponent) if success: return success if self.opponent_change_count >= 5: return False self._new_opponent() self._opponent_fleet_check_all() def _exercise_easiest_else_exp(self): """Try easiest first, if unable to beat easiest opponent then switch to max exp opponent and accept the loss. This method handles exercise refresh and exercise failure. Returns: bool: True if success to defeat one opponent. False if failed to defeat any opponent and refresh exhausted. """ method = "easiest_else_exp" restore = self.config.Exercise_LowHpThreshold threshold = self.config.Exercise_LowHpThreshold self._opponent_fleet_check_all() while 1: opponents = self._opponent_sort(method=method) logger.hr(f'Opponent {opponents[0]}', level=2) self.config.override(Exercise_LowHpThreshold=threshold) success = self._combat(opponents[0]) if success: self.config.override(Exercise_LowHpThreshold=restore) return success else: if self.opponent_change_count < 5: logger.info("Cannot beat calculated easiest opponent, refresh") self._new_opponent() self._opponent_fleet_check_all() continue else: logger.info("Cannot beat calculated easiest opponent, MAX EXP then") method = "max_exp" threshold = 0 def _get_opponent_change_count(self): """ Same day, count set to last known change count or 6 i.e. no refresh New day, count set to 0 i.e. can change up to 5 times Returns: int: """ record = self.config.Exercise_OpponentRefreshRecord update = get_server_last_update('00:00') if record.date() == update.date(): # Same Day return self.config.Exercise_OpponentRefreshValue else: # New Day self.config.set_record(Exercise_OpponentRefreshValue=0) return 0 def _get_exercise_reset_remain(self): """ Returns: datetime.timedelta """ result = OCR_PERIOD_REMAIN.ocr(self.device.image) return result def _get_exercise_strategy(self): """ Returns: int: ExercisePreserve, X times to remain list, int: Admiral trial time period """ if self.config.Exercise_ExerciseStrategy == "aggressive": preserve = 0 admiral_interval = None else: preserve = 5 admiral_interval = ADMIRAL_TRIAL_HOUR_INTERVAL[self.config.Exercise_ExerciseStrategy] return preserve, admiral_interval def _get_thursday_strategy(self, remain_hours): """ Args: remain_hours(int): Returns: bool: if use Thursday strategy """ if 96 > remain_hours >= 84 or 264 > remain_hours >= 252: return True return False def _get_downtime_strategy(self, now): """ Check smart_mgmt downtime window to decide whether to clear attempts. Returns: bool: True if today is maintenance day with end time at or after DOWNTIME_END_HOUR_THRESHOLD. False if available but today is not maintenance day. None if unavailable (disabled, non-CN server, window missing, or parse failed). """ if self.config.SERVER != 'cn': return None if not self.config.cross_get('DowntimeMgmt.DowntimeStrategy.ManageExercise', False): return None window = self.config.cross_get('DowntimeMgmt.Downtime.Window', None) if not window: return None end_dt = parse_downtime(window) if end_dt is None: return None if end_dt <= now: return False return end_dt.date() == now.date() and end_dt.hour >= self.DOWNTIME_END_HOUR_THRESHOLD def _should_force_clear(self, now, remain_time, admiral_interval): """Decide whether to clear all exercise attempts this run.""" if admiral_interval is None or not remain_time: return False remain_hours = int(remain_time.total_seconds() // 3600) admiral_start, admiral_end = admiral_interval downtime_strategy = self._get_downtime_strategy(now) if downtime_strategy is True: logger.info('Downtime today, using all attempts before downtime') return True elif downtime_strategy is None: # Fallback to thursday strategy when smart_mgmt is unavailable if self._get_thursday_strategy(remain_hours): logger.info('Reach Thursday, using all attempts before downtime') return True if admiral_start > remain_hours >= admiral_end: logger.info('Reach set time for admiral trial, using all attempts') return True if remain_hours < 6: logger.info('Exercise period remain less than 6 hours, using all attempts') return True return False def _should_delay(self, now, next_update, forced_clear): """Check whether to delay exercise to the configured time before next update.""" if forced_clear: return False seconds_until_update = (next_update - now).seconds return seconds_until_update > 3600 * self.config.Exercise_DelayUntilHoursBeforeNextUpdate def _exercise(self, preserve): """Consume exercise attempts until remain <= preserve or refresh exhausted.""" while 1: self.remain = OCR_EXERCISE_REMAIN.ocr(self.device.image) if self.remain <= preserve: break logger.hr(f'Exercise remain {self.remain}', level=1) if self.config.Exercise_OpponentChooseMode == "easiest_else_exp": success = self._exercise_easiest_else_exp() else: success = self._exercise_once() if not success: logger.info('New opponent exhausted') break def run(self): self.ui_ensure(page_exercise) server_update = self.config.Scheduler_ServerUpdate self.opponent_change_count = self._get_opponent_change_count() logger.attr("Change_opponent_count", self.opponent_change_count) logger.attr('Exercise_ExerciseStrategy', self.config.Exercise_ExerciseStrategy) self.preserve, admiral_interval = self._get_exercise_strategy() now = datetime.datetime.now() remain_time = self._get_exercise_reset_remain() logger.info(f'Exercise period remain: {remain_time}') forced_clear = self._should_force_clear(now, remain_time, admiral_interval) if forced_clear: self.preserve = 0 logger.attr("exercise preserve", self.preserve) next_update = get_server_next_update(server_update) should_delay = self._should_delay(now, next_update, forced_clear) if should_delay: logger.warning(f'Exercise should run at {self.config.Exercise_DelayUntilHoursBeforeNextUpdate} ' f'hours before next update. Delay task to it.') else: self._exercise(self.preserve) # self.equipment_take_off_when_finished() # Scheduler with self.config.multi_set(): self.config.set_record(Exercise_OpponentRefreshValue=self.opponent_change_count) if self.remain <= self.preserve or self.opponent_change_count >= 5: next_run = next_update - datetime.timedelta(hours=self.config.Exercise_DelayUntilHoursBeforeNextUpdate) if next_run >= now or should_delay: minutes_to_delay = int((next_run - now).total_seconds() / 60 + 1) self.config.task_delay(minute=minutes_to_delay) else: self.config.task_delay(server_update=True) else: self.config.task_delay(success=False)