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.downtime_fetch import get_downtime_end 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 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 ExerciseClearTime. False if available but today is not maintenance day. None if unavailable (disabled, window missing, or parse failed). """ end_dt = get_downtime_end(self.config) if end_dt is None: return None if end_dt <= now: return False threshold_str = self.config.cross_get('DowntimeMgmt.DowntimeStrategy.ExerciseClearTime', '15:00:00') threshold_time = datetime.datetime.strptime(threshold_str, '%H:%M:%S').time() return end_dt.date() == now.date() and end_dt.time() >= threshold_time 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)