Devices API

ADB Device

pymordialdroid.devices.adb_device

Android Debug Bridge (ADB) device implementation for PymordialDroid.

AdbDevice

Bases: PymordialBridgeDevice

Handles Android device communication via pure Python ADB.

Fulfills the PymordialBridgeDevice contract with native package resolution, activity detection, focused window parsing, and resilient app lifecycle commands.

Source code in .venv/lib/python3.13/site-packages/pymordialdroid/devices/adb_device.py

class AdbDevice(PymordialBridgeDevice):
    """Handles Android device communication via pure Python ADB.

    Fulfills the PymordialBridgeDevice contract with native package resolution,
    activity detection, focused window parsing, and resilient app lifecycle commands.
    """

    name: str = "adb"
    version: str = "0.1.0"

    def __init__(
        self,
        host: str = "127.0.0.1",
        port: int = 5555,
        signer: PythonRSASigner | None = None,
        system_config: SystemConfig | None = None,
    ) -> None:
        self.host = host
        self.port = port
        self.system_config = system_config or resolve_system_config()
        self.signer = signer or self.system_config.get_signer()

        self._device: AdbDeviceTcp | None = None
        self._latest_frame: bytes | None = None
        self._is_streaming: bool = False
        self._frame_provider: Any = None

    def initialize(self, config: Any = None) -> None:
        """Initializes the ADB device plugin."""
        pass

    def shutdown(self) -> None:
        """Disconnects and cleans up resources."""
        self.disconnect()

    # --- CONNECTION MANAGEMENT ---

    def _try_auto_heal_tcpip(self) -> bool:
        """Attempts to auto-heal synchronous ADB connection via CLI binary."""
        import subprocess

        adb_bin = str(self.system_config.adb_bin_path)
        log.info(f"Attempting auto-heal for ADB device {self.host}:{self.port}...")
        try:
            if self.port != 5555:
                endpoint = f"{self.host}:{self.port}"
                subprocess.run(
                    [adb_bin, "connect", endpoint],
                    capture_output=True,
                    text=True,
                    timeout=5,
                    check=False,
                )
                subprocess.run(
                    [adb_bin, "-s", endpoint, "tcpip", "5555"],
                    capture_output=True,
                    text=True,
                    timeout=5,
                    check=False,
                )

            subprocess.run(
                [adb_bin, "connect", f"{self.host}:5555"],
                capture_output=True,
                text=True,
                timeout=5,
                check=False,
            )

            self.port = 5555
            self._device = AdbDeviceTcp(self.host, 5555, default_transport_timeout_s=9)
            self._device.connect(rsa_keys=[self.signer], auth_timeout_s=5)
            log.info(f"Auto-healed ADB connection to {self.host}:5555")
            return True
        except Exception as e:
            log.warning(f"AdbDevice auto-heal failed: {e}")
            self._device = None
            return False

    def connect(self, auto_heal: bool = True) -> bool:
        """Connects to the Android device via TCP socket with RSA authentication."""
        log.debug(f"Connecting ADB device to {self.host}:{self.port}...")
        if self._device is None:
            self._device = AdbDeviceTcp(
                self.host, self.port, default_transport_timeout_s=9
            )

        if self._device.available:
            return True

        try:
            self._device.connect(rsa_keys=[self.signer], auth_timeout_s=5)
            log.info(f"Connected to device {self.host}:{self.port}")
            return True
        except Exception as e:
            log.warning(f"Error connecting to ADB device {self.host}:{self.port}: {e}")
            self._device = None
            if auto_heal and self._try_auto_heal_tcpip():
                return True
            return False

    def ensure_cli_endpoint(self) -> bool:
        """Registers this endpoint with the CLI adb server (`adb connect`).

        Required before launching ``scrcpy`` binaries / live-stream helpers,
        which shell out to ``adb.exe`` and cannot see pure-Python
        :class:`AdbDeviceTcp` sockets. Best-effort: never raises.
        """
        import subprocess

        try:
            subprocess.run(
                [
                    str(self.system_config.adb_bin_path),
                    "connect",
                    f"{self.host}:{self.port}",
                ],
                check=False,
                stdout=subprocess.DEVNULL,
                stderr=subprocess.DEVNULL,
                timeout=5,
            )
            return True
        except Exception:
            return False

    def is_connected(self) -> bool:
        """Checks if ADB connection is currently active and responsive."""
        return self._device is not None and self._device.available

    def disconnect(self) -> bool:
        """Disconnects the ADB device."""
        self.stop_stream()
        if self._device is None:
            return True
        try:
            self._device.close()
            self._device = None
            log.debug(f"Disconnected from ADB device {self.host}:{self.port}")
            return True
        except Exception as e:
            log.error(f"Error disconnecting ADB device: {e}")
            return False

    def run_command(self, command: str, decode: bool = True) -> str | bytes | None:
        """Executes a shell command on the device."""
        if not self.is_connected():
            if not self.connect():
                return None
        try:
            output = self._device.shell(command, decode=decode)
            return output.strip() if decode and isinstance(output, str) else output
        except Exception as e:
            log.error(f"Failed to execute command '{command}': {e}")
            return None

    # --- ANDROID PACKAGE & ACTIVITY RESOLUTION ---

    @staticmethod
    def parse_package_list(output: str) -> list[str]:
        """Extracts package names from `pm list packages` output.

        Only ``package:``-prefixed lines are kept; device warnings (e.g.
        Samsung multi-user ``SecurityException`` preambles) are ignored so
        they can never match a keyword or be force-stopped.
        """
        packages = []
        for line in output.splitlines():
            stripped = line.strip()
            if stripped.startswith("package:"):
                name = stripped[len("package:") :].strip()
                if name:
                    packages.append(name)
        return packages

    @staticmethod
    def rank_package(keyword: str, packages: list[str]) -> str | None:
        """Ranks packages for a keyword: exact > component > substring.

        1. Case-insensitive exact match.
        2. Keyword equals a dot-separated component (``settings`` matches
           ``com.android.settings`` but not ``com.sec.usbsettings``);
           ties break toward the shortest full name.
        3. Plain substring match; ties break toward the shortest full name.
        """
        kw = keyword.lower()
        for pkg in packages:
            if pkg.lower() == kw:
                return pkg
        component = [p for p in packages if kw in p.lower().split(".")]
        if component:
            return min(component, key=len)
        substring = [p for p in packages if kw in p.lower()]
        if substring:
            return min(substring, key=len)
        return None

    def find_package_by_keyword(self, keyword: str) -> str | None:
        """Finds an installed package matching a keyword using 'pm list packages'."""
        output = self.run_command("pm list packages", decode=True)
        if not output or not isinstance(output, str):
            return None

        packages = self.parse_package_list(output)

        # 1. Exact match (preserves historical fast path)
        if keyword in packages:
            return keyword

        return self.rank_package(keyword, packages)

    def get_launch_activity(self, package_name: str) -> str | None:
        """Queries Android's activity manager to determine the exact launchable activity."""
        cmd = f"cmd package resolve-activity --brief {package_name}"
        output = self.run_command(cmd, decode=True)
        if not output or not isinstance(output, str):
            return None

        lines = output.strip().splitlines()
        if lines:
            activity = lines[-1].strip()
            if "/" in activity and "No activity found" not in activity:
                return activity

        return None

    def get_focused_app(self) -> dict[str, str] | None:
        """Parses 'dumpsys window' to detect the currently focused package and activity."""
        output = self.run_command(
            "dumpsys window | grep -E 'mCurrentFocus|mFocusedApp'", decode=True
        )
        if not output or not isinstance(output, str):
            return None

        match = re.search(r"([a-zA-Z0-9._]+)/([a-zA-Z0-9._$]+)", output)
        if match:
            pkg, activity = match.groups()
            return {"package": pkg, "activity": activity}

        return None

    # --- APPLICATION LIFECYCLE ---

    def open_app(
        self,
        package_name: str,
        app_name: str | None = None,
        timeout: float = 10.0,
        wait_time: float = 1.0,
    ) -> bool:
        """Launches an app via resolved Activity or Monkey fallback, verifying it started."""
        pkg = package_name or (
            self.find_package_by_keyword(app_name) if app_name else None
        )
        if not pkg:
            log.error(f"Could not resolve package for app: {app_name}")
            return False

        # 1. Try activity-based launch first (fast and deterministic)
        activity = self.get_launch_activity(pkg)
        if activity:
            self.run_command(f"am start -n {activity}")
            log.debug(f"Launched {pkg} via activity: {activity}")
        else:
            # 2. Fallback to Monkey intent launcher
            self.run_command(f"monkey -p {pkg} -c android.intent.category.LAUNCHER 1")
            log.debug(f"Launched {pkg} via Monkey fallback")

        # Verify app is running within timeout
        start_time = time.time()
        while time.time() - start_time < timeout:
            if self.is_app_running(pkg, max_retries=1, wait_time=0):
                return True
            time.sleep(wait_time)

        log.warning(f"App {pkg} did not start within {timeout}s")
        return False

    def close_app(
        self,
        package_name: str | None = None,
        app_name: str | None = None,
        timeout: float = 5.0,
        wait_time: float = 0.5,
    ) -> bool:
        """Force stops an app and polls pidof to confirm closure."""
        pkg = package_name or (
            self.find_package_by_keyword(app_name) if app_name else None
        )
        if not pkg:
            log.error(f"Could not resolve package to close for app: {app_name}")
            return False

        self.run_command(f"am force-stop {pkg}")

        start_time = time.time()
        while time.time() - start_time < timeout:
            if not self.is_app_running(pkg, max_retries=1, wait_time=0):
                return True
            time.sleep(wait_time)

        return not self.is_app_running(pkg, max_retries=1, wait_time=0)

    def is_app_running(
        self,
        package_name: str | None = None,
        app_name: str | None = None,
        max_retries: int = 2,
        wait_time: float = 1.0,
    ) -> bool:
        """Checks if an app is actively running by querying process PID."""
        pkg = package_name or (
            self.find_package_by_keyword(app_name) if app_name else None
        )
        if not pkg:
            return False

        for attempt in range(max_retries):
            output = self.run_command(f"pidof {pkg}", decode=True)
            if output and str(output).strip():
                return True
            if attempt < max_retries - 1 and wait_time > 0:
                time.sleep(wait_time)

        return False

    def show_recent_apps(self) -> bool:
        """Opens recent apps / overview."""
        res = self.run_command("input keyevent 187")
        return res is not None

    def close_all_apps(self, exclude: list[str] | None = None) -> int:
        """Force stops all installed third-party/user packages to clear device state."""
        output = self.run_command("pm list packages", decode=True)
        if not output or not isinstance(output, str):
            return 0

        packages = self.parse_package_list(output)
        exclude_list = exclude or []
        count = 0

        for pkg in packages:
            if pkg in exclude_list:
                continue
            self.run_command(f"am force-stop {pkg}")
            count += 1

        log.debug(f"Closed {count} applications.")
        return count

    def get_current_app(self) -> str | None:
        """Gets the package name of the currently focused application."""
        focused = self.get_focused_app()
        if focused and "package" in focused:
            return focused["package"]
        return None

    # --- USER INPUT ACTIONS ---

    def tap(
        self,
        coords: tuple[int, int] | list[int] | int,
        y: int | None = None,
        times: int = 1,
    ) -> bool:
        """Sends tap events to coordinates on screen.

        Accepts either a tuple/list (x, y) or separate x, y integers.
        """
        if isinstance(coords, (tuple, list)):
            target_x, target_y = coords[0], coords[1]
        elif y is not None:
            target_x, target_y = coords, y
        else:
            log.warning(f"Invalid tap coordinates: coords={coords}, y={y}")
            return False

        for _ in range(times):
            self.run_command(f"input tap {target_x} {target_y}")
            if times > 1:
                time.sleep(0.1)
        return True

    def swipe(
        self,
        start_x: int,
        start_y: int,
        end_x: int,
        end_y: int,
        duration: int = 300,
    ) -> bool:
        """Performs a touch swipe gesture from start to end coordinates."""
        res = self.run_command(
            f"input swipe {start_x} {start_y} {end_x} {end_y} {duration}"
        )
        return res is not None

    def type_text(self, text: str, enter: bool = False) -> bool:
        """Types text into the focused input field, escaping shell characters."""
        # Escape characters that could break ADB shell input
        escaped_text = text.replace("\\", "\\\\")
        escaped_text = escaped_text.replace(" ", "%s")
        escaped_text = escaped_text.replace("'", "\\'")
        escaped_text = escaped_text.replace('"', '\\"')
        escaped_text = escaped_text.replace("&", "\\&")
        escaped_text = escaped_text.replace("<", "\\<")
        escaped_text = escaped_text.replace(">", "\\>")
        escaped_text = escaped_text.replace(";", "\\;")
        escaped_text = escaped_text.replace("|", "\\|")

        res = self.run_command(f"input text '{escaped_text}'")
        if enter:
            self.press_enter()
        return res is not None

    def go_home(self) -> None:
        """Simulates Home button press."""
        self.run_command("input keyevent 3")

    def go_back(self) -> None:
        """Simulates Back button press."""
        self.run_command("input keyevent 4")

    def press_enter(self) -> None:
        """Simulates Enter key press."""
        self.run_command("input keyevent 66")

    def press_esc(self) -> None:
        """Simulates Back / Escape key press."""
        self.run_command("input keyevent 4")

    # --- SCREENSHOT & STREAMING (Pure ADB, No PyAV) ---

    def set_frame_provider(self, provider: Any | None) -> None:
        """Sets an optional headless-feed frame provider.

        The provider is a zero-arg callable returning PNG bytes (or None).
        Typically ``ScrcpyDevice.get_latest_frame``. When set,
        :meth:`capture_screenshot` tries it first for a high-FPS frame and
        falls back to ``screencap -p`` on None/exception.
        """
        self._frame_provider = provider

    def clear_frame_provider(self) -> None:
        """Removes the headless-feed frame provider."""
        self._frame_provider = None

    def capture_screenshot(self) -> bytes | None:
        """Captures a PNG screenshot, preferring the scrcpy headless feed.

        Tries the frame provider first (fast path); falls back to pure
        'screencap -p' so behavior never regresses when no feed is running.
        """
        if self._frame_provider is not None:
            try:
                frame = self._frame_provider()
            except Exception as e:
                log.debug(f"Frame provider failed, falling back to screencap: {e}")
                frame = None
            if frame:
                self._latest_frame = frame
                return frame
        if not self.is_connected():
            if not self.connect():
                return None
        try:
            raw_png = self._device.shell("screencap -p", decode=False)
            if raw_png:
                self._latest_frame = raw_png
                return raw_png
        except Exception as e:
            log.error(f"Screenshot capture failed: {e}")
        return None

    def start_stream(self) -> None:
        """Starts stream mode (tracks frames from screenshot captures)."""
        self._is_streaming = True

    def stop_stream(self) -> None:
        """Stops stream mode."""
        self._is_streaming = False

    def get_latest_frame(self) -> bytes | None:
        """Retrieves the most recent frame bytes or captures a fresh screenshot."""
        if self._latest_frame is None:
            return self.capture_screenshot()
        return self._latest_frame

    # --- APK INSTALLATION ---

    def install_apk(self, apk_path: str | Path, update: bool = True) -> bool:
        """Pushes and installs an APK file via adb."""
        path_obj = Path(apk_path)
        if not path_obj.exists():
            log.warning(f"APK not found: {apk_path}")
            return False

        adb = str(self.system_config.adb_bin_path)
        cmd = [adb, "-s", f"{self.host}:{self.port}", "install"]
        if update:
            cmd.append("-r")
        cmd.append(str(path_obj))

        import subprocess

        proc = subprocess.run(cmd, capture_output=True, text=True, check=False)
        return proc.returncode == 0

