import re from module.base.timer import Timer from module.exception import ScriptError from module.logger import logger from module.ocr.ocr import Digit from tasks.combat.assets.assets_combat_prepare import COMBAT_PREPARE, WAVE_MINUS, WAVE_PLUS from tasks.combat.assets.assets_combat_obtain import * from tasks.dungeon.keywords import DungeonList from tasks.planner.keywords import ITEM_CLASSES from tasks.planner.model import ObtainedAmmount, PlannerMixin from tasks.planner.result import OcrItemName class OcrItemAmount(Digit): def format_result(self, result): res = re.split(r'[:：;；]', result) result = res[-1] return super().format_result(result) class CombatObtain(PlannerMixin): """ Parse items that can be obtained from dungeon Pages: in: COMBAT_PREPARE """ # False to click again when combat ends # True to exit and reenter to get obtained items obtain_frequent_check = False def _obtain_enter(self, entry, skip_first_screenshot=True): """ Args: entry: Item entry skip_first_screenshot: Pages: in: COMBAT_PREPARE out: ITEM_CLOSE """ logger.info(f'Obtain enter {entry}') self.interval_clear(COMBAT_PREPARE) while 1: if skip_first_screenshot: skip_first_screenshot = False else: self.device.screenshot() if self.appear(ITEM_CLOSE): break if self.appear(COMBAT_PREPARE, interval=2): self.device.click(entry) self.interval_reset(COMBAT_PREPARE) continue # Wait animation timeout = Timer(1.4, count=7).start() skip_first_screenshot = True while 1: if skip_first_screenshot: skip_first_screenshot = False else: self.device.screenshot() if self.image_color_count(ITEM_NAME, color=(0, 0, 0), threshold=221, count=50): break if timeout.reached(): logger.warning('Wait obtain item timeout') break def _obtain_close(self, skip_first_screenshot=True): """ Args: skip_first_screenshot: Pages: in: ITEM_CLOSE out: COMBAT_PREPARE """ logger.info(f'Obtain close') self.interval_clear(ITEM_CLOSE) while 1: if skip_first_screenshot: skip_first_screenshot = False else: self.device.screenshot() if not self.appear(ITEM_CLOSE) and self.appear(COMBAT_PREPARE): if self.image_color_count(WAVE_MINUS, color=(246, 246, 246), threshold=221, count=100) \ or self.image_color_count(WAVE_PLUS, color=(246, 246, 246), threshold=221, count=100): break if self.appear_then_click(ITEM_CLOSE, interval=2): continue @staticmethod def _obtain_get_entry(dungeon: DungeonList, index: int = 1, prev: ObtainedAmmount = None): """ Args: dungeon: Current dungeon index: 1 to 3, index to check prev: Previous item checked Returns: ButtonWrapper: Item entry, or None if no more check needed """ if (index > 1 and prev is None) or (index <= 1 and prev is not None): raise ScriptError(f'_obtain_get_entry: index and prev must be set together, index={index}, prev={prev}') if index > 3: return None def may_obtain_one(): if prev is None: return OBTAIN_1 else: return None def may_obtain_multi(): if prev is None: return OBTAIN_1 # End at the item with the lowest rarity if prev.item.is_rarity_green: return None if index == 2: return OBTAIN_2 if index == 3: return OBTAIN_3 if dungeon is None: return may_obtain_multi() if dungeon.is_Echo_of_War: return may_obtain_one() if dungeon.is_Cavern_of_Corrosion: return None if dungeon.is_Stagnant_Shadow: return may_obtain_one() if dungeon.is_Calyx_Golden: if dungeon.is_Calyx_Golden_Treasures: return may_obtain_one() else: return may_obtain_multi() if dungeon.is_Calyx_Crimson: return may_obtain_multi() raise ScriptError(f'_obtain_get_entry: Cannot get entry from {dungeon}') def _obtain_parse(self) -> ObtainedAmmount | None: """ Pages: in: ITEM_CLOSE """ ocr = OcrItemName(ITEM_NAME) item = ocr.matched_single_line(self.device.image, keyword_classes=ITEM_CLASSES) ocr = OcrItemAmount(ITEM_AMOUNT) amount = ocr.ocr_single_line(self.device.image) if item is None: logger.warning('_obtain_parse: Unknown item name') return None # logger.info(f'ObtainedAmmount: item={item}, value={amount}') return ObtainedAmmount( item=item, value=amount, ) def obtain_get(self, dungeon=None, skip_first_screenshot=True) -> list[ObtainedAmmount]: """ Args: dungeon: Current dungeon, or None for no early stop optimization skip_first_screenshot: Returns: list[ObtainedAmmount]: Pages: in: COMBAT_PREPARE out: COMBAT_PREPARE """ logger.hr('Obtain get', level=2) if not skip_first_screenshot: self.device.screenshot() index = 1 prev = None items = [] for _ in range(5): entry = self._obtain_get_entry(dungeon, index=index, prev=prev) if entry is None: logger.info('Obtain get end') break self._obtain_enter(entry) item = self._obtain_parse() if item is not None: items.append(item) index += 1 prev = item self._obtain_close() logger.hr('Obtained Result') for item in items: # Pretend everything is full # item.value += 1000 logger.info(f'Obtained item: {item.item.name}, {item.value}') """ <<< OBTAIN GET RESULT >>> ItemAmount: Arrow_of_the_Starchaser, 15 ItemAmount: Arrow_of_the_Demon_Slayer, 68 ItemAmount: Arrow_of_the_Beast_Hunter, 85 """ self.planner.load_obtained_amount(items) self.planner_write() return items def obtained_is_full(self, dungeon: DungeonList | None) -> bool: if dungeon is None: self.obtain_frequent_check = False return False row = self.planner.row_come_from_dungeon(dungeon) if row is None: self.obtain_frequent_check = False return False # Update self.obtain_get(dungeon) # Check progress row = self.planner.row_come_from_dungeon(dungeon) if row is None: logger.error(f'obtained_is_full: Row disappeared after obtain_get') self.obtain_frequent_check = False return False if not row.need_farm(): logger.info('Planner row full') self.obtain_frequent_check = False return True # obtain_frequent_check self.obtain_frequent_check = True return False if __name__ == '__main__': self = CombatObtain('src') self.device.screenshot() self.obtain_get()