scikit error: Warning: Trying to unpickle estimator from version 1.1.1 when using 1.7.1

[puesoy]

Hi there,

I recently got the repo running, but I'm seeing a warning regarding a scikit-learn version mismatch. It looks like the model was trained on v1.1.1, but I am running v1.7.1.

Console Output:

InconsistentVersionWarning: Trying to unpickle estimator LogisticRegression from version 1.1.1 when using version 1.7.1. This might lead to breaking code or invalid results. Use at your own risk.

Do you have a guide on how to update/retrain the model for the new version, or should I downgrade my local environment to 1.1.1 to avoid issues?

Thanks!

[puesoy]

If you want to check it out:
I had to add screenshots for the floors and the coop button:

and this is the code:

"""
Rush Royale Bot Core - Python 3.13 Compatible
Enhanced error handling and modern Python features
"""
from __future__ import annotations

import os
import time
import numpy as np
import pandas as pd
import logging
import subprocess
import shutil
from subprocess import Popen, DEVNULL, PIPE
from typing import Optional, Dict, Any, Tuple
from pathlib import Path

# Android ADB - Updated for pure-python-adb + scrcpy hybrid
try:
    from ppadb.client import Client as AdbClient
    from ppadb.device import Device
    ADB_AVAILABLE = True

    # Try to import scrcpy for enhanced screenshot capability
    try:
        import scrcpy
        SCRCPY_AVAILABLE = True
    except ImportError:
        SCRCPY_AVAILABLE = False

    # Create constants for touch actions (replacing scrcpy const)
    class TouchConstants:
        ACTION_DOWN = 0
        ACTION_UP = 1
        KEYCODE_BACK = 4

    const = TouchConstants()
except ImportError:
    # Fallback for missing dependencies
    class AdbClient:
        def __init__(self, host='127.0.0.1', port=5037):
            self.host = host
            self.port = port
        def devices(self):
            return []

    class Device:
        def __init__(self):
            self.serial = None
        def shell(self, command):
            pass
        def input_tap(self, x, y):
            pass
        def input_swipe(self, x1, y1, x2, y2, duration=1000):
            pass

    class TouchConstants:
        ACTION_DOWN = 0
        ACTION_UP = 1
        KEYCODE_BACK = 4

    ADB_AVAILABLE = False
    SCRCPY_AVAILABLE = False

    const = TouchConstants()
    ADB_AVAILABLE = False

# Image processing
import cv2
# internal
import bot_perception
import port_scan

SLEEP_DELAY = 0.1


