diff --git a/.gitignore b/.gitignore index 71afaed..c729144 100644 --- a/.gitignore +++ b/.gitignore @@ -1,9 +1,10 @@ .idea/ -clashroyalebuildabot/debug .venv .vscode *.egg-info dist/ build/ __pycache__/ -*.pyc \ No newline at end of file +*.pyc +clashroyalebuildabot/debug +clashroyalebuildabot/emulator/platform-tools \ No newline at end of file diff --git a/clashroyalebuildabot/bot/example/custom_bot.py b/clashroyalebuildabot/bot/example/custom_bot.py index 7ce2050..00aa294 100644 --- a/clashroyalebuildabot/bot/example/custom_bot.py +++ b/clashroyalebuildabot/bot/example/custom_bot.py @@ -110,5 +110,5 @@ def run(self): while True: self.step() except (KeyboardInterrupt, Exception): - self.emulator.blitz_device.quit() + self.emulator.quit() logger.info("Thanks for using CRBAB, see you next time!") diff --git a/clashroyalebuildabot/config.yaml b/clashroyalebuildabot/config.yaml index e988462..bd24a6e 100644 --- a/clashroyalebuildabot/config.yaml +++ b/clashroyalebuildabot/config.yaml @@ -10,11 +10,6 @@ adb: # make sure to use the device's IP address. ip: "127.0.0.1" - # The ADB port your emulator is open on. - # Should be 5555 in most cases. If you're unable to get this to work, - # visit X/wiki/ADB-Port for more information. - port: 5555 - # The serial number of your device # Use adb devices to obtain it device_serial: emulator-5554 diff --git a/clashroyalebuildabot/constants.py b/clashroyalebuildabot/constants.py index 8c4fb68..aa6f306 100644 --- a/clashroyalebuildabot/constants.py +++ b/clashroyalebuildabot/constants.py @@ -8,6 +8,8 @@ MODELS_DIR = os.path.join(SRC_DIR, "models") IMAGES_DIR = os.path.join(SRC_DIR, "images") EMULATOR_DIR = os.path.join(SRC_DIR, "emulator") +ADB_DIR = os.path.join(EMULATOR_DIR, "platform-tools") +ADB_PATH = os.path.normpath(os.path.join(ADB_DIR, "adb")) SCREENSHOTS_DIR = os.path.join(DEBUG_DIR, "screenshots") LABELS_DIR = os.path.join(DEBUG_DIR, "labels") diff --git a/clashroyalebuildabot/emulator/adbblitz.py b/clashroyalebuildabot/emulator/adbblitz.py deleted file mode 100644 index 39d172c..0000000 --- a/clashroyalebuildabot/emulator/adbblitz.py +++ /dev/null @@ -1,217 +0,0 @@ -# pylint: disable=R1732 - -import atexit -from contextlib import contextmanager -import shutil -import socket -import subprocess -import time - -import av -from get_free_port import get_dynamic_ports -import kthread -from subprocesskiller import kill_pid -from subprocesskiller import kill_process_children_parents -from subprocesskiller import kill_subprocs - -from clashroyalebuildabot.constants import EMULATOR_DIR - - -@atexit.register -def kill_them_all(): - kill_subprocs(dontkill=()) - - -@contextmanager -def ignored(*exceptions): - try: - yield - except exceptions: - pass - - -class AdbShotTCP: - def __init__( - self, - device_serial, - max_video_width, - ip="127.0.0.1", - port=5555, - ): - r"""Class for capturing screenshots from an Android device over TCP/IP using ADB. - - Args: - device_serial (str): Serial number or IP address of the target device. - ip (str, optional): IP address of the device. Defaults to "127.0.0.1". - port (int, optional): Port number to connect to the device. Defaults to 5555. - - Raises: - Exception: If connection to the device fails. - - Methods: - quit(): Stops capturing and closes the connection to the device. - take_screenshot(): Retrieves a screenshot from the device. - __exit__(exc_type, exc_value, traceback): Context manager exit point. - """ - self.device_serial = device_serial - self.max_video_width = max_video_width - self.ip = ip - self.port = port - - self.adb_path = shutil.which("adb") - if self.adb_path is None: - raise ValueError("ADB is not on the PATH") - self.video_socket = None - self.screenshot_thread = None - self.screenshot = None - self.scrcpy_proc = None - self.codec = av.codec.CodecContext.create("h264", "r") - self.forward_port = get_dynamic_ports(qty=1)[0] - - self._copy_scrcpy() - self._forward_port() - self._start_scrcpy() - self._connect_to_server() - self._start_capturing() - - def __exit__(self, exc_type, exc_value, traceback): - self.quit() - - def _copy_scrcpy(self): - subprocess.run( - [ - self.adb_path, - "-s", - self.device_serial, - "push", - "scrcpy-server.jar", - "/data/local/tmp/", - ], - check=True, - capture_output=True, - cwd=EMULATOR_DIR, - ) - - def _start_scrcpy(self): - command = [ - self.adb_path, - "-s", - self.device_serial, - "shell", - "CLASSPATH=/data/local/tmp/scrcpy-server.jar", - "app_process", - "/", - "com.genymobile.scrcpy.Server", - "2.0", - "tunnel_forward=true", - "control=false", - "cleanup=true", - "clipboard_autosync=false", - "video_bit_rate=8000000", - "audio=false", - "lock_video_orientation=0", - "downsize_on_error=false", - "send_dummy_byte=true", - "raw_video_stream=true", - f"max_size={self.max_video_width}", - ] - self.scrcpy_proc = subprocess.Popen( - command, - stderr=subprocess.PIPE, - stdout=subprocess.PIPE, - cwd=EMULATOR_DIR, - ) - - def _forward_port(self): - subprocess.run( - [ - self.adb_path, - "-s", - self.device_serial, - "forward", - f"tcp:{self.forward_port}", - "localabstract:scrcpy", - ], - cwd=EMULATOR_DIR, - capture_output=True, - check=True, - ) - - def _connect_to_server(self): - dummy_byte = b"" - while not dummy_byte: - with ignored(Exception): - self.video_socket = socket.socket( - socket.AF_INET, socket.SOCK_STREAM - ) - self.video_socket.connect((self.ip, self.forward_port)) - - self.video_socket.setblocking(False) - self.video_socket.settimeout(1) - dummy_byte = self.video_socket.recv(1) - if len(dummy_byte) == 0: - self.video_socket.close() - - def _start_capturing(self): - self.screenshot_thread = kthread.KThread( - target=self._update_screenshot, name="update_screenshot_thread" - ) - self.screenshot_thread.start() - - def _update_screenshot(self): - while True: - with ignored(Exception): - packets = self.codec.parse(self.video_socket.recv(131072)) - if len(packets) == 0: - continue - - frames = self.codec.decode(packets[-1]) - if len(frames) == 0: - continue - - frame = frames[-1] - self.screenshot = ( - frame.to_rgb() - .reformat( - width=frame.width, height=frame.height, format="rgb24" - ) - .to_ndarray() - ) - - def quit(self): - while self.screenshot_thread.is_alive(): - with ignored(Exception): - self.screenshot_thread.kill() - self.video_socket.close() - - with ignored(Exception): - self.scrcpy_proc.stdout.close() - - with ignored(Exception): - self.scrcpy_proc.stdin.close() - - with ignored(Exception): - self.scrcpy_proc.stderr.close() - - with ignored(Exception): - self.scrcpy_proc.wait(timeout=2) - - with ignored(Exception): - self.scrcpy_proc.kill() - - with ignored(Exception): - kill_process_children_parents( - pid=self.scrcpy_proc.pid, max_parent_exe="adb.exe", dontkill=() - ) - time.sleep(2) - - with ignored(Exception): - kill_pid(pid=self.scrcpy_proc.pid) - - def take_screenshot(self): - while self.screenshot is None: - time.sleep(0.01) - screenshot = self.screenshot - self.screenshot = None - - return screenshot diff --git a/clashroyalebuildabot/emulator/emulator.py b/clashroyalebuildabot/emulator/emulator.py index b561968..07cf3ac 100644 --- a/clashroyalebuildabot/emulator/emulator.py +++ b/clashroyalebuildabot/emulator/emulator.py @@ -1,14 +1,44 @@ +# pylint: disable=R1732 + +import atexit +from contextlib import contextmanager import os +import platform +import socket +import subprocess +import time +import zipfile -from adb_shell.adb_device import AdbDeviceTcp +import av +from get_free_port import get_dynamic_ports +import kthread from loguru import logger from PIL import Image +import requests +from subprocesskiller import kill_pid +from subprocesskiller import kill_process_children_parents +from subprocesskiller import kill_subprocs import yaml +from clashroyalebuildabot.constants import ADB_DIR +from clashroyalebuildabot.constants import ADB_PATH +from clashroyalebuildabot.constants import EMULATOR_DIR from clashroyalebuildabot.constants import SCREENSHOT_HEIGHT from clashroyalebuildabot.constants import SCREENSHOT_WIDTH from clashroyalebuildabot.constants import SRC_DIR -from clashroyalebuildabot.emulator.adbblitz import AdbShotTCP + + +@atexit.register +def kill_them_all(): + kill_subprocs() + + +@contextmanager +def ignored(*exceptions): + try: + yield + except exceptions: + pass class Emulator: @@ -18,45 +48,195 @@ def __init__(self): config = yaml.safe_load(file) adb_config = config["adb"] - serial, ip, port = [ - adb_config[s] for s in ["device_serial", "ip", "port"] - ] + self.serial, self.ip = [adb_config[s] for s in ["device_serial", "ip"]] + + self.video_socket = None + self.screenshot_thread = None + self.frame = None + self.scrcpy_proc = None + self.codec = av.codec.CodecContext.create("h264", "r") + self.forward_port = get_dynamic_ports(qty=1)[0] + + self._install_adb() + self.width, self.height = self._get_width_and_height() + self._copy_scrcpy() + self._forward_port() + self._start_scrcpy() + self._connect_to_server() + self._start_capturing() + + @staticmethod + def _install_adb(): + if os.path.isdir(ADB_DIR): + return + + os_name = platform.system().lower() + adb_url = f"https://dl.google.com/android/repository/platform-tools-latest-{os_name}.zip" + zip_path = f"platform-tools-latest-{os_name}.zip" + + response = requests.get(adb_url, stream=True, timeout=60) + response.raise_for_status() + + with open(zip_path, "wb") as file: + for chunk in response.iter_content(chunk_size=1024): + if chunk: + file.write(chunk) + + with zipfile.ZipFile(zip_path, "r") as zip_ref: + zip_ref.extractall(EMULATOR_DIR) + + def __exit__(self, exc_type, exc_value, traceback): + self.quit() + + def quit(self): + while self.screenshot_thread.is_alive(): + with ignored(Exception): + self.screenshot_thread.kill() + self.video_socket.close() + + with ignored(Exception): + self.scrcpy_proc.stdout.close() + + with ignored(Exception): + self.scrcpy_proc.stdin.close() + + with ignored(Exception): + self.scrcpy_proc.stderr.close() + + with ignored(Exception): + self.scrcpy_proc.wait(timeout=2) + + with ignored(Exception): + self.scrcpy_proc.kill() + + with ignored(Exception): + kill_process_children_parents( + pid=self.scrcpy_proc.pid, max_parent_exe="adb.exe", dontkill=() + ) + time.sleep(2) + + with ignored(Exception): + kill_pid(pid=self.scrcpy_proc.pid) - self.device = AdbDeviceTcp(ip, port) + def _run_command(self, command): + command = [ADB_PATH, "-s", self.serial, *command] + logger.debug(" ".join(command)) try: - self.device.connect() - window_size = self.device.shell("wm size") - window_size = window_size.replace("Physical size: ", "") - self.size = tuple(int(i) for i in window_size.split("x")) - except Exception as e: - logger.critical(f"Error getting screen size: {e}") - logger.critical("Exiting due to device connection error.") - raise SystemExit() from e - - self.blitz_device = AdbShotTCP( - device_serial=serial, - ip=ip, - max_video_width=self.size[0], + result = subprocess.run( + command, + cwd=EMULATOR_DIR, + capture_output=True, + check=True, + text=True, + ) + except subprocess.CalledProcessError as e: + logger.error(f"Error executing command: {e}") + logger.error(f"Output: {e.stdout}") + logger.error(f"Error output: {e.stderr}") + self.quit() + raise + + if result.returncode != 0: + logger.error(f"Error executing command: {result.stderr}") + self.quit() + raise RuntimeError("ADB command failed") + + return result.stdout + + def _copy_scrcpy(self): + self._run_command(["push", "scrcpy-server.jar", "/data/local/tmp/"]) + + def _start_scrcpy(self): + command = [ + ADB_PATH, + "-s", + self.serial, + "shell", + "CLASSPATH=/data/local/tmp/scrcpy-server.jar", + "app_process", + "/", + "com.genymobile.scrcpy.Server", + "2.0", + "tunnel_forward=true", + "control=false", + "cleanup=true", + "clipboard_autosync=false", + "video_bit_rate=8000000", + "audio=false", + "lock_video_orientation=0", + "downsize_on_error=false", + "send_dummy_byte=true", + "raw_video_stream=true", + f"max_size={self.width}", + ] + self.scrcpy_proc = subprocess.Popen( + command, + stderr=subprocess.PIPE, + stdout=subprocess.PIPE, + cwd=EMULATOR_DIR, ) - def click(self, x, y): - self.device.shell(f"input tap {x} {y}") + def _forward_port(self): + self._run_command( + ["forward", f"tcp:{self.forward_port}", "localabstract:scrcpy"] + ) - def _take_screenshot(self): - image = self.blitz_device.take_screenshot() - image = Image.fromarray(image) - image = image.resize( - (SCREENSHOT_WIDTH, SCREENSHOT_HEIGHT), Image.Resampling.BILINEAR + def _connect_to_server(self): + dummy_byte = b"" + while not dummy_byte: + with ignored(Exception): + self.video_socket = socket.socket( + socket.AF_INET, socket.SOCK_STREAM + ) + self.video_socket.connect((self.ip, self.forward_port)) + + self.video_socket.setblocking(False) + self.video_socket.settimeout(1) + dummy_byte = self.video_socket.recv(1) + if len(dummy_byte) == 0: + self.video_socket.close() + + def _start_capturing(self): + self.screenshot_thread = kthread.KThread( + target=self._update_screenshot, name="update_screenshot_thread" ) + self.screenshot_thread.start() - return image + def _update_screenshot(self): + while True: + with ignored(Exception): + packets = self.codec.parse(self.video_socket.recv(131072)) + if len(packets) == 0: + continue + + frames = self.codec.decode(packets[-1]) + if len(frames) == 0: + continue + + self.frame = frames[-1] + + def _get_width_and_height(self): + window_size = self._run_command(["shell", "wm", "size"]) + window_size = window_size.replace("Physical size: ", "") + width, height = tuple(int(i) for i in window_size.split("x")) + return width, height + + def click(self, x, y): + self._run_command(["shell", "input", "tap", str(x), str(y)]) def take_screenshot(self): logger.debug("Starting to take screenshot...") - try: - image = self._take_screenshot() - except Exception as e: - logger.error(f"ADB command failed: {e}") - raise + while self.frame is None: + time.sleep(0.01) + frame, self.frame = self.frame, None + screenshot = ( + frame.to_rgb() + .reformat(width=frame.width, height=frame.height, format="rgb24") + .to_ndarray() + ) + screenshot = Image.fromarray(screenshot) + screenshot = screenshot.resize( + (SCREENSHOT_WIDTH, SCREENSHOT_HEIGHT), Image.Resampling.BILINEAR + ) - return image + return screenshot diff --git a/clashroyalebuildabot/screen.py b/clashroyalebuildabot/screen.py deleted file mode 100644 index 2b9dd34..0000000 --- a/clashroyalebuildabot/screen.py +++ /dev/null @@ -1,104 +0,0 @@ -import os -import tempfile - -from adb_shell.adb_device import AdbDeviceTcp -from loguru import logger -from PIL import Image -import yaml - -from clashroyalebuildabot.constants import SRC_DIR - - -# Load configuration from config.yaml -def load_config(): - config_path = os.path.join(SRC_DIR, "config.yaml") - with open( - config_path, "r", encoding="utf-8" - ) as file: # Specify encoding here - config = yaml.safe_load(file) - return config - - -# Initialize ADB connection -def init_adb_connection(host, port): - device = AdbDeviceTcp(host, port, default_transport_timeout_s=9.0) - device.connect() - return device - - -# Get emulator resolution -def get_emulator_resolution(device): - result = device.shell("wm size") - return result.strip() - - -# Get emulator density -def get_emulator_density(device): - result = device.shell("wm density") - return result.strip() - - -# Take a screenshot -def take_screenshot(device): - screenshot_path = "/sdcard/screen.png" - device.shell(f"screencap -p {screenshot_path}") - return screenshot_path - - -# Pull the screenshot to a temporary file -def pull_screenshot(device, remote_path): - local_fd, local_path = tempfile.mkstemp(suffix=".png") - os.close(local_fd) # Close the file descriptor immediately - device.pull(remote_path, local_path) - return local_path - - -# Open the screenshot -def open_screenshot(local_path): - img = Image.open(local_path) - img.show() - - -# Delete the remote and local screenshot files -def delete_screenshot(device, remote_path, local_path): - device.shell(f"rm {remote_path}") - os.remove(local_path) - - -# Check emulator properties -def check_emulator_properties(): - config = load_config() - adb_config = config["adb"] - - device = init_adb_connection(adb_config["ip"], adb_config["port"]) - - resolution = get_emulator_resolution(device) - density = get_emulator_density(device) - - resolution_correct = "720x1280" in resolution - density_correct = "240" in density - - if resolution_correct and density_correct: - logger.info( - "The emulator has the correct resolution (720x1280) and density (240 dpi)." - ) - - screenshot_path = take_screenshot(device) - local_screenshot_path = pull_screenshot(device, screenshot_path) - open_screenshot(local_screenshot_path) - - delete_screenshot(device, screenshot_path, local_screenshot_path) - - logger.info( - "ClashRoyaleBuildABot can now be started with `python main.py`." - ) - else: - logger.info("The emulator does not have the correct properties.") - if not resolution_correct: - logger.info(f"Current resolution: {resolution}") - if not density_correct: - logger.info(f"Current density: {density}") - - -if __name__ == "__main__": - check_emulator_properties() diff --git a/pyproject.toml b/pyproject.toml index 96f54f3..394ba5e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -20,11 +20,9 @@ dependencies = [ "scipy>=1.13.1", "rich>=13.7.1", "loguru>=0.7.2", - "adb-shell[async]==0.4.4", "PyYAML", "pybind11>=2.12", "requests>=2.25.1", - "adbutils", "av", "get_free_port", "kthread", @@ -38,7 +36,6 @@ dev = [ "flake8==7.0.0", "isort==5.13.2", "pylint==3.1.0", - "adb-shell[async]==0.4.4", ] [tool.black]