Refactor: Abstract CampaignStatus class and fix task balancer in Raid

2026-04-02 10:44:08 +08:00 · 2022-10-30 03:56:58 +08:00
parent 3182175ec2
commit 5c70364353
7 changed files with 170 additions and 138 deletions
--- a/module/campaign/campaign_status.py
+++ b/module/campaign/campaign_status.py
@@ -0,0 +1,104 @@
+import re
+
+import cv2
+import numpy as np
+
+from module.base.timer import Timer
+from module.campaign.assets import OCR_EVENT_PT, OCR_COIN, OCR_OIL
+from module.logger import logger
+from module.ocr.ocr import Ocr, Digit
+from module.ui.ui import UI
+
+OCR_OIL = Digit(OCR_OIL, name='OCR_OIL', letter=(247, 247, 247), threshold=128)
+OCR_COIN = Digit(OCR_COIN, name='OCR_COIN', letter=(239, 239, 239), threshold=128)
+
+
+class PtOcr(Ocr):
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, lang='azur_lane', alphabet='X0123456789', **kwargs)
+
+    def pre_process(self, image):
+        """
+        Args:
+            image (np.ndarray): Shape (height, width, channel)
+
+        Returns:
+            np.ndarray: Shape (width, height)
+        """
+        # Use MAX(r, g, b)
+        r, g, b = cv2.split(cv2.subtract((255, 255, 255, 0), image))
+        image = cv2.min(cv2.min(r, g), b)
+        # Remove background, 0-192 => 0-255
+        image = cv2.multiply(image, 255 / 192)
+
+        return image.astype(np.uint8)
+
+
+OCR_PT = PtOcr(OCR_EVENT_PT)
+
+
+class CampaignStatus(UI):
+    def get_event_pt(self):
+        """
+        Returns:
+            int: PT amount, or 0 if unable to parse
+        """
+        pt = OCR_PT.ocr(self.device.image)
+
+        res = re.search(r'X(\d+)', pt)
+        if res:
+            pt = int(res.group(1))
+            logger.attr('Event_PT', pt)
+            return pt
+        else:
+            logger.warning(f'Invalid pt result: {pt}')
+            return 0
+
+    def get_coin(self, skip_first_screenshot=True):
+        """
+        Returns:
+            int: Coin amount
+        """
+        amount = 0
+        timeout = Timer(1, count=2).start()
+        while 1:
+            if skip_first_screenshot:
+                skip_first_screenshot = False
+            else:
+                self.device.screenshot()
+
+            if timeout.reached():
+                logger.warning('Get coin timeout')
+                break
+
+            amount = OCR_COIN.ocr(self.device.image)
+            if amount >= 100:
+                break
+
+        return amount
+
+    def _get_oil(self):
+        return OCR_OIL.ocr(self.device.image)
+
+    def get_oil(self, skip_first_screenshot=True):
+        """
+        Returns:
+            int: Coin amount
+        """
+        amount = 0
+        timeout = Timer(1, count=2).start()
+        while 1:
+            if skip_first_screenshot:
+                skip_first_screenshot = False
+            else:
+                self.device.screenshot()
+
+            if timeout.reached():
+                logger.warning('Get coin timeout')
+                break
+
+            amount = OCR_OIL.ocr(self.device.image)
+            if amount >= 100:
+                break
+
+        return amount