class Bot:

    def __init__(self, device=None):
        self.bot_stop = False
        self.combat = self.output = self.grid_df = self.unit_series = self.merge_series = self.df_groups = self.info = self.combat_step = None
        self.logger = logging.getLogger('__main__')
        if device is None:
            device = port_scan.get_device()
        if not device:
            raise Exception("No device found!")
        self.device = device
        self.bot_id = self.device.split(':')[-1]

        # Initialize ADB client
        self.adb_client = AdbClient()
        self.adb_device = None

        # Initialize scrcpy process for screenshots
        self.scrcpy_process = None
        self.scrcpy_executable = self.find_scrcpy_executable()

        # Connect to device
        devices = self.adb_client.devices()
        for dev in devices:
            if dev.serial == self.device:
                self.adb_device = dev
                break

        if not self.adb_device:
            # Try to connect
            self.shell(f'adb connect {self.device}')
            devices = self.adb_client.devices()
            for dev in devices:
                if dev.serial == self.device:
                    self.adb_device = dev
                    break

        if not self.adb_device:
            raise Exception(f"Could not connect to device {self.device}")

        # Launch application through ADB shell
        self.adb_device.shell('monkey -p com.my.defense 1')

        # Check if 'bot_feed.png' exists
        if not os.path.isfile(f'bot_feed_{self.bot_id}.png'):
            self.getScreen()
        self.screenRGB = cv2.imread(f'bot_feed_{self.bot_id}.png')

        self.logger.info('Connected to Android device via ADB')
        time.sleep(0.5)

    def __exit__(self, exc_type, exc_value, traceback):
        self.bot_stop = True
        self.logger.info('Exiting bot')
        # Stop scrcpy process if running
        if self.scrcpy_process:
            self.stop_scrcpy()

    def find_scrcpy_executable(self) -> Optional[str]:
        """Find scrcpy executable in common locations"""
        possible_paths = [
            'scrcpy.exe',  # In PATH
            r'C:\Program Files\scrcpy\scrcpy.exe',
            r'C:\Program Files (x86)\scrcpy\scrcpy.exe',
            r'.\scrcpy\scrcpy.exe',  # Local directory
            r'.\bin\scrcpy.exe',
        ]

        for path in possible_paths:
            if shutil.which(path) or os.path.exists(path):
                self.logger.info(f'Found scrcpy at: {path}')
                return path

        self.logger.warning('scrcpy executable not found - will use ADB screencap fallback')
        return None

    def start_scrcpy(self) -> bool:
        """Start scrcpy process for screen mirroring"""
        if not self.scrcpy_executable:
            return False

        try:
            # Start scrcpy in window mode with no controls (view only)
            cmd = [
                self.scrcpy_executable,
                '--serial', self.device,
                '--no-control',  # View only
                '--window-title', f'RR Bot {self.device}',
                '--window-width', '800',
                '--window-height', '450'
            ]

            self.scrcpy_process = Popen(cmd, stdout=DEVNULL, stderr=DEVNULL)
            self.logger.info('Started scrcpy process for screen mirroring')
            time.sleep(2)  # Give scrcpy time to start
            return True

        except Exception as e:
            self.logger.error(f'Failed to start scrcpy: {e}')
            self.scrcpy_process = None
            return False

    def stop_scrcpy(self):
        """Stop scrcpy process"""
        if self.scrcpy_process:
            try:
                self.scrcpy_process.terminate()
                self.scrcpy_process.wait(timeout=5)
                self.logger.info('Stopped scrcpy process')
            except subprocess.TimeoutExpired:
                self.scrcpy_process.kill()
                self.logger.warning('Force killed scrcpy process')
            except Exception as e:
                self.logger.error(f'Error stopping scrcpy: {e}')
            finally:
                self.scrcpy_process = None

    # Function to send ADB shell command
    def shell(self, cmd):
        if self.adb_device:
            return self.adb_device.shell(cmd)
        else:
            # Fallback to system ADB
            p = Popen(['adb', '-s', self.device, 'shell', cmd], stdout=DEVNULL, stderr=DEVNULL)
            p.wait()

    # Send ADB to click screen
    def click(self, x, y, delay_mult=1):
        if self.adb_device:
            self.adb_device.input_tap(x, y)
        else:
            # Fallback to shell command
            self.shell(f'input tap {x} {y}')
        time.sleep(SLEEP_DELAY * delay_mult)

    # Click button coords offset and extra delay
    def click_button(self, pos):
        coords = np.array(pos) + 10
        self.click(*coords)
        time.sleep(SLEEP_DELAY * 10)

    # Swipe on combat grid to merge units
    def swipe(self, start, end):
        boxes, box_size = get_grid()
        # Offset from box edge
        offset = 60
        start_pos = boxes[start[0], start[1]] + offset
        end_pos = boxes[end[0], end[1]] + offset

        if self.adb_device:
            self.adb_device.input_swipe(start_pos[0], start_pos[1], end_pos[0], end_pos[1], 300)
        else:
            # Fallback to shell command
            self.shell(f'input swipe {start_pos[0]} {start_pos[1]} {end_pos[0]} {end_pos[1]} 300')

    # Send key command
    def key_input(self, key):
        if self.adb_device:
            self.adb_device.input_keyevent(key)
        else:
            self.shell(f'input keyevent {key}')

    # Force restart the game through ADB, or spam 10 disconnects to abandon match
    def restart_RR(self, quick_disconnect=False):
        if quick_disconnect:
            for i in range(15):
                if self.adb_device:
                    self.adb_device.shell('monkey -p com.my.defense 1')
                else:
                    self.shell('monkey -p com.my.defense 1')  # disconnects really quick for unknown reasons
            return
        # Force kill game through ADB shell
        if self.adb_device:
            self.adb_device.shell('am force-stop com.my.defense')
        else:
            self.shell('am force-stop com.my.defense')
        time.sleep(2)
        # Launch application through ADB shell
        if self.adb_device:
            self.adb_device.shell('monkey -p com.my.defense 1')
        else:
            self.shell('monkey -p com.my.defense 1')
        time.sleep(10)  # wait for app to load

    # Take screenshot of device screen and load pixel values
    def getScreen(self):
        bot_id = self.device.split(':')[-1]
        screenshot_path = f'bot_feed_{bot_id}.png'

        # Method 1: Try scrcpy executable screenshot (fastest, highest quality)
        if self.scrcpy_executable and self._try_scrcpy_screenshot(screenshot_path):
            self.logger.debug('Screenshot taken via scrcpy executable')
        # Method 2: Try pure-python-adb (reliable)
        elif self._try_adb_screenshot(screenshot_path):
            self.logger.debug('Screenshot taken via pure-python-adb')
        # Method 3: Fallback to shell ADB (last resort)
        elif self._try_shell_screenshot(screenshot_path):
            self.logger.debug('Screenshot taken via ADB shell')
        else:
            self.logger.error('All screenshot methods failed!')
            return

        # Load screenshot and validate
        try:
            new_img = cv2.imread(screenshot_path)
            if new_img is not None and new_img.shape[0] > 0 and new_img.shape[1] > 0:
                self.screenRGB = new_img
                self.logger.debug(f'Screenshot loaded successfully: {new_img.shape}')
            else:
                self.logger.warning(f'Invalid screenshot file: {screenshot_path}')
        except Exception as e:
            self.logger.error(f'Failed to load screenshot: {e}')

    def _try_scrcpy_screenshot(self, output_path: str) -> bool:
        """Try taking screenshot using scrcpy executable"""
        if not self.scrcpy_executable:
            return False
        try:
            cmd = [
                self.scrcpy_executable,
                '--serial', self.device,
                '--no-display',  # No window
                '--record', output_path.replace('.png', '.mp4'),
                '--time-limit', '1'  # Record for 1 second
            ]
            # Alternative: use scrcpy screenshot feature if available
            cmd = ['adb', '-s', self.device, 'exec-out', 'screencap', '-p']
            with open(output_path, 'wb') as f:
                p = subprocess.run(cmd, stdout=f, stderr=DEVNULL, timeout=10)
                return p.returncode == 0
        except Exception:
            return False

    def _try_adb_screenshot(self, output_path: str) -> bool:
        """Try taking screenshot using pure-python-adb"""
        try:
            if self.adb_device:
                screencap = self.adb_device.screencap()
                if screencap and len(screencap) > 1000:  # Reasonable size check
                    with open(output_path, 'wb') as f:
                        f.write(screencap)
                    return True
        except Exception as e:
            self.logger.debug(f'ADB screencap failed: {e}')
        return False

    def _try_shell_screenshot(self, output_path: str) -> bool:
        """Try taking screenshot using shell ADB command"""
        try:
            cmd = ['adb', '-s', self.device, 'exec-out', 'screencap', '-p']
            with open(output_path, 'wb') as f:
                p = subprocess.run(cmd, stdout=f, stderr=DEVNULL, timeout=10)
                return p.returncode == 0
        except Exception:
            return False

    # Crop latest screenshot taken
    def crop_img(self, x, y, dx, dy, name='icon.png'):
        # Load screen
        img_rgb = self.screenRGB
        img_rgb = img_rgb[y:y + dy, x:x + dx]
        cv2.imwrite(name, img_rgb)

    def getMana(self):
        return int(self.getText(220, 1360, 90, 50, new=False, digits=True))

    # find icon on screen
    def getXYByImage(self, target, new=True):
        valid_targets = ['battle_icon', 'pvp_button', 'back_button', 'cont_button', 'fighting']
        if not target in valid_targets:
            return "INVALID TARGET"
        if new:
            self.getScreen()
        imgSrc = f'icons/{target}.png'
        img_rgb = self.screenRGB
        img_gray = cv2.cvtColor(img_rgb, cv2.COLOR_BGR2GRAY)
        template = cv2.imread(imgSrc, 0)
        res = cv2.matchTemplate(img_gray, template, cv2.TM_CCOEFF_NORMED)
        threshold = 0.8
        loc = np.where(res >= threshold)
        if len(loc[0]) > 0:
            y = loc[0][0]
            x = loc[1][0]
            return [x, y]

    def get_store_state(self):
        x, y = [140, 1412]
        store_states_names = ['refresh', 'new_store', 'nothing', 'new_offer', 'spin_only']
        store_states = np.array([[255, 255, 255], [27, 235, 206], [63, 38, 12], [48, 253, 251], [80, 153, 193]])
        store_rgb = self.screenRGB[y:y + 1, x:x + 1]
        store_rgb = store_rgb[0][0]
        # Take mean square of rgb value and store states
        store_mse = ((store_states - store_rgb)**2).mean(axis=1)
        closest_state = store_mse.argmin()
        return store_states_names[closest_state]

    # Check if any icons are on screen
    def get_current_icons(self, new=True, available=False, icon_list=None):
        """
        Scans the screen for specified icons or all icons if none specified.
        Optionally filters for available icons.
        param: new - bool, whether to take a new screenshot
        param: available - bool, whether to return only available icons
        param: icon_list - list of strings, specific icon filenames to search for (e.g., ['chapter_1.png', 'floor5.png'])
        """
        current_icons = []
        # Update screen and load screenshot as grayscale
        if new:
            self.getScreen()
        img_rgb = self.screenRGB
        if img_rgb is None:
            self.logger.warning('Screenshot is None - cannot detect icons')
            return pd.DataFrame(columns=['icon', 'available', 'pos [X,Y]'])

        img_gray = cv2.cvtColor(img_rgb, cv2.COLOR_BGR2GRAY)
        # Light blur to reduce noise and improve template match stability
        img_gray_blur = cv2.GaussianBlur(img_gray, (3, 3), 0)
        self.logger.debug(f'Screenshot shape: {img_gray.shape}')

        def match_template_multi_scale(src_gray, tmpl_gray, base_thresh, is_chapter=False):
            """Return (found:boolean, (x,y):tuple, max_val:float). Tries multiple scales when needed."""
            best = (False, (0, 0), 0.0)
            # per-icon scaling tries
            scales = [1.0]
            if is_chapter:
                scales = [0.9, 1.0, 1.1]
            for sc in scales:
                if sc != 1.0:
                    new_w = max(1, int(tmpl_gray.shape[1] * sc))
                    new_h = max(1, int(tmpl_gray.shape[0] * sc))
                    tmpl = cv2.resize(tmpl_gray, (new_w, new_h), interpolation=cv2.INTER_AREA)
                else:
                    tmpl = tmpl_gray
                if src_gray.shape[0] < tmpl.shape[0] or src_gray.shape[1] < tmpl.shape[1]:
                    continue
                res = cv2.matchTemplate(src_gray, tmpl, cv2.TM_CCOEFF_NORMED)
                min_val, max_val, min_loc, max_loc = cv2.minMaxLoc(res)
                # Keep best regardless of threshold, decide later
                if max_val > best[2]:
                    best = (max_val >= base_thresh, (max_loc[0], max_loc[1]), float(max_val))
            # Fallback: slightly relax if near-threshold for chapters
            if is_chapter and not best[0] and best[2] >= (base_thresh - 0.05):
                return (True, best[1], best[2])
            return best

        # Determine which icons to check
        if icon_list is None:
            # Check every target in dir if no list is provided
            icons_to_check = os.listdir("icons")
        else:
            # Only check icons in the provided list
            icons_to_check = icon_list

        icon_count = 0
        for target in icons_to_check:
            x = 0  # reset position
            y = 0
            # Load icon
            imgSrc = f'icons/{target}'
            # Check if icon file exists
            if not os.path.isfile(imgSrc):
                 self.logger.debug(f'Icon file not found: {imgSrc}')
                 continue

            template = cv2.imread(imgSrc, 0)
            if template is None:
                self.logger.debug(f'Could not load template: {imgSrc}')
                continue
            # Slight blur for template too
            template_blur = cv2.GaussianBlur(template, (3, 3), 0)
            # Per-icon threshold tuning
            is_chapter = ('chapter_' in target)
            threshold = 0.8
            if is_chapter:
                threshold = 0.75
            elif target in ['dungeon_page.png']:
                threshold = 0.60 # Reduced threshold for dungeon_page.png
            elif 'floor' in target: # Add threshold for floor icons
                 threshold = 0.95 # Keep threshold for floor icons at 0.95
            elif target in ['random.png']: # Add threshold for random coop button
                 threshold = 0.85 # Adjust as needed based on image
            elif target in ['pve_button.png']: # Threshold for PvE button on home screen
                 threshold = 0.85 # Adjust as needed

            # Compare images using robust best-location extraction
            found, (best_x, best_y), max_val = match_template_multi_scale(img_gray_blur, template_blur, threshold, is_chapter=is_chapter)
            icon_found = found

            # Debug for key icons
            if target in ['home_screen.png', 'battle_icon.png', 'dungeon_page.png', 'random.png', 'pve_button.png'] or 'chapter_' in target or 'floor' in target:
                self.logger.debug(f'Icon {target}: max_val={max_val:.3f}, found={icon_found}')


            if icon_found:
                y = int(best_y)
                x = int(best_x)
                icon_count += 1
            current_icons.append([target, icon_found, (x, y)])

        self.logger.debug(f'Total icons found: {icon_count}/{len(current_icons)} from list of {len(icons_to_check)} to check')
        icon_df = pd.DataFrame(current_icons, columns=['icon', 'available', 'pos [X,Y]'])
        # filter out only available buttons
        if available:
            icon_df = icon_df[icon_df['available'] == True].reset_index(drop=True)
        return icon_df

    # Scan battle grid, update OCR images
    def scan_grid(self, new=False):
        boxes, box_size = get_grid()
        # should be enabled by default
        if new:
            self.getScreen()
        box_list = boxes.reshape(15, 2)
        names = []
        if not os.path.isdir('OCR_inputs'):
            os.mkdir('OCR_inputs')
        for i in range(len(box_list)):
            file_name = f'OCR_inputs/icon_{str(i)}.png'
            self.crop_img(*box_list[i], *box_size, name=file_name)
            names.append(file_name)
        return names

    # Take random unit in series, find corresponding dataframe and merge two random ones
    def merge_unit(self, df_split, merge_series):
        # Pick a random filtered target
        if len(merge_series) > 0:
            merge_target = merge_series.sample().index[0]
        else:
            return merge_series
        # Collect unit dataframe
        merge_df = df_split.get_group(merge_target)
        if len(merge_df) > 1:
            merge_df = merge_df.sample(n=2)
        else:
            return merge_df
        self.log_merge(merge_df)
        # Extract unit position from dataframe
        unit_chosen = merge_df['grid_pos'].tolist()
        # Send Merge
        self.swipe(*unit_chosen)
        time.sleep(0.2)
        return merge_df

    # Merge special units ['harlequin.png','dryad.png','mime.png','scrapper.png']
    # Add logging event
    def merge_special_unit(self, df_split, merge_series, special_type):
        # Get special merge unit
        special_unit, normal_unit = [
            adv_filter_keys(merge_series, units=special_type, remove=remove) for remove in [False, True]
        ]  # scrapper support not tested
        # Get corresponding dataframes
        special_df, normal_df = [df_split.get_group(unit.index[0]).sample() for unit in [special_unit, normal_unit]]
        merge_df = pd.concat([special_df, normal_df])
        self.log_merge(merge_df)
        # Merge 'em
        unit_chosen = merge_df['grid_pos'].tolist()
        self.swipe(*unit_chosen)
        time.sleep(0.2)
        return merge_df

    def log_merge(self, merge_df):
        merge_df['unit'] = merge_df['unit'].apply(lambda x: x.replace('.png', ''))
        unit1, unit2 = merge_df.iloc[0:2]['unit']
        rank = merge_df.iloc[0]['rank']
        log_msg = f"Rank {rank} {unit1}-> {unit2}"
        # Determine log level from rank
        if rank > 4:
            self.logger.error(log_msg)
        elif rank > 2:
            self.logger.debug(log_msg)
        else:
            self.logger.info(log_msg)

    # Find targets for special merge
    def special_merge(self, df_split, merge_series, target='zealot.png'):
        merge_df = None
        # Try to rank up dryads
        dryads_series = adv_filter_keys(merge_series, units='dryad.png')
        if not dryads_series.empty:
            dryads_rank = dryads_series.index.get_level_values('rank')
            for rank in dryads_rank:
                merge_series_dryad = adv_filter_keys(merge_series, units=['harlequin.png', 'dryad.png'], ranks=rank)
                merge_series_zealot = adv_filter_keys(merge_series, units=['dryad.png', target], ranks=rank)
                if len(merge_series_dryad.index) == 2:
                    merge_df = self.merge_special_unit(df_split, merge_series_dryad, special_type='harlequin.png')
                    break
                if len(merge_series_zealot.index) == 2:
                    merge_df = self.merge_special_unit(df_split, merge_series_zealot, special_type='dryad.png')
                    break
        return merge_df

    # Harley Merge target
    def harley_merge(self, df_split, merge_series, target='knight_statue.png'):
        merge_df = None
        # Try to copy target
        hq_series = adv_filter_keys(merge_series, units='harlequin.png')
        if not hq_series.empty:
            hq_rank = hq_series.index.get_level_values('rank')
            for rank in hq_rank:
                merge_series_target = adv_filter_keys(merge_series, units=['harlequin.png', target], ranks=rank)
                if len(merge_series_target.index) == 2:
                    merge_df = self.merge_special_unit(df_split, merge_series_target, special_type='harlequin.png')
                    break
        return merge_df

    # Try to find a merge target and merge it
    def try_merge(self, rank=1, prev_grid=None, merge_target='zealot.png'):
        info = ''
        merge_df = None
        names = self.scan_grid(new=False)
        grid_df = bot_perception.grid_status(names, prev_grid=prev_grid)
        df_split, unit_series, df_groups, group_keys = grid_meta_info(grid_df)
        # Select stuff to merge
        merge_series = unit_series.copy()
        # Remove empty groups
        merge_series = adv_filter_keys(merge_series, units='empty.png', remove=True)
        # Do special merge with dryad/Harley
        self.special_merge(df_split, merge_series, merge_target)
        # Use harely on high dps targets
        if merge_target == 'demon_hunter.png':
            self.harley_merge(df_split, merge_series, target=merge_target)
            # Remove all demons (for co-op)
            demons = adv_filter_keys(merge_series, units='demon_hunter.png')
            num_demon = sum(demons)
            if num_demon >= 11:
                # If board is mostly demons, chill out
                self.logger.info(f'Board is full of demons, waiting...')
                time.sleep(10)
            # Check if self.config exists and has the section/option before accessing
            if hasattr(self, 'config') and self.config.has_section('bot') and self.config.has_option('bot', 'require_shaman') and self.config.getboolean('bot', 'require_shaman'):
                 merge_series = adv_filter_keys(merge_series, units='demon_hunter.png', remove=True)
        merge_series = preserve_unit(merge_series, target='chemist.png')
        # Remove 4x cauldrons
        for _ in range(4):
            merge_series = preserve_unit(merge_series, target='cauldron.png', keep_min=True)
        # Try to keep knight_statue numbers even (can conflict if special_merge already merged)
        num_knight = sum(adv_filter_keys(merge_series, units='knight_statue.png'))
        if num_knight % 2 == 1:
            self.harley_merge(df_split, merge_series, target='knight_statue.png')
        # Preserve 2 highest knight statues
        for _ in range(2):
            merge_series = preserve_unit(merge_series, target='knight_statue.png')
        # Select stuff to merge
        merge_series = merge_series[merge_series >= 2]  # At least 2 units
        merge_series = adv_filter_keys(merge_series, ranks=7, remove=True)  # Remove max ranks
        # Try to merge high priority units
        merge_prio = adv_filter_keys(merge_series,
                                     units=['chemist.png', 'bombardier.png', 'summoner.png', 'knight_statue.png'])
        if not merge_prio.empty:
            info = 'Merging High Priority!'
            merge_df = self.merge_unit(df_split, merge_prio)
        # Merge if board is getting full
        if df_groups['empty.png'] <= 2:
            info = 'Merging!'
            # Add criteria
            low_series = adv_filter_keys(merge_series, ranks=rank, remove=False)
            if not low_series.empty:
                merge_df = self.merge_unit(df_split, low_series)
            else:
                # If grid seems full, merge more units
                info = 'Merging high level!'
                merge_series = adv_filter_keys(merge_series,
                                               ranks=[3, 4, 5, 6, 7],
                                               units=['zealot.png', 'crystal.png', 'bruser.png', merge_target],
                                               remove=True)
                if not merge_series.empty:
                    merge_df = self.merge_unit(df_split, merge_series)
        else:
            info = 'need more units!'
        return grid_df, unit_series, merge_series, merge_df, info

    # Mana level cards
    def mana_level(self, cards, hero_power=False):
        upgrade_pos_dict = {1: [100, 1500], 2: [200, 1500], 3: [350, 1500], 4: [500, 1500], 5: [650, 1500]}
        # Level each card
        for card in cards:
            self.click(*upgrade_pos_dict[card])
        if hero_power:
            self.click(800, 1500)

    # Start a dungeon floor from PvE page
    def play_dungeon(self, floor=5):
        self.logger.debug(f'Starting Dungeon floor {floor}')

        # Map floor numbers to image filenames
        floor_icons = {
            1: 'floor1.png', 2: 'floor2.png', 3: 'floor3.png',
            4: 'floor4.png', 5: 'floor5.png', 6: 'floor6.png',
            7: 'floor7.png', 8: 'floor8.png', 9: 'floor9.png',
            10: 'floor10.png', 11: 'floor11.png', 12: 'floor12.png' # Assuming floors go up to 12
        }
        target_floor_icon_name = floor_icons.get(floor)
        if not target_floor_icon_name:
            self.logger.error(f'Invalid floor number: {floor}')
            return

        # Determine the chapter the target floor belongs to
        target_chapter = int(np.ceil(floor / 3))
        target_chapter_icon_name = f'chapter_{target_chapter}.png' # Define chapter icon name here

        self.logger.debug(f'Looking for target chapter: {target_chapter}, target floor: {target_floor_icon_name}')

        # Check if on dungeon page - Add a retry loop for robustness
        on_dungeon_page = False
        check_attempts = 0
        max_check_attempts = 5
        while not on_dungeon_page and check_attempts < max_check_attempts:
            # Only scan for dungeon_page icon initially
            avail_buttons = self.get_current_icons(available=True, new=True, icon_list=['dungeon_page.png'])
            if (avail_buttons['icon'] == 'dungeon_page.png').any():
                 self.logger.info('Successfully navigated to dungeon page.')
                 on_dungeon_page = True
                 break
            else:
                self.logger.warning(f'Not on dungeon page (attempt {check_attempts + 1}/{max_check_attempts}). Attempting navigation.')
                # Attempt to navigate to dungeon page if not there
                # Call battle_screen with pve=True but start=False to navigate towards PvE if needed
                self.battle_screen(pve=True, start=False)
                time.sleep(2) # Wait for the transition


            check_attempts += 1

        if not on_dungeon_page:
             self.logger.error('Failed to navigate to dungeon page after multiple attempts.')
             return


        # Swipe to the top to ensure we start from the beginning
        self.logger.debug('Scrolling to top of dungeon menu')
        for _ in range(5): # Swipe up a few times
            self.swipe([0, 0], [2, 0])
            time.sleep(0.5)
        self.click(30, 600, 5)  # Stop scroll
        time.sleep(1) # Add small delay after stopping scroll before scanning

        found_target_chapter = False
        chapter_pos = None
        scroll_attempts_chapter = 0
        max_scroll_attempts_chapter = 10 # Limit scrolling for chapters

        # First, find and expand the target chapter
        while not found_target_chapter and scroll_attempts_chapter < max_scroll_attempts_chapter:
            # Only scan for chapter icons during this phase
            # Make sure target_chapter_icon_name is accessible here
            avail_buttons = self.get_current_icons(available=True, new=True, icon_list=[f'chapter_{i}.png' for i in range(1, 13)]) # Scan for all chapter icons
            self.logger.debug(f'Chapter scroll attempt {scroll_attempts_chapter + 1}: Available icons: {avail_buttons["icon"].tolist()}')

            # Look for correct chapter
            # Ensure target_chapter_icon_name is correctly used here
            if (avail_buttons['icon'] == target_chapter_icon_name).any():
                chapter_pos = get_button_pos(avail_buttons, target_chapter_icon_name)
                self.logger.info(f'Found target chapter {target_chapter} ({target_chapter_icon_name}) at position {chapter_pos}')

                # Always click the chapter icon to ensure it's expanded
                self.logger.info(f'Clicking chapter {target_chapter} to ensure it is expanded.')
                # Use the position of the found chapter icon directly for clicking to expand
                self.click_button(chapter_pos) # Click the chapter icon itself
                time.sleep(1) # Give time for expansion animation

                found_target_chapter = True
                break # Exit chapter scrolling loop

            # Scroll down if target chapter is not visible
            self.logger.debug('Target chapter not visible. Scrolling down to find chapter.')
            self.swipe([2, 0], [0, 0]) # Swipe down
            time.sleep(1) # Give time for scrolling
            self.click(30, 600)  # Stop scroll
            time.sleep(1) # Add small delay after stopping scroll before scanning
            scroll_attempts_chapter += 1

        if not found_target_chapter:
            self.logger.error(f'Could not find target chapter {target_chapter} ({target_chapter_icon_name}) after {max_scroll_attempts_chapter} scroll attempts.')
            return

        # Now that the chapter is expanded (or we attempted to expand it), search for the target floor icon
        found_target_floor = False
        scroll_attempts_floor = 0
        max_scroll_attempts_floor = 5 # Limit scrolling for floors within an expanded chapter
        play_button_offset = np.array([330, 1030]) - np.array([120, 790]) # Offset from floor icon (120, 790) -> Play (330, 1030) relative
        floor_pos = None # Initialize floor_pos

        while not found_target_floor and scroll_attempts_floor < max_scroll_attempts_floor:
             # Only scan for the target floor icon during this phase
             # Ensure we are taking a new screenshot in this loop
             avail_buttons = self.get_current_icons(available=True, new=True, icon_list=[target_floor_icon_name])
             self.logger.debug(f'Floor scroll attempt {scroll_attempts_floor + 1}: Available icons: {avail_buttons["icon"].tolist()}')

             if (avail_buttons['icon'] == target_floor_icon_name).any():
                 floor_pos = get_button_pos(avail_buttons, target_floor_icon_name)
                 self.logger.info(f'Found target floor {floor} ({target_floor_icon_name}) at position {floor_pos}')

                 # Check if the floor icon is too low for the Play button to be visible
                 # Based on user input, if floor_pos[1] > 1090, the Play button might be off-screen
                 if floor_pos[1] > 1090:
                     self.logger.info(f'Floor icon is low ({floor_pos[1]}). Scrolling down slightly to reveal Play button.')
                     # Perform a small scroll down
                     self.swipe([2, 0], [0, 0]) # Swipe down
                     time.sleep(1) # Give time for scrolling
                     self.click(30, 600)  # Stop scroll
                     time.sleep(1) # Add small delay after stopping scroll before scanning
                     # After scrolling, re-scan the screen and find the floor icon again
                     # This is important because its position will have changed
                     # Ensure new=True here to get the updated screen after scrolling
                     avail_buttons_after_scroll = self.get_current_icons(available=True, new=True, icon_list=[target_floor_icon_name]) # Force new screenshot
                     if (avail_buttons_after_scroll['icon'] == target_floor_icon_name).any():
                         floor_pos = get_button_pos(avail_buttons_after_scroll, target_floor_icon_name)
                         self.logger.info(f'Found target floor again after scrolling: {floor_pos}')
                     else:
                         self.logger.warning('Could not find floor icon after scrolling to reveal Play button. Proceeding with last known position.')
                         # Keep the last known floor_pos and hope for the best, or implement more robust retry logic
                         # For now, we continue with the potentially outdated floor_pos


                 # Calculate and click the Play button relative to the floor icon position
                 play_button_pos = floor_pos + play_button_offset
                 self.logger.info(f'Calculated Play button position: {play_button_pos}')
                 self.click_button(play_button_pos) # Click the calculated Play button position

                 found_target_floor = True
                 break # Exit floor scrolling loop

             # Scroll down to find the floor icon within the expanded chapter
             self.logger.debug('Target floor icon not visible. Scrolling down to find floor.')
             self.swipe([2, 0], [0, 0]) # Swipe down
             time.sleep(1) # Give time for scrolling
             self.click(30, 600)  # Stop scroll
             time.sleep(1) # Add small delay after stopping scroll before scanning
             scroll_attempts_floor += 1


        if not found_target_floor:
            self.logger.error(f'Could not find target floor {floor} ({target_floor_icon_name}) after {max_scroll_attempts_floor} scroll attempts within the expanded chapter.')
            return

        # --- Logic to handle the intermediate co-op screen (re-added) ---
        self.logger.info('Clicked Play button. Waiting for co-op partner selection screen.')
        coop_screen_visible = False
        coop_check_attempts = 0
        max_coop_check_attempts = 15 # Increased max attempts to find the coop screen

        while not coop_screen_visible and coop_check_attempts < max_coop_check_attempts:
             # Look for the random co-op partner button
             # Added random.png to the icon_list for get_current_icons
             avail_buttons = self.get_current_icons(available=True, new=True, icon_list=['random.png'])
             if (avail_buttons['icon'] == 'random.png').any():
                 self.logger.info('Found random co-op partner button. Clicking it.')
                 coop_button_pos = get_button_pos(avail_buttons, 'random.png')
                 self.click_button(coop_button_pos)
                 coop_screen_visible = True
                 break
             else:
                 self.logger.debug(f'Random co-op button not found (attempt {coop_check_attempts + 1}/{max_coop_check_attempts}). Waiting...')
                 time.sleep(2) # Wait for the screen to load

             coop_check_attempts += 1

        if not coop_screen_visible:
             self.logger.error('Could not find random co-op partner button after multiple attempts.')
             # Decide how to handle this failure: exit, retry dungeon selection, etc.
             # For now, we will log and exit.
             return

        # Wait for the match to start (after clicking random coop)
        self.logger.info('Clicked random co-op button. Waiting for match to start.')
        # Increased wait time for match start, as finding a partner can take time
        for i in range(30):
            time.sleep(2)
            avail_buttons = self.get_current_icons(available=True)
            if avail_buttons['icon'].isin(['back_button.png', 'fighting.png']).any():
                self.logger.info(f'Match started detected after {i*2} seconds.')
                return # Exit function

        self.logger.error('Match start not detected within timeout.')


    # Mana level cards
    def mana_level(self, cards, hero_power=False):
        upgrade_pos_dict = {1: [100, 1500], 2: [200, 1500], 3: [350, 1500], 4: [500, 1500], 5: [650, 1500]}
        # Level each card
        for card in cards:
            self.click(*upgrade_pos_dict[card])
        if hero_power:
            self.click(800, 1500)

    # Start a dungeon floor from PvE page
    def battle_screen(self, start=False, pve=True, floor=5):
        # Scan screen for any key buttons
        df = self.get_current_icons(available=True)
        if not df.empty:
            # list of buttons
            if (df['icon'] == 'fighting.png').any() and not (df['icon'] == '0cont_button.png').any():
                return df, 'fighting'
            if (df['icon'] == 'friend_menu.png').any():
                self.click_button(np.array([100, 600]))
                return df, 'friend_menu'
            # Start pvp if homescreen
            if (df['icon'] == 'home_screen.png').any() and (df['icon'] == 'battle_icon.png').any():
                if pve and start:
                    # Instead of a fixed click, let's try to find the PvE button icon
                    pve_button_df = self.get_current_icons(available=True, new=False, icon_list=['pve_button.png'])
                    if (pve_button_df['icon'] == 'pve_button.png').any():
                         pve_button_pos = get_button_pos(pve_button_df, 'pve_button.png')
                         self.logger.info(f'Found PvE button at {pve_button_pos}. Clicking to start dungeon navigation.')
                         self.click_button(pve_button_pos)
                         time.sleep(2) # Give time for transition
                         # After clicking PvE button, we should be on the dungeon page.
                         # Call play_dungeon from here to continue the dungeon selection process.
                         self.play_dungeon(floor=floor)
                    else:
                         self.logger.warning('On home screen but PvE button not found. Falling back to general battle icon.')
                         # Fallback to clicking the general battle icon and hoping for the best? Or exit?
                         # For now, let's try clicking the general battle icon as a fallback
                         battle_icon_pos = get_button_pos(df, 'battle_icon.png')
                         self.click_button(battle_icon_pos)
                         time.sleep(2)
                         # After clicking battle icon, we might be on a PvP/PvE selection screen or something else.
                         # Let's re-evaluate the screen state to see if we reached the dungeon page
                         return self.battle_screen(start=start, pve=pve, floor=floor) # Recursive call


                elif start: # PvP start logic
                    self.click_button(np.array([140, 1259])) # Original PvP button click
                time.sleep(1)
                return df, 'home'
            # Check first button is clickable
            df_click = df[df['icon'].isin(['back_button.png', 'battle_icon.png', '0cont_button.png', '1quit.png'])]
            if not df_click.empty:
                button_pos = df_click['pos [X,Y]'].tolist()[0]
                self.click_button(button_pos)
                return df, 'menu'
        # Handle cases where none of the above buttons are found (e.g., end of match screens, ads)
        # Check for common end-of-match/ad related icons before forcing back
        # Removed ad-related icons from this list in the previous step
        end_screen_icons_filtered = ['cont_button.png'] # Keep only non-ad related end screens like continue


        end_screen_buttons = self.get_current_icons(available=True, icon_list=end_screen_icons_filtered)

        if not end_screen_buttons.empty:
             self.logger.info(f"Detected non-ad end screen icon(s): {end_screen_buttons['icon'].tolist()}. Attempting to click the first one.")
             first_end_button_pos = get_button_pos(end_screen_buttons, end_screen_buttons['icon'].iloc[0])
             self.click_button(first_end_button_pos)
             time.sleep(2)
             # After clicking a non-ad end screen button, we should re-evaluate the screen state
             # The next state might be the home screen or a menu, so we recursively call battle_screen
             return self.battle_screen(start=start, pve=pve, floor=floor) # Recursive call to re-check screen


        # If no known buttons (including filtered end screens) are found, force back
        self.key_input(const.KEYCODE_BACK)  # Force back
        return df, 'lost'

    # Navigate and locate store refresh button from battle screen
    def find_store_refresh(self):
        self.click_button((100, 1500))  # Click store button
        [self.swipe([0, 0], [2, 0]) for i in range(5)]  # swipe to top
        self.click(30, 150)  # stop scroll
        avail_buttons = self.get_current_icons(available=True)
        if (avail_buttons['icon'] == 'refresh_button.png').any():
            pos = get_button_pos(avail_buttons, 'refresh_button.png')
            return pos

    # Refresh items in shop when available
    def refresh_shop(self):
        self.click_button((100, 1500))  # Click store button
        self.click_button((475, 1300))  # Click store button
        # Scroll up and find the refresh button
        pos = self.find_store_refresh()
        if isinstance(pos, np.ndarray):
            self.click_button(pos - [300, 820])  # Click first (free) item
            self.click(400, 1165)  # buy
            self.click(30, 150)  # remove pop-up
            self.click_button(pos + [400, -400])  # Click last item (possible legendary)
            self.click(400, 1165)  # buy
            self.click(30, 150)  # remove pop-up
            self.logger.warning('Bought store units!')
            # Try to refresh shop (watch ad)
            self.click_button(pos)

    def watch_ads(self):
        # This function is now effectively unused since its calls are commented out or the icons are filtered.
        # Keeping it here for completeness but it won't be executed via battle_screen anymore.
        avail_buttons = self.get_current_icons(available=True)
        # Watch ad if available
        if (avail_buttons['icon'] == 'quest_done.png').any():
            pos = get_button_pos(avail_buttons, 'quest_done.png')
            self.click_button(pos)
            self.click(700, 600)  # collect second completed quest
            self.click(700, 400)  # collect second completed quest
            [self.click(150, 250) for i in range(2)]  # click dailies twice
            self.click(420, 420)  # collect ad chest
        elif (avail_buttons['icon'] == 'ad_season.png').any():
            pos = get_button_pos(avail_buttons, 'ad_season.png')
            self.click_button(pos)
        elif (avail_buttons['icon'] == 'ad_pve.png').any():
            pos = get_button_pos(avail_buttons, 'ad_pve.png')
            self.click_button(pos)
        elif (avail_buttons['icon'] == 'battle_icon.png').any():
            self.refresh_shop()
        else:
            #self.logger.info('Watched all ads!')
            return
        # Check if ad was started
        avail_buttons, status = self.battle_screen()
        if status == 'menu' or status == 'home' or (avail_buttons['icon'] == 'refresh_button.png').any():
            self.logger.info('FINISHED AD')
        # Watch ad
        else:
            time.sleep(30)
            # Keep watching until back in menu
            for i in range(10):
                avail_buttons, status = self.battle_screen()
                if status == 'menu' or status == 'home':
                    self.logger.info('FINISHED AD')
                    return  # Exit function
                time.sleep(2)
                self.click(870, 30)  # skip forward/click X
                self.click(870, 100)  # click X playstore popup
                if i > 5:
                    self.key_input(const.KEYCODE_BACK)  # Force back
                self.logger.info(f'AD TIME {i} {status}')
            # Restart game if can't escape ad
            self.restart_RR()