capture_screenshot()

Captures a PNG screenshot, preferring the scrcpy headless feed.

Tries the frame provider first (fast path); falls back to pure 'screencap -p' so behavior never regresses when no feed is running.

Source code in .venv/lib/python3.13/site-packages/pymordialdroid/devices/adb_device.py

def capture_screenshot(self) -> bytes | None:
    """Captures a PNG screenshot, preferring the scrcpy headless feed.

    Tries the frame provider first (fast path); falls back to pure
    'screencap -p' so behavior never regresses when no feed is running.
    """
    if self._frame_provider is not None:
        try:
            frame = self._frame_provider()
        except Exception as e:
            log.debug(f"Frame provider failed, falling back to screencap: {e}")
            frame = None
        if frame:
            self._latest_frame = frame
            return frame
    if not self.is_connected():
        if not self.connect():
            return None
    try:
        raw_png = self._device.shell("screencap -p", decode=False)
        if raw_png:
            self._latest_frame = raw_png
            return raw_png
    except Exception as e:
        log.error(f"Screenshot capture failed: {e}")
    return None

clear_frame_provider()

Removes the headless-feed frame provider.

Source code in .venv/lib/python3.13/site-packages/pymordialdroid/devices/adb_device.py

def clear_frame_provider(self) -> None:
    """Removes the headless-feed frame provider."""
    self._frame_provider = None

close_all_apps(exclude=None)

Force stops all installed third-party/user packages to clear device state.

Source code in .venv/lib/python3.13/site-packages/pymordialdroid/devices/adb_device.py

def close_all_apps(self, exclude: list[str] | None = None) -> int:
    """Force stops all installed third-party/user packages to clear device state."""
    output = self.run_command("pm list packages", decode=True)
    if not output or not isinstance(output, str):
        return 0

    packages = self.parse_package_list(output)
    exclude_list = exclude or []
    count = 0

    for pkg in packages:
        if pkg in exclude_list:
            continue
        self.run_command(f"am force-stop {pkg}")
        count += 1

    log.debug(f"Closed {count} applications.")
    return count

close_app(package_name=None, app_name=None, timeout=5.0, wait_time=0.5)

Force stops an app and polls pidof to confirm closure.

Source code in .venv/lib/python3.13/site-packages/pymordialdroid/devices/adb_device.py

def close_app(
    self,
    package_name: str | None = None,
    app_name: str | None = None,
    timeout: float = 5.0,
    wait_time: float = 0.5,
) -> bool:
    """Force stops an app and polls pidof to confirm closure."""
    pkg = package_name or (
        self.find_package_by_keyword(app_name) if app_name else None
    )
    if not pkg:
        log.error(f"Could not resolve package to close for app: {app_name}")
        return False

    self.run_command(f"am force-stop {pkg}")

    start_time = time.time()
    while time.time() - start_time < timeout:
        if not self.is_app_running(pkg, max_retries=1, wait_time=0):
            return True
        time.sleep(wait_time)

    return not self.is_app_running(pkg, max_retries=1, wait_time=0)

connect(auto_heal=True)

Connects to the Android device via TCP socket with RSA authentication.

Source code in .venv/lib/python3.13/site-packages/pymordialdroid/devices/adb_device.py

def connect(self, auto_heal: bool = True) -> bool:
    """Connects to the Android device via TCP socket with RSA authentication."""
    log.debug(f"Connecting ADB device to {self.host}:{self.port}...")
    if self._device is None:
        self._device = AdbDeviceTcp(
            self.host, self.port, default_transport_timeout_s=9
        )

    if self._device.available:
        return True

    try:
        self._device.connect(rsa_keys=[self.signer], auth_timeout_s=5)
        log.info(f"Connected to device {self.host}:{self.port}")
        return True
    except Exception as e:
        log.warning(f"Error connecting to ADB device {self.host}:{self.port}: {e}")
        self._device = None
        if auto_heal and self._try_auto_heal_tcpip():
            return True
        return False

disconnect()

Disconnects the ADB device.

Source code in .venv/lib/python3.13/site-packages/pymordialdroid/devices/adb_device.py

def disconnect(self) -> bool:
    """Disconnects the ADB device."""
    self.stop_stream()
    if self._device is None:
        return True
    try:
        self._device.close()
        self._device = None
        log.debug(f"Disconnected from ADB device {self.host}:{self.port}")
        return True
    except Exception as e:
        log.error(f"Error disconnecting ADB device: {e}")
        return False

ensure_cli_endpoint()

Registers this endpoint with the CLI adb server (adb connect).

Required before launching scrcpy binaries / live-stream helpers, which shell out to adb.exe and cannot see pure-Python :class:AdbDeviceTcp sockets. Best-effort: never raises.

Source code in .venv/lib/python3.13/site-packages/pymordialdroid/devices/adb_device.py

def ensure_cli_endpoint(self) -> bool:
    """Registers this endpoint with the CLI adb server (`adb connect`).

    Required before launching ``scrcpy`` binaries / live-stream helpers,
    which shell out to ``adb.exe`` and cannot see pure-Python
    :class:`AdbDeviceTcp` sockets. Best-effort: never raises.
    """
    import subprocess

    try:
        subprocess.run(
            [
                str(self.system_config.adb_bin_path),
                "connect",
                f"{self.host}:{self.port}",
            ],
            check=False,
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
            timeout=5,
        )
        return True
    except Exception:
        return False

find_package_by_keyword(keyword)

Finds an installed package matching a keyword using 'pm list packages'.

Source code in .venv/lib/python3.13/site-packages/pymordialdroid/devices/adb_device.py

def find_package_by_keyword(self, keyword: str) -> str | None:
    """Finds an installed package matching a keyword using 'pm list packages'."""
    output = self.run_command("pm list packages", decode=True)
    if not output or not isinstance(output, str):
        return None

    packages = self.parse_package_list(output)

    # 1. Exact match (preserves historical fast path)
    if keyword in packages:
        return keyword

    return self.rank_package(keyword, packages)

get_current_app()

Gets the package name of the currently focused application.

Source code in .venv/lib/python3.13/site-packages/pymordialdroid/devices/adb_device.py

def get_current_app(self) -> str | None:
    """Gets the package name of the currently focused application."""
    focused = self.get_focused_app()
    if focused and "package" in focused:
        return focused["package"]
    return None

get_focused_app()

Parses 'dumpsys window' to detect the currently focused package and activity.

Source code in .venv/lib/python3.13/site-packages/pymordialdroid/devices/adb_device.py

def get_focused_app(self) -> dict[str, str] | None:
    """Parses 'dumpsys window' to detect the currently focused package and activity."""
    output = self.run_command(
        "dumpsys window | grep -E 'mCurrentFocus|mFocusedApp'", decode=True
    )
    if not output or not isinstance(output, str):
        return None

    match = re.search(r"([a-zA-Z0-9._]+)/([a-zA-Z0-9._$]+)", output)
    if match:
        pkg, activity = match.groups()
        return {"package": pkg, "activity": activity}

    return None

get_latest_frame()

Retrieves the most recent frame bytes or captures a fresh screenshot.

Source code in .venv/lib/python3.13/site-packages/pymordialdroid/devices/adb_device.py

def get_latest_frame(self) -> bytes | None:
    """Retrieves the most recent frame bytes or captures a fresh screenshot."""
    if self._latest_frame is None:
        return self.capture_screenshot()
    return self._latest_frame

get_launch_activity(package_name)

Queries Android's activity manager to determine the exact launchable activity.

Source code in .venv/lib/python3.13/site-packages/pymordialdroid/devices/adb_device.py

def get_launch_activity(self, package_name: str) -> str | None:
    """Queries Android's activity manager to determine the exact launchable activity."""
    cmd = f"cmd package resolve-activity --brief {package_name}"
    output = self.run_command(cmd, decode=True)
    if not output or not isinstance(output, str):
        return None

    lines = output.strip().splitlines()
    if lines:
        activity = lines[-1].strip()
        if "/" in activity and "No activity found" not in activity:
            return activity

    return None

go_back()

Simulates Back button press.

Source code in .venv/lib/python3.13/site-packages/pymordialdroid/devices/adb_device.py

def go_back(self) -> None:
    """Simulates Back button press."""
    self.run_command("input keyevent 4")

go_home()

Simulates Home button press.

Source code in .venv/lib/python3.13/site-packages/pymordialdroid/devices/adb_device.py

