wess09/module/os_combat/combat.py

from module.combat.assets import *
from module.combat.combat import Combat as Combat_
from module.logger import logger
from module.os_combat.assets import *
from module.os_handler.assets import *
from module.os_handler.map_event import MapEventHandler
from module.base.timer import Timer
from module.exception import GameBugError
from module.statistics.opsi_runtime import finish_battle_timer, start_battle_timer


class ContinuousCombat(Exception):
    pass


class Combat(Combat_, MapEventHandler):
    def combat_appear(self):
        """
        Returns:
            bool: If enter combat.
        """
        if self.is_in_map():
            return False

        if self.is_combat_loading():
            return True

        # Check if already in combat execution (PAUSE button visible)
        # This handles cases where auto search skips battle preparation
        if self.is_combat_executing():
            return True

        if self.appear(BATTLE_PREPARATION):
            return True
        if self.appear(SIREN_PREPARATION, offset=(20, 20)):
            return True
        if self.appear(BATTLE_PREPARATION_WITH_OVERLAY) and self.handle_combat_automation_confirm():
            return True

        return False

    def combat_preparation(self, balance_hp=False, emotion_reduce=False, auto='combat_auto', fleet_index=1):
        """
        Args:
            balance_hp (bool):
            emotion_reduce (bool):
            auto (str):
            fleet_index (int):
        """
        logger.info('Combat preparation.')
        self.device.stuck_record_clear()
        self.device.click_record_clear()
        skip_first_screenshot = True

        # if emotion_reduce:
        #     self.emotion.wait(fleet=fleet_index)
        # if balance_hp:
        #     self.hp_balance()

        for _ in self.loop():

            if self.appear(BATTLE_PREPARATION):
                if self.handle_combat_automation_set(auto=auto == 'combat_auto'):
                    continue
            if self.handle_retirement():
                continue
            # if self.handle_combat_low_emotion():
            #     continue
            # if balance_hp and self.handle_emergency_repair_use():
            #     continue
            if self.appear_then_click(BATTLE_PREPARATION, interval=2):
                continue
            if self.appear_then_click(SIREN_PREPARATION, offset=(20, 20), interval=2):
                continue
            if self.handle_popup_confirm('ENHANCED_ENEMY'):
                continue
            if self.handle_combat_automation_confirm():
                continue
            if self.handle_story_skip():
                continue

            # End
            pause = self.is_combat_executing()
            if pause:
                logger.attr('BattleUI', pause)
                # if emotion_reduce:
                #     self.emotion.reduce(fleet_index)
                break

    def _get_exp_info_sleep(self):
        return (1.5, 2) if self.__os_combat_drop else (0.25, 0.5)

    def handle_exp_info(self):
        if self.is_combat_executing():
            return False
        sleep = self._get_exp_info_sleep()
        if self.appear_then_click(EXP_INFO_S):
            self.device.sleep(sleep)
            return True
        if self.appear_then_click(EXP_INFO_A):
            self.device.sleep(sleep)
            return True
        if self.appear_then_click(EXP_INFO_B):
            self.device.sleep(sleep)
            return True
        if self.appear_then_click(EXP_INFO_C):
            self.device.sleep(sleep)
            return True
        if self.appear_then_click(EXP_INFO_D):
            self.device.sleep(sleep)
            return True

        return False

    def handle_get_items(self, drop=None):
        """
        Click CLICK_SAFE_AREA instead of button itself.

        Args:
            drop (DropImage):

        Returns:
            bool:
        """
        if getattr(self, '_disable_handle_get_items', False):
            return False
        if self.appear(GET_ITEMS_1, offset=5, interval=self.battle_status_click_interval):
            if drop:
                drop.handle_add(self, before=2)
            self.device.click(CLICK_SAFE_AREA)
            self.interval_reset(BATTLE_STATUS_S)
            self.interval_reset(BATTLE_STATUS_A)
            self.interval_reset(BATTLE_STATUS_B)
            return True
        if self.appear(GET_ITEMS_2, offset=5, interval=self.battle_status_click_interval):
            if drop:
                drop.handle_add(self, before=2)
            self.device.click(CLICK_SAFE_AREA)
            self.interval_reset(BATTLE_STATUS_S)
            self.interval_reset(BATTLE_STATUS_A)
            self.interval_reset(BATTLE_STATUS_B)
            return True
        if self.appear(GET_ADAPTABILITY, offset=5, interval=self.battle_status_click_interval):
            if drop:
                drop.handle_add(self, before=2)
            self.device.click(CLICK_SAFE_AREA)
            self.interval_reset(BATTLE_STATUS_S)
            self.interval_reset(BATTLE_STATUS_A)
            self.interval_reset(BATTLE_STATUS_B)
            return True

        return False

    def _os_combat_expected_end(self):
        if self.handle_map_event(drop=self.__os_combat_drop):
            return False
        if self.combat_appear():
            raise ContinuousCombat

        return self.handle_os_in_map()

    __os_combat_drop = None

    def combat_status(self, drop=None, expected_end=None):
        self.__os_combat_drop = drop
        if expected_end is None:
            expected_end = self._os_combat_expected_end
        # disable handle_get_items and use only handle_map_get_items
        self._disable_handle_get_items = True
        try:
            super().combat_status(drop=drop, expected_end=expected_end)
        finally:
            self._disable_handle_get_items = False

    def combat(self, *args, save_get_items=False, **kwargs):
        """
        This handle continuous combat in operation siren.

        In siren scanning device, there are 2 ambush enemies with no interval.
        Fleet goto siren scanning device, attack one enemy, skip TB, attack another.
        Function `combat` has to confirm that combat was finished, and is_in_map.
        When handling siren scanning device, it will stuck in the second combat.
        This function inherits it and detect the second combat.
        """
        for count in range(3):
            if count >= 2:
                logger.warning('Too many continuous combat')

            try:
                super().combat(*args, save_get_items=save_get_items, **kwargs)
                break
            except ContinuousCombat:
                logger.info('Continuous combat detected')
                continue

    def _handle_single_battle_status(self, status_button, status_letter, drop):
        if self.appear(status_button, interval=self.battle_status_click_interval):
            if status_letter == 'S':
                logger.info(f'Battle Status {status_letter}')
            else:
                logger.warning(f'Battle Status {status_letter}')
            if drop:
                drop.handle_add(self)
            else:
                self.device.sleep((0.25, 0.5))
            self.device.click(status_button)
            return True
        return False

    def handle_auto_search_battle_status(self, drop=None):
        for status_button, status_letter in [
            (BATTLE_STATUS_S, 'S'),
            (BATTLE_STATUS_A, 'A'),
            (BATTLE_STATUS_B, 'B'),
            (BATTLE_STATUS_C, 'C'),
            (BATTLE_STATUS_D, 'D'),
        ]:
            if self._handle_single_battle_status(status_button, status_letter, drop):
                return True
        return False

    def handle_auto_search_exp_info(self):
        sleep = self._get_exp_info_sleep()
        for exp_info_button in [EXP_INFO_S, EXP_INFO_A, EXP_INFO_B, EXP_INFO_C, EXP_INFO_D]:
            if self.appear_then_click(exp_info_button):
                self.device.sleep(sleep)
                return True
        return False

    def auto_search_combat(self, drop=None):
        """
        Args:
            drop (DropImage):

        Returns:
            bool: True if enemy cleared, False if fleet died.

        Pages:
            in: is_combat_loading()
            out: combat status
        """
        # Keep combat focused on state transitions; the metrics layer decides
        # whether this task should produce CL1/short-meow timing samples.
        battle_timer_source = start_battle_timer(self.config)
        
        cl1_combat_timer = Timer(300, count=300)
        
        logger.info('Auto search combat loading')
        self.device.stuck_record_clear()
        self.device.click_record_clear()
        self.device.screenshot_interval_set('combat')
        while 1:
            self.device.screenshot()

            if self.handle_combat_automation_confirm():
                continue

            # End
            if self.handle_os_auto_search_map_option(drop=drop):
                break
            pause = self.is_combat_executing()
            if pause:
                logger.attr('BattleUI', pause)
                break
            if self.is_in_map():
                break

        logger.info('Auto Search combat execute')
        self.submarine_call_reset()
        self.device.stuck_record_clear()
        self.device.click_record_clear()
        submarine_mode = 'do_not_use'
        if self.config.Submarine_Fleet:
            submarine_mode = self.config.Submarine_Mode

        if battle_timer_source == 'cl1':
            cl1_combat_timer.start()

        success = True
        while 1:
            self.device.screenshot()

            if battle_timer_source == 'cl1' and cl1_combat_timer.reached():
                logger.warning('CL1 combat timeout (5 minutes limit reached)')
                raise GameBugError('CL1 combat timeout')

            if self.handle_submarine_call(submarine_mode):
                continue
            # Don't change auto search option if failed
            enable = success if success is not None else None
            if self.handle_os_auto_search_map_option(drop=drop, enable=enable):
                continue

            # End
            if self.is_in_map():
                self.device.screenshot_interval_set()
                break
            if self.is_combat_executing():
                continue
            if self.handle_auto_search_battle_status():
                success = None
                continue
            if self.config.OpsiGeneral_RepairThreshold > 0 and self.handle_auto_search_exp_info():
                success = None
                continue
            if self.handle_map_event():
                continue
            
        logger.info('Combat end.')
        
        # Finish through the same metrics source so CL1 and short-meow samples
        # cannot accidentally share a storage key.
        finish_battle_timer(self.config, battle_timer_source)
        
        return success