####
#### END OF CLASS
####


# Get fight grid pixel values
def get_grid():
    #Grid dimensions
    top_box = (153, 945)
    box_size = (120, 120)
    gap = 0
    height = 3
    width = 5
    # x_cords
    x_cord = list(range(top_box[0], top_box[0] + (box_size[0] + gap) * width, box_size[0] + gap))
    y_cord = list(range(top_box[1], top_box[1] + (box_size[1] + gap) * height, box_size[1] + gap))
    boxes = []
    # Create list of all boxes
    for y_point in y_cord:
        for x_point in x_cord:
            boxes.append((x_point, y_point))
    # Convert to np array (4x4) with x,y coords
    boxes = np.array(boxes).reshape(height, width, 2)
    return boxes, box_size


def get_unit_count(grid_df):
    df_split = grid_df.groupby("unit")
    df_groups = df_split["unit"].count()
    if not 'empty.png' in df_groups:
        df_groups['empty.png'] = 0
    unit_list = list(df_groups.index)
    return df_split, df_groups, unit_list


# Removes 1x of the highest rank unit from the merge_series
def preserve_unit(unit_series, target='chemist.png', keep_min=False):
    """
    Remove 1x of the highest rank unit from the merge_series
    param: merge_series - pandas series of units to remove
    param: target - target unit to keep
    param: keep_min - if true, keep the lowest rank unit instead of highest
    """
    merge_series = unit_series.copy()
    preserve_series = adv_filter_keys(merge_series, units=target, remove=False)
    if not preserve_series.empty:
        if keep_min:
            preserve_unit = preserve_series.index.min()
        else:
            preserve_unit = preserve_series.index.max()
        # Remove 1 count of highest/lowest rank
        merge_series[merge_series.index == preserve_unit] = merge_series[merge_series.index == preserve_unit] - 1
        # Remove 0 counts
        return merge_series[merge_series > 0]
    else:
        return merge_series