def go_home(self) -> None:
    """Simulates Home button press."""
    self.run_command("input keyevent 3")

initialize(config=None)

Initializes the ADB device plugin.

Source code in .venv/lib/python3.13/site-packages/pymordialdroid/devices/adb_device.py

def initialize(self, config: Any = None) -> None:
    """Initializes the ADB device plugin."""
    pass

install_apk(apk_path, update=True)

Pushes and installs an APK file via adb.

Source code in .venv/lib/python3.13/site-packages/pymordialdroid/devices/adb_device.py

def install_apk(self, apk_path: str | Path, update: bool = True) -> bool:
    """Pushes and installs an APK file via adb."""
    path_obj = Path(apk_path)
    if not path_obj.exists():
        log.warning(f"APK not found: {apk_path}")
        return False

    adb = str(self.system_config.adb_bin_path)
    cmd = [adb, "-s", f"{self.host}:{self.port}", "install"]
    if update:
        cmd.append("-r")
    cmd.append(str(path_obj))

    import subprocess

    proc = subprocess.run(cmd, capture_output=True, text=True, check=False)
    return proc.returncode == 0

is_app_running(package_name=None, app_name=None, max_retries=2, wait_time=1.0)

Checks if an app is actively running by querying process PID.

Source code in .venv/lib/python3.13/site-packages/pymordialdroid/devices/adb_device.py

def is_app_running(
    self,
    package_name: str | None = None,
    app_name: str | None = None,
    max_retries: int = 2,
    wait_time: float = 1.0,
) -> bool:
    """Checks if an app is actively running by querying process PID."""
    pkg = package_name or (
        self.find_package_by_keyword(app_name) if app_name else None
    )
    if not pkg:
        return False

    for attempt in range(max_retries):
        output = self.run_command(f"pidof {pkg}", decode=True)
        if output and str(output).strip():
            return True
        if attempt < max_retries - 1 and wait_time > 0:
            time.sleep(wait_time)

    return False

is_connected()

Checks if ADB connection is currently active and responsive.

Source code in .venv/lib/python3.13/site-packages/pymordialdroid/devices/adb_device.py

def is_connected(self) -> bool:
    """Checks if ADB connection is currently active and responsive."""
    return self._device is not None and self._device.available

open_app(package_name, app_name=None, timeout=10.0, wait_time=1.0)

Launches an app via resolved Activity or Monkey fallback, verifying it started.

Source code in .venv/lib/python3.13/site-packages/pymordialdroid/devices/adb_device.py

def open_app(
    self,
    package_name: str,
    app_name: str | None = None,
    timeout: float = 10.0,
    wait_time: float = 1.0,
) -> bool:
    """Launches an app via resolved Activity or Monkey fallback, verifying it started."""
    pkg = package_name or (
        self.find_package_by_keyword(app_name) if app_name else None
    )
    if not pkg:
        log.error(f"Could not resolve package for app: {app_name}")
        return False

    # 1. Try activity-based launch first (fast and deterministic)
    activity = self.get_launch_activity(pkg)
    if activity:
        self.run_command(f"am start -n {activity}")
        log.debug(f"Launched {pkg} via activity: {activity}")
    else:
        # 2. Fallback to Monkey intent launcher
        self.run_command(f"monkey -p {pkg} -c android.intent.category.LAUNCHER 1")
        log.debug(f"Launched {pkg} via Monkey fallback")

    # Verify app is running within timeout
    start_time = time.time()
    while time.time() - start_time < timeout:
        if self.is_app_running(pkg, max_retries=1, wait_time=0):
            return True
        time.sleep(wait_time)

    log.warning(f"App {pkg} did not start within {timeout}s")
    return False

parse_package_list(output) staticmethod

Extracts package names from pm list packages output.

Only package:-prefixed lines are kept; device warnings (e.g. Samsung multi-user SecurityException preambles) are ignored so they can never match a keyword or be force-stopped.

Source code in .venv/lib/python3.13/site-packages/pymordialdroid/devices/adb_device.py

@staticmethod
def parse_package_list(output: str) -> list[str]:
    """Extracts package names from `pm list packages` output.

    Only ``package:``-prefixed lines are kept; device warnings (e.g.
    Samsung multi-user ``SecurityException`` preambles) are ignored so
    they can never match a keyword or be force-stopped.
    """
    packages = []
    for line in output.splitlines():
        stripped = line.strip()
        if stripped.startswith("package:"):
            name = stripped[len("package:") :].strip()
            if name:
                packages.append(name)
    return packages

press_enter()

Simulates Enter key press.

Source code in .venv/lib/python3.13/site-packages/pymordialdroid/devices/adb_device.py

def press_enter(self) -> None:
    """Simulates Enter key press."""
    self.run_command("input keyevent 66")

press_esc()

Simulates Back / Escape key press.

Source code in .venv/lib/python3.13/site-packages/pymordialdroid/devices/adb_device.py

def press_esc(self) -> None:
    """Simulates Back / Escape key press."""
    self.run_command("input keyevent 4")

rank_package(keyword, packages) staticmethod

Ranks packages for a keyword: exact > component > substring.

1. Case-insensitive exact match.
2. Keyword equals a dot-separated component (settings matches com.android.settings but not com.sec.usbsettings); ties break toward the shortest full name.
3. Plain substring match; ties break toward the shortest full name.

Source code in .venv/lib/python3.13/site-packages/pymordialdroid/devices/adb_device.py

@staticmethod
def rank_package(keyword: str, packages: list[str]) -> str | None:
    """Ranks packages for a keyword: exact > component > substring.

    1. Case-insensitive exact match.
    2. Keyword equals a dot-separated component (``settings`` matches
       ``com.android.settings`` but not ``com.sec.usbsettings``);
       ties break toward the shortest full name.
    3. Plain substring match; ties break toward the shortest full name.
    """
    kw = keyword.lower()
    for pkg in packages:
        if pkg.lower() == kw:
            return pkg
    component = [p for p in packages if kw in p.lower().split(".")]
    if component:
        return min(component, key=len)
    substring = [p for p in packages if kw in p.lower()]
    if substring:
        return min(substring, key=len)
    return None

run_command(command, decode=True)

Executes a shell command on the device.

Source code in .venv/lib/python3.13/site-packages/pymordialdroid/devices/adb_device.py

def run_command(self, command: str, decode: bool = True) -> str | bytes | None:
    """Executes a shell command on the device."""
    if not self.is_connected():
        if not self.connect():
            return None
    try:
        output = self._device.shell(command, decode=decode)
        return output.strip() if decode and isinstance(output, str) else output
    except Exception as e:
        log.error(f"Failed to execute command '{command}': {e}")
        return None

set_frame_provider(provider)

Sets an optional headless-feed frame provider.

The provider is a zero-arg callable returning PNG bytes (or None). Typically ScrcpyDevice.get_latest_frame. When set, :meth:capture_screenshot tries it first for a high-FPS frame and falls back to screencap -p on None/exception.

Source code in .venv/lib/python3.13/site-packages/pymordialdroid/devices/adb_device.py

def set_frame_provider(self, provider: Any | None) -> None:
    """Sets an optional headless-feed frame provider.

    The provider is a zero-arg callable returning PNG bytes (or None).
    Typically ``ScrcpyDevice.get_latest_frame``. When set,
    :meth:`capture_screenshot` tries it first for a high-FPS frame and
    falls back to ``screencap -p`` on None/exception.
    """
    self._frame_provider = provider

show_recent_apps()

Opens recent apps / overview.

Source code in .venv/lib/python3.13/site-packages/pymordialdroid/devices/adb_device.py

def show_recent_apps(self) -> bool:
    """Opens recent apps / overview."""
    res = self.run_command("input keyevent 187")
    return res is not None

shutdown()

Disconnects and cleans up resources.

Source code in .venv/lib/python3.13/site-packages/pymordialdroid/devices/adb_device.py

def shutdown(self) -> None:
    """Disconnects and cleans up resources."""
    self.disconnect()

start_stream()

Starts stream mode (tracks frames from screenshot captures).

Source code in .venv/lib/python3.13/site-packages/pymordialdroid/devices/adb_device.py

def start_stream(self) -> None:
    """Starts stream mode (tracks frames from screenshot captures)."""
    self._is_streaming = True

stop_stream()

Stops stream mode.

Source code in .venv/lib/python3.13/site-packages/pymordialdroid/devices/adb_device.py

def stop_stream(self) -> None:
    """Stops stream mode."""
    self._is_streaming = False

swipe(start_x, start_y, end_x, end_y, duration=300)

Performs a touch swipe gesture from start to end coordinates.

Source code in .venv/lib/python3.13/site-packages/pymordialdroid/devices/adb_device.py

def swipe(
    self,
    start_x: int,
    start_y: int,
    end_x: int,
    end_y: int,
    duration: int = 300,
) -> bool:
    """Performs a touch swipe gesture from start to end coordinates."""
    res = self.run_command(
        f"input swipe {start_x} {start_y} {end_x} {end_y} {duration}"
    )
    return res is not None

tap(coords, y=None, times=1)

Sends tap events to coordinates on screen.

Accepts either a tuple/list (x, y) or separate x, y integers.

Source code in .venv/lib/python3.13/site-packages/pymordialdroid/devices/adb_device.py

def tap(
    self,
    coords: tuple[int, int] | list[int] | int,
    y: int | None = None,
    times: int = 1,
) -> bool:
    """Sends tap events to coordinates on screen.

    Accepts either a tuple/list (x, y) or separate x, y integers.
    """
    if isinstance(coords, (tuple, list)):
        target_x, target_y = coords[0], coords[1]
    elif y is not None:
        target_x, target_y = coords, y
    else:
        log.warning(f"Invalid tap coordinates: coords={coords}, y={y}")
        return False

    for _ in range(times):
        self.run_command(f"input tap {target_x} {target_y}")
        if times > 1:
            time.sleep(0.1)
    return True

type_text(text, enter=False)

Types text into the focused input field, escaping shell characters.

Source code in .venv/lib/python3.13/site-packages/pymordialdroid/devices/adb_device.py

def type_text(self, text: str, enter: bool = False) -> bool:
    """Types text into the focused input field, escaping shell characters."""
    # Escape characters that could break ADB shell input
    escaped_text = text.replace("\\", "\\\\")
    escaped_text = escaped_text.replace(" ", "%s")
    escaped_text = escaped_text.replace("'", "\\'")
    escaped_text = escaped_text.replace('"', '\\"')
    escaped_text = escaped_text.replace("&", "\\&")
    escaped_text = escaped_text.replace("<", "\\<")
    escaped_text = escaped_text.replace(">", "\\>")
    escaped_text = escaped_text.replace(";", "\\;")
    escaped_text = escaped_text.replace("|", "\\|")

    res = self.run_command(f"input text '{escaped_text}'")
    if enter:
        self.press_enter()
    return res is not None

BlueStacks Device

pymordialblue.devices.bluestacks_device

Controller for managing the BlueStacks emulator.

BluestacksDevice

Bases: PymordialEmulatorDevice

Controls the BlueStacks emulator.

Attributes:

Name	Type	Description
running_apps	list[PymordialApp] | list	A list of currently running PymordialApp instances.
state	list[PymordialApp] | list	The state machine managing the BlueStacks lifecycle state.
elements	list[PymordialApp] | list	A container for BlueStacks UI elements.
elements	list[PymordialApp] | list	A container for BlueStacks UI elements.
config		The configuration dictionary for BlueStacks.

Source code in src/pymordialblue/devices/bluestacks_device.py

class BluestacksDevice(PymordialEmulatorDevice):
    """Controls the BlueStacks emulator.

    Attributes:
        running_apps: A list of currently running PymordialApp instances.
        state: The state machine managing the BlueStacks lifecycle state.
        elements: A container for BlueStacks UI elements.
        elements: A container for BlueStacks UI elements.
        config: The configuration dictionary for BlueStacks.
    """

    name: str = "bluestacks"
    version: str = "0.1.0"

    def __init__(
        self,
        adb_bridge_device: AdbDevice | None = None,
        vision_device: AndroidUiDevice | None = None,
        config: BluestacksConfig | None = None,
    ) -> None:
        """Initializes the BluestacksDevice.

        Args:
            adb_bridge_device: The bridge device (e.g. AdbDevice) used for
                low-level ADB interactions.
            vision_device: The vision device used for screen analysis.
            config: A TypedDict containing BlueStacks configuration options.
                Defaults to package defaults if None.
        """
        self.logger = getLogger("BluestacksDevice")
        basicConfig(
            level=DEBUG, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
        )
        self.logger.info("Initializing BluestacksDevice...")
        super().__init__()
        self.config = copy.deepcopy(config or get_config()["bluestacks"])
        self.running_apps: list[PymordialApp] | list = list()

        self._adb_bridge_device: AdbDevice | None = adb_bridge_device
        self._vision_device: AndroidUiDevice | None = vision_device
        self._ref_window_size: tuple[int, int] = tuple(
            self.config["default_resolution"]
        )

        self._filepath: str | None = None
        self._hd_player_exe: str = self.config["hd_player_exe"]

        self.state.register_handler(EmulatorState.LOADING, self.wait_for_load, None)
        self.state.register_handler(EmulatorState.READY, self._connect_adb, None)

        self._autoset_filepath()

        self.logger.debug(
            f"BluestacksDevice initialized with the following state:\n{self.state}\n"
        )

    def _connect_adb(self) -> None:
        """Connects the ADB bridge device if available."""
        if self._adb_bridge_device:
            self._adb_bridge_device.connect()
        else:
            self.logger.warning(
                "ADB bridge device not set, cannot connect during READY transition."
            )

    def initialize(self, config: "BlueConfig") -> None:
        """Initializes the BlueStacks device plugin with configuration.

        Args:
            config: Global Pymordial configuration dictionary.
        """
        pass

    def set_dependencies(
        self,
        adb_bridge_device: AdbDevice,
        vision_device: AndroidUiDevice,
    ) -> None:
        """Sets external dependencies (dependency injection).

        Args:
            adb_bridge_device: The ADB bridge device.
            vision_device: The vision device.
        """
        self._adb_bridge_device = adb_bridge_device
        self._vision_device = vision_device

    def shutdown(self) -> None:
        """Kills the emulator process."""
        self.close()

    @property
    def ref_window_size(self) -> tuple[int, int] | None:
        """Gets the reference window size.

        Returns:
            A tuple containing (width, height) in pixels, or None if not set.
        """
        return self._ref_window_size

    @ref_window_size.setter
    @log_property_setter
    def ref_window_size(self, width_height: tuple[int | str, int | str]) -> None:
        """Sets the reference window size.

        Args:
            width_height: A tuple containing (width, height) in pixels. Values
                can be integers or string representations of integers.

        Raises:
            ValueError: If the provided width or height are not integers (or
                strings representing integers), or if they are not positive.
        """
        width = width_height[0]
        height = width_height[1]
        if not isinstance(width, int):
            if isinstance(width, str) and width.isdigit():
                width: int = int(width)
                if width <= 0:
                    self.logger.warning(
                        "ValueError while trying to set BluestacksDevice 'ref_window_size': Provided width must be positive integers!"
                    )
                    raise ValueError("Provided width must be positive integers")
            else:
                self.logger.warning(
                    "ValueError while trying to set BluestacksDevice 'ref_window_size': Provided width must be an integer or the string representation of an integer!"
                )
                raise ValueError(
                    "Provided width must be integer or the string representation of an integer!"
                )

        if not isinstance(height, int):
            if isinstance(height, str) and height.isdigit():
                height: int = int(height)
                if height <= 0:
                    self.logger.warning(
                        "ValueError while trying to set BluestacksDevice 'ref_window_size': Provided height must be positive integers!"
                    )
                    raise ValueError("Provided height must be positive integers")
            else:
                self.logger.warning(
                    "ValueError while trying to set BluestacksDevice 'ref_window_size': Provided height must be an integer or the string representation of an integer!"
                )
                raise ValueError(
                    "Provided height must be integer or the string representation of an integer!"
                )

        self._ref_window_size = (width, height)

    @property
    def filepath(self) -> str | None:
        """Gets the BlueStacks executable filepath.

        Returns:
            The absolute path to the HD-Player.exe file as a string, or None
            if it has not been determined.
        """
        return self._filepath

    @filepath.setter
    @log_property_setter
    def filepath(self, filepath: str) -> None:
        """Sets the BlueStacks executable filepath.

        Args:
            filepath: The absolute path to HD-Player.exe.

        Raises:
            ValueError: If the provided filepath is not a string or if the path
                does not exist on the filesystem.
        """
        if not isinstance(filepath, str):
            self.logger.warning(
                "ValueError while trying to set BluestacksDevice 'filepath': Provided filepath must be a string!"
            )
            raise ValueError("Provided filepath must be a string")

        if not os.path.exists(filepath):
            self.logger.warning(
                "ValueError while trying to set BluestacksDevice 'filepath': Provided filepath does not exist!"
            )
            raise ValueError("Provided filepath does not exist")

        self._filepath: str = filepath

    def open(
        self,
        max_retries: int | None = None,
        wait_time: int | None = None,
        timeout_s: int | None = None,
    ) -> None:
        """Opens the BlueStacks emulator application.

        Args:
            max_retries: The maximum number of attempts to detect the process
                after launching. Defaults to the configuration value.
            wait_time: The time in seconds to wait between detection attempts.
                Defaults to the configuration value.
            timeout_s: The maximum total time in seconds to wait for the process
                to appear before timing out. Defaults to the configuration value.

        Raises:
            ValueError: If BlueStacks fails to start due to an OS error.
            Exception: If the BlueStacks process window is not found after the
                specified retries or timeout period.
        """
        max_retries: int = validate_and_convert_int(
            max_retries or self.config["default_open_app_max_retries"], "max_retries"
        )
        wait_time: int = validate_and_convert_int(
            wait_time or self.config["default_open_app_wait_time"], "wait_time"
        )
        timeout_s: int = validate_and_convert_int(
            timeout_s or self.config["default_open_app_timeout"], "timeout_s"
        )
        match self.state.current_state:
            case EmulatorState.CLOSED:
                self.logger.info("Opening Bluestacks controller...")
                if not self._filepath:
                    self._autoset_filepath()
                try:
                    os.startfile(self._filepath)
                except Exception as e:
                    self.logger.error(f"Failed to start Bluestacks: {e}")
                    raise ValueError(f"Failed to start Bluestacks: {e}")

                start_time: float = time.time()

                for attempt in range(max_retries):
                    is_open: bool = any(
                        p.name().lower() == self._hd_player_exe.lower()
                        for p in psutil.process_iter(["name"])
                    )
                    if is_open:
                        self.logger.info("Bluestacks controller opened successfully.")
                        # Transition to LOADING - state handler will automatically call wait_for_load()
                        self.state.transition_to(EmulatorState.LOADING)
                        return

                    if time.time() - start_time > timeout_s:
                        self.logger.error(
                            "Timeout waiting for Bluestacks window to appear"
                        )
                        raise Exception(
                            "Timeout waiting for Bluestacks window to appear"
                        )

                    self.logger.warning(
                        f"Attempt {attempt + 1}/{max_retries}: Could not find Bluestacks window."
                    )
                    time.sleep(wait_time)

                self.logger.error(
                    f"Failed to find Bluestacks window after all attempts {attempt + 1}/{max_retries}"
                )
                raise Exception(
                    f"Failed to find Bluestacks window after all attempts {attempt + 1}/{max_retries}"
                )
            case EmulatorState.LOADING:
                self.logger.info(
                    "Bluestacks controller is already open and currently loading."
                )
                return
            case EmulatorState.READY:
                self.logger.info("Bluestacks controller is already open and ready.")
                return

    def open_settings(self) -> bool:
        """Opens the Settings app using a verified activity name.

        Returns:
            True if opened successfully, False otherwise.
        """
        # Based on manual verification:
        # package: com.bluestacks.settings
        # activity: .SettingsActivity
        self.logger.info("Opening Settings...")
        return self._adb_bridge_device.open_app(
            package_name="com.bluestacks.settings",
            app_name="settings",
        )

    def wait_for_load(self, timeout_s: int | None = None) -> None:
        """Waits for Bluestacks to finish loading by polling the ADB connection.

        This method blocks until the emulator is responsive via ADB or the
        timeout is reached.

        Args:
            timeout_s: The maximum number of seconds to wait for the emulator
                to load. Defaults to the configuration value.
        """
        self.logger.debug("Waiting for Bluestacks to load (ADB check)...")
        start_time = time.time()
        timeout_s = timeout_s or self.config["default_load_timeout"]

        while self.state.current_state == EmulatorState.LOADING:
            # Try to connect to ADB
            if self._adb_bridge_device.connect():
                self._adb_bridge_device.disconnect()
                default_ui_load_wait_time: int = self.config[
                    "default_ui_load_wait_time"
                ]
                self.logger.debug(
                    f"Waiting {default_ui_load_wait_time} seconds for UI to stabilize..."
                )
                time.sleep(default_ui_load_wait_time)
                self._adb_bridge_device.connect()
                self.logger.info("Bluestacks is loaded & ready.")
                self.state.transition_to(EmulatorState.READY)
                return

            # Check timeout
            if time.time() - start_time > timeout_s:
                self.logger.error(
                    f"Timeout waiting for Bluestacks to load after {timeout_s} seconds."
                )
                # We transition to READY anyway to allow retry logic elsewhere if needed,
                # or maybe we should raise? For now, mimicking previous behavior.
                self.state.transition_to(EmulatorState.READY)
                return

            time.sleep(self.config["default_load_wait_time"])

    def is_ready(self) -> bool:
        """Checks if BlueStacks is in the READY state.

        Returns:
            True if the current state is EmulatorState.READY, False otherwise.
        """
        return self.state.current_state == EmulatorState.READY

    def close(self) -> bool:
        """Kills the Bluestacks controller process.

        This will also disconnect the ADB bridge device.

        Returns:
            True if the Bluestacks process was found and killed, or if no
            process was found running. False if the process was found but
            could not be killed.

        Raises:
            ValueError: If an unexpected error occurs during the process
                killing routine.
        """
        self.logger.info("Killing Bluestacks controller...")

        try:
            process_found = False
            for proc in psutil.process_iter(["pid", "name"]):
                if proc.info["name"] == self._hd_player_exe:
                    process_found = True
                    try:
                        self._adb_bridge_device.disconnect()
                    except Exception as e:
                        self.logger.warning(
                            f"Error in close method while trying to disconnect adb bridge: {e}\nContinuing to close the Bluestacks process..."
                        )
                    try:
                        proc.kill()
                        proc.wait(timeout=self.config["default_process_kill_timeout"])
                    except (
                        psutil.NoSuchProcess,
                        psutil.AccessDenied,
                        psutil.ZombieProcess,
                    ):
                        return False

            if not process_found:
                self.logger.debug("Bluestacks process was not found.")
                return False

            if self.state.current_state != EmulatorState.CLOSED:
                self.state.transition_to(EmulatorState.CLOSED)

            self.logger.info("Bluestacks process killed.")
            return True

        except Exception as e:
            self.logger.error(f"Error in close: {e}")
            raise ValueError(f"Failed to kill Bluestacks: {e}")

    def _autoset_filepath(self) -> None:
        """Automatically detects and sets the BlueStacks executable path.

        This method attempts to locate `HD-Player.exe` by searching:
        1. Standard "Program Files" locations.
        2. Common custom installation paths.
        3. The current working directory.
        4. A broad walk of the C: drive (if initial checks fail).

        Raises:
            FileNotFoundError: If `HD-Player.exe` cannot be located automatically
                in any of the searched locations.
        """
        self.logger.debug("Setting filepath...")

        # Common installation paths for BlueStacks
        search_paths = [
            # Standard Program Files locations
            os.path.join(
                os.environ.get("ProgramFiles", ""),
                "BlueStacks_nxt",
                self._hd_player_exe,
            ),
            os.path.join(
                os.environ.get("ProgramFiles(x86)", ""),
                "BlueStacks_nxt",
                self._hd_player_exe,
            ),
            # Alternative BlueStacks versions
            os.path.join(
                os.environ.get("ProgramFiles", ""), "BlueStacks", self._hd_player_exe
            ),
            os.path.join(
                os.environ.get("ProgramFiles(x86)", ""),
                "BlueStacks",
                self._hd_player_exe,
            ),
            # Common custom installation paths
            f"C:\\Program Files\\BlueStacks_nxt\\{self._hd_player_exe}",
            f"C:\\Program Files (x86)\\BlueStacks_nxt\\{self._hd_player_exe}",
            f"C:\\BlueStacks\\{self._hd_player_exe}",
            f"C:\\BlueStacks_nxt\\{self._hd_player_exe}",
            # Check if file exists in current directory or subdirectories
            self._hd_player_exe,
        ]

        # Remove empty paths from environment variables
        search_paths = [
            path for path in search_paths if path and path != self._hd_player_exe
        ]

        # Add current working directory relative paths
        cwd = os.getcwd()
        search_paths.extend(
            [
                os.path.join(cwd, "BlueStacks_nxt", self._hd_player_exe),
                os.path.join(cwd, "BlueStacks", self._hd_player_exe),
            ]
        )

        self.logger.debug(
            f"Searching for HD-Player.exe in {len(search_paths)} locations"
        )

        for potential_path in search_paths:
            if os.path.exists(potential_path) and os.path.isfile(potential_path):
                self._filepath = potential_path
                self.logger.debug(f"HD-Player.exe filepath set to {self._filepath}.")
                return
            else:
                self.logger.debug(f"Checked path (does not exist): {potential_path}")

        # If we still haven't found it, try a broader search
        self.logger.debug("Performing broader search for HD-Player.exe...")
        try:
            for root, dirs, files in os.walk("C:\\"):
                if self._hd_player_exe in files:
                    potential_path = os.path.join(root, self._hd_player_exe)
                    if "bluestacks" in root.lower():
                        self._filepath = potential_path
                        self.logger.debug(
                            f"HD-Player.exe found via broad search: {self._filepath}"
                        )
                        return
        except Exception as e:
            self.logger.debug(f"Broad search failed: {e}")

        self.logger.error(
            "Could not find HD-Player.exe. Please ensure BlueStacks is installed or manually specify the filepath."
        )
        self.logger.error(f"Searched paths: {search_paths}")
        self.logger.error(f"Current working directory: {os.getcwd()}")
        self.logger.error(f"ProgramFiles: {os.environ.get('ProgramFiles')}")
        self.logger.error(f"ProgramFiles(x86): {os.environ.get('ProgramFiles(x86)')}")
        raise FileNotFoundError(
            "Could not find HD-Player.exe. Please ensure BlueStacks is installed or manually specify the filepath."
        )