def grid_meta_info(grid_df, min_age=0):
    """
    Split grid df into unique units and ranks
    Shows total count of unit and count of each rank
    param: grid_df - pandas dataframe of grid
    param: min_age - minimum age of unit to include in meta info
    """
    # Split by unique unit
    df_groups = get_unit_count(grid_df)[1]
    grid_df = grid_df[grid_df['Age'] >= min_age].reset_index(drop=True)
    df_split = grid_df.groupby(['unit', 'rank'])
    # Count number of unit of each rank
    unit_series = df_split['unit'].count()
    #unit_series = unit_series.sort_values(ascending=False)
    group_keys = list(unit_series.index)
    return df_split, unit_series, df_groups, group_keys


def filter_units(unit_series, units):
    if not isinstance(units, list):  # Make units a list if not already
        # If it's a single item (int or str), wrap it in a list
        if isinstance(units, (int, str)):
             units = [units]
        else:
             # If it's neither list, int, nor str, something is fundamentally wrong
             raise TypeError(f"Units must be a string, integer, or a list of strings/integers, but got {type(units)}")

    # Create temp series to hold matches
    series = []
    merge_series = unit_series.copy()
    for token in units:
        if isinstance(token, int):
            exists = merge_series.index.get_level_values('rank').isin([token]).any()
            if exists:
                series.append(merge_series.xs(token, level='rank', drop_level=False))
            else:
                continue  # skip if nothing matches criteria
        elif isinstance(token, str):
            if token in merge_series:
                series.append(merge_series.xs(token, level='unit', drop_level=False))
            else:
                continue
        else:
            # Handle unexpected types within the list
            self.logger.warning(f"Unexpected item type in units list: {type(token)}. Skipping.")
            continue

    if not len(series) == 0:
        temp_series = pd.concat(series)
        # Select all entries from original series that are in temp_series
        merge_series = merge_series[merge_series.index.isin(temp_series.index)]
        return merge_series
    else:
        return pd.Series(dtype=object)