filepath property writable

Gets the BlueStacks executable filepath.

Returns:

Type	Description
str | None	The absolute path to the HD-Player.exe file as a string, or None
str | None	if it has not been determined.

ref_window_size property writable

Gets the reference window size.

Returns:

Type	Description
tuple[int, int] | None	A tuple containing (width, height) in pixels, or None if not set.

__init__(adb_bridge_device=None, vision_device=None, config=None)

Initializes the BluestacksDevice.

Parameters:

Name	Type	Description	Default
adb_bridge_device	AdbDevice | None	The bridge device (e.g. AdbDevice) used for low-level ADB interactions.	None
vision_device	AndroidUiDevice | None	The vision device used for screen analysis.	None
config	BluestacksConfig | None	A TypedDict containing BlueStacks configuration options. Defaults to package defaults if None.	None

Source code in src/pymordialblue/devices/bluestacks_device.py

def __init__(
    self,
    adb_bridge_device: AdbDevice | None = None,
    vision_device: AndroidUiDevice | None = None,
    config: BluestacksConfig | None = None,
) -> None:
    """Initializes the BluestacksDevice.

    Args:
        adb_bridge_device: The bridge device (e.g. AdbDevice) used for
            low-level ADB interactions.
        vision_device: The vision device used for screen analysis.
        config: A TypedDict containing BlueStacks configuration options.
            Defaults to package defaults if None.
    """
    self.logger = getLogger("BluestacksDevice")
    basicConfig(
        level=DEBUG, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
    )
    self.logger.info("Initializing BluestacksDevice...")
    super().__init__()
    self.config = copy.deepcopy(config or get_config()["bluestacks"])
    self.running_apps: list[PymordialApp] | list = list()

    self._adb_bridge_device: AdbDevice | None = adb_bridge_device
    self._vision_device: AndroidUiDevice | None = vision_device
    self._ref_window_size: tuple[int, int] = tuple(
        self.config["default_resolution"]
    )

    self._filepath: str | None = None
    self._hd_player_exe: str = self.config["hd_player_exe"]

    self.state.register_handler(EmulatorState.LOADING, self.wait_for_load, None)
    self.state.register_handler(EmulatorState.READY, self._connect_adb, None)

    self._autoset_filepath()

    self.logger.debug(
        f"BluestacksDevice initialized with the following state:\n{self.state}\n"
    )

close()

Kills the Bluestacks controller process.

This will also disconnect the ADB bridge device.

Returns:

Type	Description
bool	True if the Bluestacks process was found and killed, or if no
bool	process was found running. False if the process was found but
bool	could not be killed.

Raises:

Type	Description
ValueError	If an unexpected error occurs during the process killing routine.

Source code in src/pymordialblue/devices/bluestacks_device.py

def close(self) -> bool:
    """Kills the Bluestacks controller process.

    This will also disconnect the ADB bridge device.

    Returns:
        True if the Bluestacks process was found and killed, or if no
        process was found running. False if the process was found but
        could not be killed.

    Raises:
        ValueError: If an unexpected error occurs during the process
            killing routine.
    """
    self.logger.info("Killing Bluestacks controller...")

    try:
        process_found = False
        for proc in psutil.process_iter(["pid", "name"]):
            if proc.info["name"] == self._hd_player_exe:
                process_found = True
                try:
                    self._adb_bridge_device.disconnect()
                except Exception as e:
                    self.logger.warning(
                        f"Error in close method while trying to disconnect adb bridge: {e}\nContinuing to close the Bluestacks process..."
                    )
                try:
                    proc.kill()
                    proc.wait(timeout=self.config["default_process_kill_timeout"])
                except (
                    psutil.NoSuchProcess,
                    psutil.AccessDenied,
                    psutil.ZombieProcess,
                ):
                    return False

        if not process_found:
            self.logger.debug("Bluestacks process was not found.")
            return False

        if self.state.current_state != EmulatorState.CLOSED:
            self.state.transition_to(EmulatorState.CLOSED)

        self.logger.info("Bluestacks process killed.")
        return True

    except Exception as e:
        self.logger.error(f"Error in close: {e}")
        raise ValueError(f"Failed to kill Bluestacks: {e}")

initialize(config)

Initializes the BlueStacks device plugin with configuration.

Parameters:

Name	Type	Description	Default
config	BlueConfig	Global Pymordial configuration dictionary.	required

Source code in src/pymordialblue/devices/bluestacks_device.py

def initialize(self, config: "BlueConfig") -> None:
    """Initializes the BlueStacks device plugin with configuration.

    Args:
        config: Global Pymordial configuration dictionary.
    """
    pass

is_ready()

Checks if BlueStacks is in the READY state.

Returns:

Type	Description
bool	True if the current state is EmulatorState.READY, False otherwise.

Source code in src/pymordialblue/devices/bluestacks_device.py

def is_ready(self) -> bool:
    """Checks if BlueStacks is in the READY state.

    Returns:
        True if the current state is EmulatorState.READY, False otherwise.
    """
    return self.state.current_state == EmulatorState.READY

open(max_retries=None, wait_time=None, timeout_s=None)

Opens the BlueStacks emulator application.

Parameters:

Name	Type	Description	Default
max_retries	int | None	The maximum number of attempts to detect the process after launching. Defaults to the configuration value.	None
wait_time	int | None	The time in seconds to wait between detection attempts. Defaults to the configuration value.	None
timeout_s	int | None	The maximum total time in seconds to wait for the process to appear before timing out. Defaults to the configuration value.	None

Raises:

Type	Description
ValueError	If BlueStacks fails to start due to an OS error.
Exception	If the BlueStacks process window is not found after the specified retries or timeout period.

Source code in src/pymordialblue/devices/bluestacks_device.py

def open(
    self,
    max_retries: int | None = None,
    wait_time: int | None = None,
    timeout_s: int | None = None,
) -> None:
    """Opens the BlueStacks emulator application.

    Args:
        max_retries: The maximum number of attempts to detect the process
            after launching. Defaults to the configuration value.
        wait_time: The time in seconds to wait between detection attempts.
            Defaults to the configuration value.
        timeout_s: The maximum total time in seconds to wait for the process
            to appear before timing out. Defaults to the configuration value.

    Raises:
        ValueError: If BlueStacks fails to start due to an OS error.
        Exception: If the BlueStacks process window is not found after the
            specified retries or timeout period.
    """
    max_retries: int = validate_and_convert_int(
        max_retries or self.config["default_open_app_max_retries"], "max_retries"
    )
    wait_time: int = validate_and_convert_int(
        wait_time or self.config["default_open_app_wait_time"], "wait_time"
    )
    timeout_s: int = validate_and_convert_int(
        timeout_s or self.config["default_open_app_timeout"], "timeout_s"
    )
    match self.state.current_state:
        case EmulatorState.CLOSED:
            self.logger.info("Opening Bluestacks controller...")
            if not self._filepath:
                self._autoset_filepath()
            try:
                os.startfile(self._filepath)
            except Exception as e:
                self.logger.error(f"Failed to start Bluestacks: {e}")
                raise ValueError(f"Failed to start Bluestacks: {e}")

            start_time: float = time.time()

            for attempt in range(max_retries):
                is_open: bool = any(
                    p.name().lower() == self._hd_player_exe.lower()
                    for p in psutil.process_iter(["name"])
                )
                if is_open:
                    self.logger.info("Bluestacks controller opened successfully.")
                    # Transition to LOADING - state handler will automatically call wait_for_load()
                    self.state.transition_to(EmulatorState.LOADING)
                    return

                if time.time() - start_time > timeout_s:
                    self.logger.error(
                        "Timeout waiting for Bluestacks window to appear"
                    )
                    raise Exception(
                        "Timeout waiting for Bluestacks window to appear"
                    )

                self.logger.warning(
                    f"Attempt {attempt + 1}/{max_retries}: Could not find Bluestacks window."
                )
                time.sleep(wait_time)

            self.logger.error(
                f"Failed to find Bluestacks window after all attempts {attempt + 1}/{max_retries}"
            )
            raise Exception(
                f"Failed to find Bluestacks window after all attempts {attempt + 1}/{max_retries}"
            )
        case EmulatorState.LOADING:
            self.logger.info(
                "Bluestacks controller is already open and currently loading."
            )
            return
        case EmulatorState.READY:
            self.logger.info("Bluestacks controller is already open and ready.")
            return

open_settings()

Opens the Settings app using a verified activity name.

Returns:

Type	Description
bool	True if opened successfully, False otherwise.

Source code in src/pymordialblue/devices/bluestacks_device.py

def open_settings(self) -> bool:
    """Opens the Settings app using a verified activity name.

    Returns:
        True if opened successfully, False otherwise.
    """
    # Based on manual verification:
    # package: com.bluestacks.settings
    # activity: .SettingsActivity
    self.logger.info("Opening Settings...")
    return self._adb_bridge_device.open_app(
        package_name="com.bluestacks.settings",
        app_name="settings",
    )

set_dependencies(adb_bridge_device, vision_device)

Sets external dependencies (dependency injection).

Parameters:

Name	Type	Description	Default
adb_bridge_device	AdbDevice	The ADB bridge device.	required
vision_device	AndroidUiDevice	The vision device.	required

Source code in src/pymordialblue/devices/bluestacks_device.py

def set_dependencies(
    self,
    adb_bridge_device: AdbDevice,
    vision_device: AndroidUiDevice,
) -> None:
    """Sets external dependencies (dependency injection).

    Args:
        adb_bridge_device: The ADB bridge device.
        vision_device: The vision device.
    """
    self._adb_bridge_device = adb_bridge_device
    self._vision_device = vision_device

shutdown()

Kills the emulator process.

Source code in src/pymordialblue/devices/bluestacks_device.py

def shutdown(self) -> None:
    """Kills the emulator process."""
    self.close()

wait_for_load(timeout_s=None)

Waits for Bluestacks to finish loading by polling the ADB connection.

This method blocks until the emulator is responsive via ADB or the timeout is reached.

Parameters:

Name	Type	Description	Default
timeout_s	int | None	The maximum number of seconds to wait for the emulator to load. Defaults to the configuration value.	None

Source code in src/pymordialblue/devices/bluestacks_device.py

def wait_for_load(self, timeout_s: int | None = None) -> None:
    """Waits for Bluestacks to finish loading by polling the ADB connection.

    This method blocks until the emulator is responsive via ADB or the
    timeout is reached.

    Args:
        timeout_s: The maximum number of seconds to wait for the emulator
            to load. Defaults to the configuration value.
    """
    self.logger.debug("Waiting for Bluestacks to load (ADB check)...")
    start_time = time.time()
    timeout_s = timeout_s or self.config["default_load_timeout"]

    while self.state.current_state == EmulatorState.LOADING:
        # Try to connect to ADB
        if self._adb_bridge_device.connect():
            self._adb_bridge_device.disconnect()
            default_ui_load_wait_time: int = self.config[
                "default_ui_load_wait_time"
            ]
            self.logger.debug(
                f"Waiting {default_ui_load_wait_time} seconds for UI to stabilize..."
            )
            time.sleep(default_ui_load_wait_time)
            self._adb_bridge_device.connect()
            self.logger.info("Bluestacks is loaded & ready.")
            self.state.transition_to(EmulatorState.READY)
            return

        # Check timeout
        if time.time() - start_time > timeout_s:
            self.logger.error(
                f"Timeout waiting for Bluestacks to load after {timeout_s} seconds."
            )
            # We transition to READY anyway to allow retry logic elsewhere if needed,
            # or maybe we should raise? For now, mimicking previous behavior.
            self.state.transition_to(EmulatorState.READY)
            return

        time.sleep(self.config["default_load_wait_time"])

UI Device

pymordialdroid.devices.ui_device

Vision and UI element detection implementing Pymordial's PymordialVisionDevice contract.

AndroidUiDevice

Bases: PymordialVisionDevice

Handles visual recognition tasks: template matching, pixel checks, and OCR.

Source code in .venv/lib/python3.13/site-packages/pymordialdroid/devices/ui_device.py

class AndroidUiDevice(PymordialVisionDevice):
    """Handles visual recognition tasks: template matching, pixel checks, and OCR."""

    name: str = "ui"
    version: str = "0.1.0"

    def __init__(
        self,
        bridge_device: PymordialBridgeDevice | None = None,
        ocr_device: PymordialOCRDevice | None = None,
    ) -> None:
        self.bridge_device = bridge_device
        self._ocr_device: PymordialOCRDevice = ocr_device or TesseractDevice()

    def initialize(self, config: Any = None) -> None:
        """Initializes the UI vision plugin."""
        pass

    def shutdown(self) -> None:
        """Performs cleanup."""
        pass

    def set_bridge_device(self, bridge_device: PymordialBridgeDevice) -> None:
        """Sets the underlying bridge device."""
        self.bridge_device = bridge_device

    def set_ocr_device(self, ocr_device: PymordialOCRDevice) -> None:
        """Sets the OCR device."""
        self._ocr_device = ocr_device

    def _ensure_screenshot(
        self, screenshot: bytes | np.ndarray | Image.Image | str | Path | None
    ) -> bytes | np.ndarray | None:
        """Returns the screenshot, capturing a fresh one via bridge_device if not provided."""
        if screenshot is not None:
            if isinstance(screenshot, (Path, str)):
                return cv2.imread(str(screenshot))
            if isinstance(screenshot, Image.Image):
                return np.array(screenshot)
            return screenshot

        if self.bridge_device:
            return self.bridge_device.capture_screenshot()

        return None

    def scale_img_to_screen(
        self,
        image_path: str | Path,
        screen_image: str | Image.Image | bytes | np.ndarray,
        ref_resolution: tuple[int, int] | None = None,
    ) -> Image.Image:
        """Scales a template image to match the device screen resolution.

        Args:
            image_path: Path to the reference image.
            screen_image: The current screen image.
            ref_resolution: Original resolution (width, height) the template was captured at.

        Returns:
            Scaled PIL Image.
        """
        if isinstance(screen_image, (bytes, bytearray)):
            screen_img = Image.open(BytesIO(screen_image))
        elif isinstance(screen_image, np.ndarray):
            screen_img = Image.fromarray(cv2.cvtColor(screen_image, cv2.COLOR_BGR2RGB))
        elif isinstance(screen_image, (str, Path)):
            screen_img = Image.open(str(screen_image))
        elif isinstance(screen_image, Image.Image):
            screen_img = screen_image
        else:
            raise ValueError(f"Unsupported screen image type: {type(screen_image)}")

        needle_img = Image.open(str(image_path))
        if not ref_resolution:
            return needle_img

        screen_w, screen_h = screen_img.size
        ref_w, ref_h = ref_resolution

        if ref_w <= 0 or ref_h <= 0:
            return needle_img

        ratio_w = screen_w / ref_w
        ratio_h = screen_h / ref_h

        scaled_size = (
            max(1, int(needle_img.size[0] * ratio_w)),
            max(1, int(needle_img.size[1] * ratio_h)),
        )
        return needle_img.resize(scaled_size, Image.Resampling.BICUBIC)

    def check_pixel_color(
        self,
        pymordial_pixel: PymordialPixel | None = None,
        pymordial_screenshot: bytes | np.ndarray | None = None,
        coords: tuple[int, int] | None = None,
        expected_color: tuple[int, int, int] | None = None,
        tolerance: int = 10,
        screenshot: bytes | np.ndarray | None = None,
    ) -> bool | None:
        """Checks if a pixel matches the target color within tolerance.

        Supports coordinate scaling when og_resolution is set on pymordial_pixel.
        """
        raw_screen = (
            pymordial_screenshot if pymordial_screenshot is not None else screenshot
        )
        screenshot_data = self._ensure_screenshot(raw_screen)
        if screenshot_data is None:
            return None

        # Resolve coordinates, expected color, and tolerance
        if pymordial_pixel:
            pos = getattr(pymordial_pixel, "position", None)
            if pos is None:
                return None
            target_coords = (int(pos[0]), int(pos[1]))
            exp_rgb = getattr(
                pymordial_pixel,
                "pixel_color",
                getattr(pymordial_pixel, "expected_color", None),
            )
            tol = getattr(pymordial_pixel, "tolerance", tolerance)
            og_res = getattr(pymordial_pixel, "og_resolution", None)
        elif coords and expected_color:
            target_coords = coords
            exp_rgb = expected_color
            tol = tolerance
            og_res = None
        else:
            return None

        if exp_rgb is None:
            return None

        # Convert screenshot to PIL for uniform coordinate and color lookup
        if isinstance(screenshot_data, (bytes, bytearray)):
            img = Image.open(BytesIO(screenshot_data))
        elif isinstance(screenshot_data, np.ndarray):
            img = Image.fromarray(screenshot_data)
        else:
            img = screenshot_data

        actual_w, actual_h = img.size

        # Scale coordinates if original resolution is specified
        if og_res:
            scale_x = actual_w / og_res[0]
            scale_y = actual_h / og_res[1]
            target_coords = (
                int(target_coords[0] * scale_x),
                int(target_coords[1] * scale_y),
            )

        x, y = target_coords
        if not (0 <= x < actual_w and 0 <= y < actual_h):
            return False

        pixel_color = img.getpixel((x, y))
        # Strip alpha channel if present
        actual_color = pixel_color[:3]
        target_rgb = exp_rgb[:3]

        return all(abs(a - t) <= tol for a, t in zip(actual_color, target_rgb))

    def where_element(
        self,
        element: PymordialElement,
        screenshot: bytes | np.ndarray | None = None,
        max_tries: int = 1,
        set_position: bool = False,
        set_size: bool = False,
        wait_time: float = 0.5,
        pymordial_screenshot: bytes | np.ndarray | None = None,
    ) -> tuple[int, int] | None:
        """Finds (center_x, center_y) coordinates of a UI element on screen."""
        target_screen = screenshot if screenshot is not None else pymordial_screenshot
        if isinstance(element, PymordialPixel):
            if self.check_pixel_color(
                pymordial_pixel=element, pymordial_screenshot=target_screen
            ):
                return (int(element.position[0]), int(element.position[1]))
            return None

        if isinstance(element, PymordialText):
            text = getattr(element, "element_text", getattr(element, "text", ""))
            strat = getattr(element, "extract_strategy", None)
            return self.find_text(text, pymordial_screenshot=screenshot, strategy=strat)

        if not isinstance(element, PymordialImage):
            # Bounding box center fallback
            return getattr(element, "center", None)

        # Handle PymordialImage
        raw_path = getattr(element, "filepath", getattr(element, "source_path", ""))
        template_path = Path(raw_path)
        if not template_path.exists():
            log.warning(f"Template image not found: {template_path}")
            return None

        og_res = getattr(element, "og_resolution", None)
        confidence = getattr(element, "confidence", 0.8)

        for attempt in range(max_tries):
            screen_data = self._ensure_screenshot(target_screen)
            if screen_data is None:
                if attempt < max_tries - 1:
                    time.sleep(wait_time)
                    continue
                return None

            try:
                # Convert screen data to PIL Image
                if isinstance(screen_data, (bytes, bytearray)):
                    haystack_pil = Image.open(BytesIO(screen_data))
                elif isinstance(screen_data, np.ndarray):
                    haystack_pil = Image.fromarray(screen_data)
                elif isinstance(screen_data, Image.Image):
                    haystack_pil = screen_data
                else:
                    haystack_pil = Image.open(str(screen_data))

                # Scale template needle to match current screen resolution
                scaled_needle_pil = self.scale_img_to_screen(
                    image_path=template_path,
                    screen_image=haystack_pil,
                    ref_resolution=og_res,
                )

                # Prepare OpenCV BGR images
                haystack_cv = cv2.cvtColor(np.array(haystack_pil), cv2.COLOR_RGB2BGR)
                needle_cv = cv2.cvtColor(np.array(scaled_needle_pil), cv2.COLOR_RGB2BGR)

                region = getattr(element, "region", None)
                if region:
                    rx, ry, rw, rh = region
                    haystack_cv = haystack_cv[ry : ry + rh, rx : rx + rw]
                    offset_x, offset_y = rx, ry
                else:
                    offset_x, offset_y = 0, 0

                # Template matching
                result = cv2.matchTemplate(haystack_cv, needle_cv, cv2.TM_CCOEFF_NORMED)
                _, max_val, _, max_loc = cv2.minMaxLoc(result)

                if max_val >= confidence:
                    match_x = max_loc[0] + offset_x
                    match_y = max_loc[1] + offset_y
                    needle_w, needle_h = scaled_needle_pil.size

                    center_coords = (
                        match_x + needle_w // 2,
                        match_y + needle_h // 2,
                    )

                    if set_position:
                        element.position = (match_x, match_y)
                    if set_size:
                        element.size = (needle_w, needle_h)

                    return center_coords

            except Exception as e:
                log.error(f"Error finding element {element.label}: {e}")

            if attempt < max_tries - 1:
                time.sleep(wait_time)
                screenshot = None  # Force fresh screenshot capture

        return None

    def where_elements(
        self,
        elements: list[PymordialElement],
        screenshot: bytes | np.ndarray | None = None,
        max_tries: int = 1,
        pymordial_screenshot: bytes | np.ndarray | None = None,
    ) -> tuple[int, int] | None:
        """Finds the coordinates of the first matching element from a list."""
        target_screen = screenshot if screenshot is not None else pymordial_screenshot
        for element in elements:
            coords = self.where_element(
                element, screenshot=target_screen, max_tries=max_tries
            )
            if coords is not None:
                return coords
        return None

    def find_text(
        self,
        text_to_find: str,
        pymordial_screenshot: Path | bytes | str | np.ndarray | None = None,
        strategy: PymordialExtractStrategy | None = None,
    ) -> tuple[int, int] | None:
        """Finds the center coordinates of specified text using the OCR device."""
        screenshot = self._ensure_screenshot(pymordial_screenshot)
        if screenshot is None:
            return None

        if hasattr(self._ocr_device, "find_text"):
            return self._ocr_device.find_text(
                text_to_find, screenshot, strategy=strategy
            )
        return None

    def check_text(
        self,
        text_to_find: str,
        pymordial_screenshot: Path | bytes | str | np.ndarray | None = None,
        case_sensitive: bool = False,
        strategy: PymordialExtractStrategy | None = None,
    ) -> bool:
        """Checks if text is visible on screen using the OCR device."""
        screenshot = self._ensure_screenshot(pymordial_screenshot)
        if screenshot is None:
            return False

        try:
            extracted = self._ocr_device.extract_text(screenshot, strategy=strategy)
            if case_sensitive:
                return text_to_find in extracted
            return text_to_find.lower() in extracted.lower()
        except Exception as e:
            log.error(f"Error checking text: {e}")
            return False

    def read_text(
        self,
        pymordial_screenshot: Path | bytes | str | np.ndarray | None = None,
        case_sensitive: bool = False,
        strategy: PymordialExtractStrategy | None = None,
    ) -> list[str]:
        """Reads text lines from the screen using the OCR device.

        Lines are always returned verbatim as recognized; ``case_sensitive``
        is accepted for signature compatibility but no longer mutates the
        output (case-insensitive *matching* belongs in :meth:`check_text`).
        """
        screenshot = self._ensure_screenshot(pymordial_screenshot)
        if screenshot is None:
            return []

        try:
            text = self._ocr_device.extract_text(screenshot, strategy=strategy)
            return [line.strip() for line in text.split("\n") if line.strip()]
        except Exception as e:
            log.error(f"Error reading text: {e}")
            return []