def adv_filter_keys(unit_series, units=None, ranks=None, remove=False):
    """
    Returns all elements which match units and ranks values
    If one of the parameters is None, it is ignored and all values are kept
    If remove is True, all elements are removed which do not match the criteria
    param: unit_series - pandas series of units to filter
    param: units - string or list of strings of units to filter by
    param: ranks - int or list of ints of ranks to filter by
    param: remove - if true, return filtered series, if false, return only matches
    """
    # return if no units in series
    if unit_series.empty:
        return pd.Series(dtype=object)
    filtered_ranks = pd.Series(dtype=object)

    # Ensure units and ranks are handled correctly, especially if None
    filtered_units = unit_series.copy()
    if units is not None:
        filtered_units = filter_units(filtered_units, units)

    filtered_ranks = filtered_units.copy()
    if ranks is not None and not filtered_units.empty: # Added check
         # Ensure ranks is treated as a list or handled correctly by filter_units
         # filter_units already has logic to handle int or str, but let's be explicit
         filtered_ranks = filter_units(filtered_units, ranks)


    # Final filtering
    series = unit_series.copy()
    if remove:
        # Need to combine the indices filtered by units and ranks for removal
        if units is not None and ranks is not None:
             # Elements to remove are those *not* in filtered_units OR *not* in filtered_ranks
             # This logic seems complex, let's simplify based on original adv_filter_keys logic
             # The original intent was to remove elements that *match* the filter criteria if remove=True
             # Let's go back to that logic but ensure the filter_units call is robust
             filtered_by_both = unit_series.copy()
             if units is not None:
                 filtered_by_both = filter_units(filtered_by_both, units)
             if ranks is not None and not filtered_by_both.empty: # Added check
                 filtered_by_both = filter_units(filtered_by_both, ranks)
             # Remove elements that match the criteria
             series = series[~series.index.isin(filtered_by_both.index)]
        elif units is not None:
            filtered_by_units = filter_units(unit_series, units)
            series = series[~series.index.isin(filtered_by_units.index)]
        elif ranks is not None:
            filtered_by_ranks = filter_units(unit_series, ranks)
            series = series[~series.index.isin(filtered_by_ranks.index)]
        else:
            # If remove=True but no units or ranks are specified, what should happen?
            # The original code didn't handle this explicitly. Let's assume it means remove nothing.
            pass # Do nothing if remove is True but no filter criteria are given
    else: # remove is False, return elements that match criteria
        series = filtered_ranks # filtered_ranks already holds the elements matching both criteria

    return series


# Will spam read all knowledge in knowledge base for free gold, roughly 3k, 100 gems
def read_knowledge(bot):
    spam_click = range(1000)
    for i in spam_click:
        bot.click(450, 1300, 0.1)


def get_button_pos(df, button):
    #button=button+'.png'
    pos = df[df['icon'] == button]['pos [X,Y]'].reset_index(drop=True)[0]
    return np.array(pos)

[mleem97]

U can put the file in as a Pull Request so I will be able to integrate it :)