check_pixel_color(pymordial_pixel=None, pymordial_screenshot=None, coords=None, expected_color=None, tolerance=10, screenshot=None)

Checks if a pixel matches the target color within tolerance.

Supports coordinate scaling when og_resolution is set on pymordial_pixel.

Source code in .venv/lib/python3.13/site-packages/pymordialdroid/devices/ui_device.py

def check_pixel_color(
    self,
    pymordial_pixel: PymordialPixel | None = None,
    pymordial_screenshot: bytes | np.ndarray | None = None,
    coords: tuple[int, int] | None = None,
    expected_color: tuple[int, int, int] | None = None,
    tolerance: int = 10,
    screenshot: bytes | np.ndarray | None = None,
) -> bool | None:
    """Checks if a pixel matches the target color within tolerance.

    Supports coordinate scaling when og_resolution is set on pymordial_pixel.
    """
    raw_screen = (
        pymordial_screenshot if pymordial_screenshot is not None else screenshot
    )
    screenshot_data = self._ensure_screenshot(raw_screen)
    if screenshot_data is None:
        return None

    # Resolve coordinates, expected color, and tolerance
    if pymordial_pixel:
        pos = getattr(pymordial_pixel, "position", None)
        if pos is None:
            return None
        target_coords = (int(pos[0]), int(pos[1]))
        exp_rgb = getattr(
            pymordial_pixel,
            "pixel_color",
            getattr(pymordial_pixel, "expected_color", None),
        )
        tol = getattr(pymordial_pixel, "tolerance", tolerance)
        og_res = getattr(pymordial_pixel, "og_resolution", None)
    elif coords and expected_color:
        target_coords = coords
        exp_rgb = expected_color
        tol = tolerance
        og_res = None
    else:
        return None

    if exp_rgb is None:
        return None

    # Convert screenshot to PIL for uniform coordinate and color lookup
    if isinstance(screenshot_data, (bytes, bytearray)):
        img = Image.open(BytesIO(screenshot_data))
    elif isinstance(screenshot_data, np.ndarray):
        img = Image.fromarray(screenshot_data)
    else:
        img = screenshot_data

    actual_w, actual_h = img.size

    # Scale coordinates if original resolution is specified
    if og_res:
        scale_x = actual_w / og_res[0]
        scale_y = actual_h / og_res[1]
        target_coords = (
            int(target_coords[0] * scale_x),
            int(target_coords[1] * scale_y),
        )

    x, y = target_coords
    if not (0 <= x < actual_w and 0 <= y < actual_h):
        return False

    pixel_color = img.getpixel((x, y))
    # Strip alpha channel if present
    actual_color = pixel_color[:3]
    target_rgb = exp_rgb[:3]

    return all(abs(a - t) <= tol for a, t in zip(actual_color, target_rgb))

check_text(text_to_find, pymordial_screenshot=None, case_sensitive=False, strategy=None)

Checks if text is visible on screen using the OCR device.

Source code in .venv/lib/python3.13/site-packages/pymordialdroid/devices/ui_device.py

def check_text(
    self,
    text_to_find: str,
    pymordial_screenshot: Path | bytes | str | np.ndarray | None = None,
    case_sensitive: bool = False,
    strategy: PymordialExtractStrategy | None = None,
) -> bool:
    """Checks if text is visible on screen using the OCR device."""
    screenshot = self._ensure_screenshot(pymordial_screenshot)
    if screenshot is None:
        return False

    try:
        extracted = self._ocr_device.extract_text(screenshot, strategy=strategy)
        if case_sensitive:
            return text_to_find in extracted
        return text_to_find.lower() in extracted.lower()
    except Exception as e:
        log.error(f"Error checking text: {e}")
        return False

find_text(text_to_find, pymordial_screenshot=None, strategy=None)

Finds the center coordinates of specified text using the OCR device.

Source code in .venv/lib/python3.13/site-packages/pymordialdroid/devices/ui_device.py

def find_text(
    self,
    text_to_find: str,
    pymordial_screenshot: Path | bytes | str | np.ndarray | None = None,
    strategy: PymordialExtractStrategy | None = None,
) -> tuple[int, int] | None:
    """Finds the center coordinates of specified text using the OCR device."""
    screenshot = self._ensure_screenshot(pymordial_screenshot)
    if screenshot is None:
        return None

    if hasattr(self._ocr_device, "find_text"):
        return self._ocr_device.find_text(
            text_to_find, screenshot, strategy=strategy
        )
    return None

initialize(config=None)

Initializes the UI vision plugin.

Source code in .venv/lib/python3.13/site-packages/pymordialdroid/devices/ui_device.py

def initialize(self, config: Any = None) -> None:
    """Initializes the UI vision plugin."""
    pass

read_text(pymordial_screenshot=None, case_sensitive=False, strategy=None)

Reads text lines from the screen using the OCR device.

Lines are always returned verbatim as recognized; case_sensitive is accepted for signature compatibility but no longer mutates the output (case-insensitive matching belongs in :meth:check_text).

Source code in .venv/lib/python3.13/site-packages/pymordialdroid/devices/ui_device.py

def read_text(
    self,
    pymordial_screenshot: Path | bytes | str | np.ndarray | None = None,
    case_sensitive: bool = False,
    strategy: PymordialExtractStrategy | None = None,
) -> list[str]:
    """Reads text lines from the screen using the OCR device.

    Lines are always returned verbatim as recognized; ``case_sensitive``
    is accepted for signature compatibility but no longer mutates the
    output (case-insensitive *matching* belongs in :meth:`check_text`).
    """
    screenshot = self._ensure_screenshot(pymordial_screenshot)
    if screenshot is None:
        return []

    try:
        text = self._ocr_device.extract_text(screenshot, strategy=strategy)
        return [line.strip() for line in text.split("\n") if line.strip()]
    except Exception as e:
        log.error(f"Error reading text: {e}")
        return []

scale_img_to_screen(image_path, screen_image, ref_resolution=None)

Scales a template image to match the device screen resolution.

Parameters:

Name	Type	Description	Default
image_path	str | Path	Path to the reference image.	required
screen_image	str | Image | bytes | ndarray	The current screen image.	required
ref_resolution	tuple[int, int] | None	Original resolution (width, height) the template was captured at.	None

Returns:

Type	Description
Image	Scaled PIL Image.

Source code in .venv/lib/python3.13/site-packages/pymordialdroid/devices/ui_device.py

def scale_img_to_screen(
    self,
    image_path: str | Path,
    screen_image: str | Image.Image | bytes | np.ndarray,
    ref_resolution: tuple[int, int] | None = None,
) -> Image.Image:
    """Scales a template image to match the device screen resolution.

    Args:
        image_path: Path to the reference image.
        screen_image: The current screen image.
        ref_resolution: Original resolution (width, height) the template was captured at.

    Returns:
        Scaled PIL Image.
    """
    if isinstance(screen_image, (bytes, bytearray)):
        screen_img = Image.open(BytesIO(screen_image))
    elif isinstance(screen_image, np.ndarray):
        screen_img = Image.fromarray(cv2.cvtColor(screen_image, cv2.COLOR_BGR2RGB))
    elif isinstance(screen_image, (str, Path)):
        screen_img = Image.open(str(screen_image))
    elif isinstance(screen_image, Image.Image):
        screen_img = screen_image
    else:
        raise ValueError(f"Unsupported screen image type: {type(screen_image)}")

    needle_img = Image.open(str(image_path))
    if not ref_resolution:
        return needle_img

    screen_w, screen_h = screen_img.size
    ref_w, ref_h = ref_resolution

    if ref_w <= 0 or ref_h <= 0:
        return needle_img

    ratio_w = screen_w / ref_w
    ratio_h = screen_h / ref_h

    scaled_size = (
        max(1, int(needle_img.size[0] * ratio_w)),
        max(1, int(needle_img.size[1] * ratio_h)),
    )
    return needle_img.resize(scaled_size, Image.Resampling.BICUBIC)

set_bridge_device(bridge_device)

Sets the underlying bridge device.

Source code in .venv/lib/python3.13/site-packages/pymordialdroid/devices/ui_device.py

def set_bridge_device(self, bridge_device: PymordialBridgeDevice) -> None:
    """Sets the underlying bridge device."""
    self.bridge_device = bridge_device

set_ocr_device(ocr_device)

Sets the OCR device.

Source code in .venv/lib/python3.13/site-packages/pymordialdroid/devices/ui_device.py

def set_ocr_device(self, ocr_device: PymordialOCRDevice) -> None:
    """Sets the OCR device."""
    self._ocr_device = ocr_device

shutdown()

Performs cleanup.

Source code in .venv/lib/python3.13/site-packages/pymordialdroid/devices/ui_device.py

def shutdown(self) -> None:
    """Performs cleanup."""
    pass

where_element(element, screenshot=None, max_tries=1, set_position=False, set_size=False, wait_time=0.5, pymordial_screenshot=None)

Finds (center_x, center_y) coordinates of a UI element on screen.

Source code in .venv/lib/python3.13/site-packages/pymordialdroid/devices/ui_device.py

def where_element(
    self,
    element: PymordialElement,
    screenshot: bytes | np.ndarray | None = None,
    max_tries: int = 1,
    set_position: bool = False,
    set_size: bool = False,
    wait_time: float = 0.5,
    pymordial_screenshot: bytes | np.ndarray | None = None,
) -> tuple[int, int] | None:
    """Finds (center_x, center_y) coordinates of a UI element on screen."""
    target_screen = screenshot if screenshot is not None else pymordial_screenshot
    if isinstance(element, PymordialPixel):
        if self.check_pixel_color(
            pymordial_pixel=element, pymordial_screenshot=target_screen
        ):
            return (int(element.position[0]), int(element.position[1]))
        return None

    if isinstance(element, PymordialText):
        text = getattr(element, "element_text", getattr(element, "text", ""))
        strat = getattr(element, "extract_strategy", None)
        return self.find_text(text, pymordial_screenshot=screenshot, strategy=strat)

    if not isinstance(element, PymordialImage):
        # Bounding box center fallback
        return getattr(element, "center", None)

    # Handle PymordialImage
    raw_path = getattr(element, "filepath", getattr(element, "source_path", ""))
    template_path = Path(raw_path)
    if not template_path.exists():
        log.warning(f"Template image not found: {template_path}")
        return None

    og_res = getattr(element, "og_resolution", None)
    confidence = getattr(element, "confidence", 0.8)

    for attempt in range(max_tries):
        screen_data = self._ensure_screenshot(target_screen)
        if screen_data is None:
            if attempt < max_tries - 1:
                time.sleep(wait_time)
                continue
            return None

        try:
            # Convert screen data to PIL Image
            if isinstance(screen_data, (bytes, bytearray)):
                haystack_pil = Image.open(BytesIO(screen_data))
            elif isinstance(screen_data, np.ndarray):
                haystack_pil = Image.fromarray(screen_data)
            elif isinstance(screen_data, Image.Image):
                haystack_pil = screen_data
            else:
                haystack_pil = Image.open(str(screen_data))

            # Scale template needle to match current screen resolution
            scaled_needle_pil = self.scale_img_to_screen(
                image_path=template_path,
                screen_image=haystack_pil,
                ref_resolution=og_res,
            )

            # Prepare OpenCV BGR images
            haystack_cv = cv2.cvtColor(np.array(haystack_pil), cv2.COLOR_RGB2BGR)
            needle_cv = cv2.cvtColor(np.array(scaled_needle_pil), cv2.COLOR_RGB2BGR)

            region = getattr(element, "region", None)
            if region:
                rx, ry, rw, rh = region
                haystack_cv = haystack_cv[ry : ry + rh, rx : rx + rw]
                offset_x, offset_y = rx, ry
            else:
                offset_x, offset_y = 0, 0

            # Template matching
            result = cv2.matchTemplate(haystack_cv, needle_cv, cv2.TM_CCOEFF_NORMED)
            _, max_val, _, max_loc = cv2.minMaxLoc(result)

            if max_val >= confidence:
                match_x = max_loc[0] + offset_x
                match_y = max_loc[1] + offset_y
                needle_w, needle_h = scaled_needle_pil.size

                center_coords = (
                    match_x + needle_w // 2,
                    match_y + needle_h // 2,
                )

                if set_position:
                    element.position = (match_x, match_y)
                if set_size:
                    element.size = (needle_w, needle_h)

                return center_coords

        except Exception as e:
            log.error(f"Error finding element {element.label}: {e}")

        if attempt < max_tries - 1:
            time.sleep(wait_time)
            screenshot = None  # Force fresh screenshot capture

    return None

where_elements(elements, screenshot=None, max_tries=1, pymordial_screenshot=None)

Finds the coordinates of the first matching element from a list.

Source code in .venv/lib/python3.13/site-packages/pymordialdroid/devices/ui_device.py

def where_elements(
    self,
    elements: list[PymordialElement],
    screenshot: bytes | np.ndarray | None = None,
    max_tries: int = 1,
    pymordial_screenshot: bytes | np.ndarray | None = None,
) -> tuple[int, int] | None:
    """Finds the coordinates of the first matching element from a list."""
    target_screen = screenshot if screenshot is not None else pymordial_screenshot
    for element in elements:
        coords = self.where_element(
            element, screenshot=target_screen, max_tries=max_tries
        )
        if coords is not None:
            return coords
    return None

Tesseract Device

pymordialdroid.devices.tesseract_device

Tesseract OCR device implementing Pymordial's PymordialOCRDevice contract.

TesseractDevice

Bases: PymordialOCRDevice

Optical Character Recognition device powered by Tesseract and OpenCV.

Source code in .venv/lib/python3.13/site-packages/pymordialdroid/devices/tesseract_device.py

class TesseractDevice(PymordialOCRDevice):
    """Optical Character Recognition device powered by Tesseract and OpenCV."""

    name: str = "ocr"
    version: str = "0.1.0"

    def __init__(
        self,
        config: str = "--oem 3 --psm 6",
        system_config: SystemConfig | None = None,
    ) -> None:
        self.config = config
        self.system_config = system_config or resolve_system_config()

        if self.system_config.tesseract_bin_path:
            pytesseract.pytesseract.tesseract_cmd = str(
                self.system_config.tesseract_bin_path
            )
            log.info(
                f"Using configured Tesseract: {self.system_config.tesseract_bin_path}"
            )

    def initialize(self, config: Any = None) -> None:
        """Initializes the OCR device plugin."""
        pass

    def shutdown(self) -> None:
        """Cleans up resources."""
        pass

    def _load_image(self, image_path: Path | bytes | str | np.ndarray) -> np.ndarray:
        """Loads and normalizes an image from various input types into an OpenCV BGR numpy array."""
        if isinstance(image_path, np.ndarray):
            return image_path

        if isinstance(image_path, (bytes, bytearray)):
            nparr = np.frombuffer(image_path, np.uint8)
            image = cv2.imdecode(nparr, cv2.IMREAD_COLOR)
        else:
            image = cv2.imread(str(image_path))

        if image is None:
            raise ValueError(f"Could not load image from {type(image_path)}")

        return image

    def extract_text(
        self,
        image_path: Path | bytes | str | np.ndarray,
        strategy: PymordialExtractStrategy | None = None,
    ) -> str:
        """Extracts text from an image with optional preprocessing.

        Args:
            image_path: Source image as Path, bytes, string, or numpy array.
            strategy: Optional extraction strategy (defaults to DefaultExtractStrategy).

        Returns:
            Extracted text string.
        """
        try:
            image = self._load_image(image_path)
            strat = strategy or DefaultExtractStrategy()
            processed = strat.preprocess(image)

            tess_config = (
                getattr(strat, "tesseract_config", lambda: self.config)() or self.config
            )
            text = pytesseract.image_to_string(processed, config=tess_config)

            postprocess = getattr(strat, "postprocess_text", lambda t: t.strip())
            return postprocess(text)
        except Exception as e:
            log.error(f"Error extracting text with Tesseract: {e}")
            raise ValueError(f"Failed to extract text: {e}") from e

    def find_text(
        self,
        search_text: str,
        image_path: Path | bytes | str | np.ndarray,
        strategy: PymordialExtractStrategy | None = None,
    ) -> tuple[int, int] | None:
        """Finds (center_x, center_y) coordinates of search_text in the source image.

        Args:
            search_text: Text keyword or substring to search for.
            image_path: Source image as Path, bytes, string, or numpy array.
            strategy: Optional preprocessing strategy.

        Returns:
            (center_x, center_y) coordinates if located, None otherwise.
        """
        try:
            image = self._load_image(image_path)
            strat = strategy or DefaultExtractStrategy()
            processed = strat.preprocess(image)

            tess_config = (
                getattr(strat, "tesseract_config", lambda: self.config)() or self.config
            )
            data = pytesseract.image_to_data(
                processed, config=tess_config, output_type=pytesseract.Output.DICT
            )

            search_lower = search_text.lower().strip()
            n_boxes = len(data.get("text", []))
            upscale = getattr(strat, "upscale_factor", 1) or 1

            for i in range(n_boxes):
                conf = int(data["conf"][i])
                if conf > 0:
                    box_text = data["text"][i].strip().lower()
                    if search_lower in box_text:
                        x, y, w, h = (
                            data["left"][i],
                            data["top"][i],
                            data["width"][i],
                            data["height"][i],
                        )
                        center_x = int((x + w // 2) / upscale)
                        center_y = int((y + h // 2) / upscale)
                        return (center_x, center_y)

            return None
        except Exception as e:
            log.error(f"Error finding text '{search_text}' with Tesseract: {e}")
            return None

extract_text(image_path, strategy=None)

Extracts text from an image with optional preprocessing.

Parameters:

Name	Type	Description	Default
image_path	Path | bytes | str | ndarray	Source image as Path, bytes, string, or numpy array.	required
strategy	PymordialExtractStrategy | None	Optional extraction strategy (defaults to DefaultExtractStrategy).	None

Returns:

Type	Description
str	Extracted text string.

Source code in .venv/lib/python3.13/site-packages/pymordialdroid/devices/tesseract_device.py

def extract_text(
    self,
    image_path: Path | bytes | str | np.ndarray,
    strategy: PymordialExtractStrategy | None = None,
) -> str:
    """Extracts text from an image with optional preprocessing.

    Args:
        image_path: Source image as Path, bytes, string, or numpy array.
        strategy: Optional extraction strategy (defaults to DefaultExtractStrategy).

    Returns:
        Extracted text string.
    """
    try:
        image = self._load_image(image_path)
        strat = strategy or DefaultExtractStrategy()
        processed = strat.preprocess(image)

        tess_config = (
            getattr(strat, "tesseract_config", lambda: self.config)() or self.config
        )
        text = pytesseract.image_to_string(processed, config=tess_config)

        postprocess = getattr(strat, "postprocess_text", lambda t: t.strip())
        return postprocess(text)
    except Exception as e:
        log.error(f"Error extracting text with Tesseract: {e}")
        raise ValueError(f"Failed to extract text: {e}") from e

find_text(search_text, image_path, strategy=None)

Finds (center_x, center_y) coordinates of search_text in the source image.

Parameters:

Name	Type	Description	Default
search_text	str	Text keyword or substring to search for.	required
image_path	Path | bytes | str | ndarray	Source image as Path, bytes, string, or numpy array.	required
strategy	PymordialExtractStrategy | None	Optional preprocessing strategy.	None

Returns:

Type	Description
tuple[int, int] | None	(center_x, center_y) coordinates if located, None otherwise.

Source code in .venv/lib/python3.13/site-packages/pymordialdroid/devices/tesseract_device.py

def find_text(
    self,
    search_text: str,
    image_path: Path | bytes | str | np.ndarray,
    strategy: PymordialExtractStrategy | None = None,
) -> tuple[int, int] | None:
    """Finds (center_x, center_y) coordinates of search_text in the source image.

    Args:
        search_text: Text keyword or substring to search for.
        image_path: Source image as Path, bytes, string, or numpy array.
        strategy: Optional preprocessing strategy.

    Returns:
        (center_x, center_y) coordinates if located, None otherwise.
    """
    try:
        image = self._load_image(image_path)
        strat = strategy or DefaultExtractStrategy()
        processed = strat.preprocess(image)

        tess_config = (
            getattr(strat, "tesseract_config", lambda: self.config)() or self.config
        )
        data = pytesseract.image_to_data(
            processed, config=tess_config, output_type=pytesseract.Output.DICT
        )

        search_lower = search_text.lower().strip()
        n_boxes = len(data.get("text", []))
        upscale = getattr(strat, "upscale_factor", 1) or 1

        for i in range(n_boxes):
            conf = int(data["conf"][i])
            if conf > 0:
                box_text = data["text"][i].strip().lower()
                if search_lower in box_text:
                    x, y, w, h = (
                        data["left"][i],
                        data["top"][i],
                        data["width"][i],
                        data["height"][i],
                    )
                    center_x = int((x + w // 2) / upscale)
                    center_y = int((y + h // 2) / upscale)
                    return (center_x, center_y)

        return None
    except Exception as e:
        log.error(f"Error finding text '{search_text}' with Tesseract: {e}")
        return None

initialize(config=None)

Initializes the OCR device plugin.

Source code in .venv/lib/python3.13/site-packages/pymordialdroid/devices/tesseract_device.py

def initialize(self, config: Any = None) -> None:
    """Initializes the OCR device plugin."""
    pass

shutdown()

Cleans up resources.

Source code in .venv/lib/python3.13/site-packages/pymordialdroid/devices/tesseract_device.py

def shutdown(self) -> None:
    """Cleans up resources."""
    pass