Controller API

pymordialblue.bluestacks_controller

Main controller for the Pymordial automation framework.

PymordialBluestacksController

Bases: PymordialController

Main controller that orchestrates device interaction via plugins.

This controller manages the lifecycle of connected devices (ADB, UI, Emulator) using the Plugin Registry.

Attributes:

Name	Type	Description
adb		The PymordialAdbDevice instance.
ui		The PymordialUiDevice instance.
bluestacks		The PymordialBluestacksDevice instance.

Source code in src/pymordialblue/bluestacks_controller.py

class PymordialBluestacksController(PymordialController):
    """Main controller that orchestrates device interaction via plugins.

    This controller manages the lifecycle of connected devices (ADB, UI, Emulator)
    using the Plugin Registry.


    Attributes:
        adb: The PymordialAdbDevice instance.
        ui: The PymordialUiDevice instance.
        bluestacks: The PymordialBluestacksDevice instance.
    """

    DEFAULT_CLICK_TIMES = _CONFIG["controller"]["default_click_times"]
    DEFAULT_MAX_TRIES = _CONFIG["controller"]["default_max_tries"]
    CLICK_COORD_TIMES = _CONFIG["controller"]["click_coord_times"]
    CMD_TAP = _CONFIG["adb"]["commands"]["tap"]

    def __init__(
        self,
        adb_host: str | None = None,
        adb_port: int | None = None,
        apps: list["PymordialAndroidApp"] | None = None,
    ):
        """Initializes the PymordialController.

        Args:
            adb_host: Optional ADB host address.
            adb_port: Optional ADB port.
            apps: Optional list of PymordialAndroidApp instances to register.
        """
        super().__init__(apps=apps)
        self.registry = PluginRegistry(config=_CONFIG)
        self.registry.load_from_entry_points()

        # 1. Resolve ADB
        self.adb = self._resolve_plugin(
            "adb",
            lambda: PymordialAdbDevice(host=adb_host, port=adb_port),
        )

        # 2. Resolve UI
        def configure_ui(plugin: "PymordialPlugin") -> None:
            if hasattr(plugin, "set_bridge_device"):
                plugin.set_bridge_device(self.adb)

        self.ui = self._resolve_plugin(
            "ui",
            lambda: PymordialUiDevice(bridge_device=self.adb),
            configure_found_plugin=configure_ui,
        )

        # 3. Resolve BlueStacks
        def configure_bluestacks(plugin: "PymordialPlugin") -> None:
            if hasattr(plugin, "set_dependencies"):
                plugin.set_dependencies(self.adb, self.ui)

        self.bluestacks = self._resolve_plugin(
            "bluestacks",
            lambda: PymordialBluestacksDevice(self.adb, self.ui),
            configure_found_plugin=configure_bluestacks,
        )

        self._streaming_enabled = False  # Track if streaming should be active

        if apps:
            for app in apps:
                self.add_app(app)

    def _resolve_plugin(
        self,
        name: str,
        default_factory: Callable[[], "PymordialPlugin"],
        configure_found_plugin: Callable[["PymordialPlugin"], None] | None = None,
    ) -> "PymordialPlugin":
        """Resolves a plugin from the registry or falls back to a default.

        Args:
            name: The name of the plugin to resolve (e.g., 'adb').
            default_factory: A function that returns a default plugin instance if not found.
            configure_found_plugin: Optional callback to configure the found plugin (dependency injection).

        Returns:
            The resolved or default plugin instance.
        """
        try:
            plugin = self.registry.get(name)
            logger.info("Using %s plugin: %s", name.upper(), plugin.name)
            if configure_found_plugin:
                configure_found_plugin(plugin)
            return plugin
        except KeyError:
            logger.debug(
                "%s plugin not found. Using default implementation.", name.upper()
            )
            default_plugin = default_factory()
            self.registry.register(default_plugin)
            return default_plugin

    # --- Convenience Methods (delegate to sub-controllers) ---
    # --- App Lifecycle Methods (implement base ABC) ---
    def open_app(
        self,
        app_name: str | PymordialAndroidApp,
        package_name: str | None = None,
        timeout: int | None = None,
        wait_time: int | None = None,
    ) -> bool:
        """Opens an app on the device.

        Args:
            app_name: The display name of the app or a PymordialAndroidApp instance.
            package_name: The Android package name.
            timeout: Maximum seconds to wait for launch.
            wait_time: Seconds to wait after launch command.

        Returns:
            True if the app launched successfully, False otherwise.
        """
        if isinstance(app_name, PymordialAndroidApp):
            # If a PymordialAndroidApp is passed, extract package_name if not provided
            if package_name is None and hasattr(app_name, "package_name"):
                package_name = app_name.package_name
            app_name = app_name.app_name

        # Resolve defaults from config if not provided
        timeout = timeout or _CONFIG["adb"]["app_start_timeout"]
        wait_time = wait_time or _CONFIG["adb"]["default_wait_time"]

        return self.adb.open_app(
            app_name=app_name,
            package_name=package_name,
            timeout=float(timeout),
            wait_time=float(wait_time),
        )

    def close_app(
        self,
        app_name: str | PymordialAndroidApp,
        package_name: str | None = None,
        timeout: int | None = None,
        wait_time: int | None = None,
    ) -> bool:
        """Closes an app on the device.

        Args:
            app_name: The display name of the app or a PymordialAndroidApp instance.
            package_name: The Android package name.
            timeout: Maximum seconds to wait for closure.
            wait_time: Seconds to wait after close command.

        Returns:
            True if the app closed successfully, False otherwise.
        """
        if isinstance(app_name, PymordialAndroidApp):
            if package_name is None and hasattr(app_name, "package_name"):
                package_name = app_name.package_name
            app_name = app_name.app_name

        timeout = timeout or _CONFIG["adb"]["commands"]["timeout"]
        wait_time = wait_time or _CONFIG["adb"]["default_wait_time"]

        return self.adb.close_app(
            package_name=package_name,
            app_name=app_name,
            timeout=float(timeout),
            wait_time=float(wait_time),
        )

    def capture_screen(self) -> bytes | None:
        """Captures the current screen.

        Returns:
            Screenshot as bytes, or None if failed.

        Convenience method that delegates to adb.capture_screenshot().
        """
        return self.adb.capture_screen()

    def disconnect(self) -> None:
        """Closes the ADB connection and performs cleanup."""
        if self.adb.is_connected():
            self.adb.disconnect()

    ## --- Click Methods ---
    def click_coord(
        self, coords: tuple[int, int], times: int = CLICK_COORD_TIMES
    ) -> bool:
        """Clicks specific coordinates on the screen.

        Args:
            coords: (x, y) coordinates to click.
            times: Number of times to click.

        Returns:
            True if the click was sent successfully, False otherwise.
        """
        # Ensure Bluestacks is ready before trying to click coords
        match self.bluestacks.state.current_state:
            case EmulatorState.CLOSED | EmulatorState.LOADING:
                logger.warning("Cannot click coords - Bluestacks is not ready")
                return False
            case EmulatorState.READY:
                is_connected = self.adb.is_connected()
                if not is_connected:
                    logger.warning(
                        "ADB device not connected. Skipping 'click_coords' method call."
                    )
                    return False
                single_tap = self.CMD_TAP.format(x=coords[0], y=coords[1])
                tap_command = " && ".join([single_tap] * times)

                self.adb.run_command(tap_command)
                logger.debug(
                    f"Click event sent via ADB at coords x={coords[0]}, y={coords[1]}"
                )
                return True

    def click_element(
        self,
        pymordial_element: PymordialElement,
        times: int = DEFAULT_CLICK_TIMES,
        screenshot_img_bytes: bytes | None = None,
        max_tries: int = DEFAULT_MAX_TRIES,
    ) -> bool:
        """Clicks a UI element on the screen.

        Args:
            pymordial_element: The element to click.
            times: Optional number of times to click. Defaults to DEFAULT_CLICK_TIMES config.
            screenshot_img_bytes: Optional pre-captured screenshot to look for the element in. Defaults to None.
            max_tries: Optional maximum number of retries to find the element. Defaults to DEFAULT_MAX_TRIES config.

        Returns:
            True if the element was found and clicked, False otherwise.
        """
        # Ensure Bluestacks is ready before trying to click ui
        match self.bluestacks.state.current_state:
            case EmulatorState.CLOSED | EmulatorState.LOADING:
                logger.warning("Cannot click coords - Bluestacks is not ready")
                return False
            case EmulatorState.READY:
                if not self.adb.is_connected():
                    self.adb.connect()
                    if not self.adb.is_connected():
                        logger.warning(
                            "ADB device not connected. Skipping 'click_element' method call."
                        )
                        return False
                coord: tuple[int, int] | None = self.find_element(
                    pymordial_element=pymordial_element,
                    pymordial_screenshot=screenshot_img_bytes,
                    max_tries=max_tries,
                )
                if not coord:
                    logger.debug(f"UI element {pymordial_element.label} not found")
                    return False
                if self.click_coord(coord, times=times):
                    logger.debug(
                        f"Click event sent via ADB at coords x={coord[0]}, y={coord[1]}"
                    )
                    return True
                return False
            case _:
                logger.warning(
                    "Cannot click coords - Bluestacks state is not in a valid state."
                    " Make sure it is in the 'EmulatorState.READY' state."
                )
                return False

    def click_elements(
        self,
        pymordial_elements: list[PymordialElement],
        screenshot_img_bytes: bytes | None = None,
        max_tries: int = DEFAULT_MAX_TRIES,
    ) -> bool:
        """Clicks any of the elements in the list.

        Args:
            pymordial_elements: List of elements to try clicking.
            screenshot_img_bytes: Optional pre-captured screenshot.
            max_tries: Maximum number of retries per element.

        Returns:
            True if any element was clicked, False otherwise.
        """
        return any(
            self.click_element(
                pymordial_element=pymordial_element,
                screenshot_img_bytes=screenshot_img_bytes,
                max_tries=max_tries,
            )
            for pymordial_element in pymordial_elements
        )

    def go_home(self) -> None:
        """Navigate to Android home screen.

        Convenience method that delegates to adb.go_home().
        """
        self.adb.go_home()

    def go_back(self) -> None:
        """Press Android back button.

        Convenience method that delegates to adb.go_back().
        """
        self.adb.go_back()

    def tap(self, x: int, y: int) -> None:
        """Tap at specific coordinates.

        Args:
            x: X coordinate.
            y: Y coordinate.

        Convenience method that delegates to adb.tap().
        """
        return self.adb.tap(x, y)

    def swipe(
        self, start_x: int, start_y: int, end_x: int, end_y: int, duration: int = 300
    ) -> None:
        """Perform swipe gesture.

        Args:
            start_x: Starting X coordinate.
            start_y: Starting Y coordinate.
            end_x: Ending X coordinate.
            end_y: Ending Y coordinate.
            duration: Swipe duration in milliseconds.

        Convenience method that delegates to adb.swipe().
        """
        return self.adb.swipe(start_x, start_y, end_x, end_y, duration)

    def find_element(
        self,
        pymordial_element: PymordialElement,
        pymordial_screenshot: bytes | None = None,
        max_tries: int = DEFAULT_MAX_TRIES,
    ) -> tuple[int, int] | None:
        """Finds the coordinates of a UI element on the screen.

        Args:
            pymordial_element: The element to find.
            pymordial_screenshot: Optional pre-captured screenshot.
            max_tries: Maximum number of retries.

        Returns:
            (x, y) coordinates if found, None otherwise.
        """
        if isinstance(pymordial_element, PymordialImage):
            return self.ui.where_element(
                pymordial_element=pymordial_element,
                pymordial_screenshot=pymordial_screenshot,
                max_tries=max_tries,
            )
        elif isinstance(pymordial_element, PymordialText):
            return self.ui.find_text(
                text_to_find=pymordial_element.element_text,
                pymordial_screenshot=pymordial_screenshot,
                strategy=pymordial_element.extract_strategy,
            )
        elif isinstance(pymordial_element, PymordialPixel):
            # Capture screenshot if not provided (avoid 'or' with numpy arrays)
            pixel_screenshot = (
                pymordial_screenshot
                if pymordial_screenshot is not None
                else self.capture_screen()
            )
            is_match = self.ui.check_pixel_color(
                pymordial_pixel=pymordial_element,
                pymordial_screenshot=pixel_screenshot,
            )
            return pymordial_element.position if is_match else None

        raise NotImplementedError(
            f"find_element() not implemented for this element type: {type(pymordial_element)}"
        )

    def is_element_visible(
        self,
        pymordial_element: PymordialElement,
        pymordial_screenshot: bytes | None = None,
        max_tries: int | None = None,
    ) -> bool:
        """Checks if a UI element is visible on the screen.

        Args:
            pymordial_element: The element to check for.
            pymordial_screenshot: Optional pre-captured screenshot.
            max_tries: Optional maximum number of retries.

        Returns:
            True if the element is found, False otherwise.
        """
        if not isinstance(pymordial_element, PymordialElement):
            raise TypeError(
                f"pymordial_element must be an instance of PymordialElement, not {type(pymordial_element)}"
            )

        if isinstance(pymordial_element, PymordialImage):
            return (
                self.find_element(
                    pymordial_element=pymordial_element,
                    pymordial_screenshot=pymordial_screenshot,
                    max_tries=max_tries or self.DEFAULT_MAX_TRIES,
                )
                is not None
            )
        elif isinstance(pymordial_element, PymordialText):
            # For text, we use the text controller to check existence
            # Note: This doesn't return coordinates yet, so click_element won't work for Text
            # unless find_element is implemented for Text.

            # If the element has a defined region, crop the image to that region
            if pymordial_element.region and pymordial_screenshot is not None:
                try:
                    if isinstance(pymordial_screenshot, bytes):
                        pymordial_screenshot = Image.open(BytesIO(pymordial_screenshot))
                    elif isinstance(pymordial_screenshot, np.ndarray):
                        pymordial_screenshot = Image.fromarray(pymordial_screenshot)
                    else:
                        pymordial_screenshot = None

                    if pymordial_screenshot is not None:
                        # region is (left, top, right, bottom)
                        pymordial_screenshot = pymordial_screenshot.crop(
                            pymordial_element.region
                        )
                        pymordial_screenshot = np.array(pymordial_screenshot)
                except Exception as e:
                    logger.warning(f"Failed to crop image for text detection: {e}")

            return self.ui.check_text(
                text_to_find=pymordial_element.element_text,
                pymordial_screenshot=pymordial_screenshot,
                strategy=pymordial_element.extract_strategy,
                case_sensitive=False,
            )
        elif isinstance(pymordial_element, PymordialPixel):
            return (
                self.find_element(
                    pymordial_element=pymordial_element,
                    pymordial_screenshot=pymordial_screenshot,
                    max_tries=max_tries or self.DEFAULT_MAX_TRIES,
                )
                is not None
            )
        else:
            raise NotImplementedError(
                f"is_element_visible not implemented for {type(pymordial_element)}"
            )

    # --- Input Methods ---

    def press_enter(self) -> None:
        """Press the Enter key.

        Convenience method that delegates to adb.press_enter().
        """
        return self.adb.press_enter()

    def press_esc(self) -> None:
        """Press the Esc key.

        Convenience method that delegates to adb.press_esc().
        """
        return self.adb.press_esc()

    def type_text(self, text: str, enter: bool = False) -> None:
        """Send text input to the device.

        Args:
            text: Text to send.
            enter: Whether to press enter after typing.

        Convenience method that delegates to adb.type_text().
        """
        return self.adb.type_text(text, enter)

    # --- Shell & Utility Methods ---

    def run_command(self, command: str) -> bytes | None:
        """Execute ADB shell command.

        Args:
            command: Shell command to execute.

        Returns:
            Command output as bytes, or None if failed.

        Convenience method that delegates to adb.run_command().
        """
        return self.adb.run_command(command)

    def get_current_app(self) -> str | None:
        """Get the currently running app's package name.

        Returns:
            Package name of current app, or None if failed.

        Convenience method that delegates to adb.get_current_app().
        """
        return self.adb.get_current_app()

    # --- OCR Methods ---

    def read_text(
        self,
        image_path: "Path | bytes | str",
        case_sensitive: bool = False,
        strategy: "PymordialExtractStrategy | None" = None,
    ) -> list[str]:
        """Read text from an image using OCR.

        Args:
            image_path: Path to image file, image bytes, or string path.
            strategy: Optional preprocessing strategy.

        Returns:
            List of detected text lines.

        Convenience method that delegates to text.read_text().
        """
        return self.ui.read_text(image_path, case_sensitive, strategy)

    def check_text(
        self,
        text_to_find: str,
        image_path: "Path | bytes | str",
        case_sensitive: bool = False,
        strategy: "PymordialExtractStrategy | None" = None,
    ) -> bool:
        """Check if specific text exists in an image.

        Args:
            text_to_find: Text to search for.
            image_path: Image to search in.
            case_sensitive: Whether search is case-sensitive.
            strategy: Optional preprocessing strategy.

        Returns:
            True if text found, False otherwise.

        Convenience method that delegates to text.check_text().
        """
        return self.ui.check_text(text_to_find, image_path, case_sensitive, strategy)

    # --- State Checking Methods ---

    def is_bluestacks_ready(self) -> bool:
        """Check if BlueStacks is in READY state.

        Returns:
            True if BlueStacks is ready, False otherwise.

        Convenience method that delegates to self.bluestacks.is_ready().
        """
        return self.bluestacks.is_ready()

    def is_bluestacks_loading(self) -> bool:
        """Check if BlueStacks is currently loading.

        Returns:
            True if BlueStacks is loading, False otherwise.

        Convenience method that delegates to bluestacks.is_loading().
        """
        return self.bluestacks.is_loading()

    # --- Streaming Methods ---

    def start_streaming(self) -> bool:
        """Starts the screen stream using ADB device.

        Returns:
            True if streaming started successfully, False otherwise.

        Convenience method that delegates to adb.start_stream().
        """
        result = self.adb.start_stream()
        if result:
            self._streaming_enabled = True
        return result

    def get_frame(self) -> "np.ndarray | None":
        """Get the latest frame from the active stream.

        Returns:
            Latest frame as numpy array (RGB), or None if unavailable.

        Convenience method that delegates to adb.get_latest_frame().

        Example:
            >>> frame = controller.get_frame()
            >>> if frame is not None:
            ...     # Process frame (OCR, template matching, etc.)
            ...     text = controller.read_text(frame)
        """
        return self.adb.get_latest_frame()

    def stop_streaming(self) -> None:
        """Stop the active video stream and disable auto-restart.

        Convenience method that delegates to adb.stop_stream().
        """
        self._streaming_enabled = False
        return self.adb.stop_stream()

    def __repr__(self) -> str:
        """Returns a string representation of the PymordialController."""
        return (
            f"PymordialController("
            f"apps={len(self._apps)}, "
            f"adb_connected={self.adb.is_connected()}, "
            f"bluestacks={self.bluestacks.state.current_state.name})"
        )

__init__(adb_host=None, adb_port=None, apps=None)

Initializes the PymordialController.

Parameters:

Name	Type	Description	Default
adb_host	str | None	Optional ADB host address.	None
adb_port	int | None	Optional ADB port.	None
apps	list[PymordialAndroidApp] | None	Optional list of PymordialAndroidApp instances to register.	None

Source code in src/pymordialblue/bluestacks_controller.py

def __init__(
    self,
    adb_host: str | None = None,
    adb_port: int | None = None,
    apps: list["PymordialAndroidApp"] | None = None,
):
    """Initializes the PymordialController.

    Args:
        adb_host: Optional ADB host address.
        adb_port: Optional ADB port.
        apps: Optional list of PymordialAndroidApp instances to register.
    """
    super().__init__(apps=apps)
    self.registry = PluginRegistry(config=_CONFIG)
    self.registry.load_from_entry_points()

    # 1. Resolve ADB
    self.adb = self._resolve_plugin(
        "adb",
        lambda: PymordialAdbDevice(host=adb_host, port=adb_port),
    )

    # 2. Resolve UI
    def configure_ui(plugin: "PymordialPlugin") -> None:
        if hasattr(plugin, "set_bridge_device"):
            plugin.set_bridge_device(self.adb)

    self.ui = self._resolve_plugin(
        "ui",
        lambda: PymordialUiDevice(bridge_device=self.adb),
        configure_found_plugin=configure_ui,
    )

    # 3. Resolve BlueStacks
    def configure_bluestacks(plugin: "PymordialPlugin") -> None:
        if hasattr(plugin, "set_dependencies"):
            plugin.set_dependencies(self.adb, self.ui)

    self.bluestacks = self._resolve_plugin(
        "bluestacks",
        lambda: PymordialBluestacksDevice(self.adb, self.ui),
        configure_found_plugin=configure_bluestacks,
    )

    self._streaming_enabled = False  # Track if streaming should be active

    if apps:
        for app in apps:
            self.add_app(app)

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

Opens an app on the device.

Parameters:

Name	Type	Description	Default
app_name	str | PymordialAndroidApp	The display name of the app or a PymordialAndroidApp instance.	required
package_name	str | None	The Android package name.	None
timeout	int | None	Maximum seconds to wait for launch.	None
wait_time	int | None	Seconds to wait after launch command.	None

Returns:

Type	Description
bool	True if the app launched successfully, False otherwise.

Source code in src/pymordialblue/bluestacks_controller.py

def open_app(
    self,
    app_name: str | PymordialAndroidApp,
    package_name: str | None = None,
    timeout: int | None = None,
    wait_time: int | None = None,
) -> bool:
    """Opens an app on the device.

    Args:
        app_name: The display name of the app or a PymordialAndroidApp instance.
        package_name: The Android package name.
        timeout: Maximum seconds to wait for launch.
        wait_time: Seconds to wait after launch command.

    Returns:
        True if the app launched successfully, False otherwise.
    """
    if isinstance(app_name, PymordialAndroidApp):
        # If a PymordialAndroidApp is passed, extract package_name if not provided
        if package_name is None and hasattr(app_name, "package_name"):
            package_name = app_name.package_name
        app_name = app_name.app_name

    # Resolve defaults from config if not provided
    timeout = timeout or _CONFIG["adb"]["app_start_timeout"]
    wait_time = wait_time or _CONFIG["adb"]["default_wait_time"]

    return self.adb.open_app(
        app_name=app_name,
        package_name=package_name,
        timeout=float(timeout),
        wait_time=float(wait_time),
    )

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

Closes an app on the device.

Parameters:

Name	Type	Description	Default
app_name	str | PymordialAndroidApp	The display name of the app or a PymordialAndroidApp instance.	required
package_name	str | None	The Android package name.	None
timeout	int | None	Maximum seconds to wait for closure.	None
wait_time	int | None	Seconds to wait after close command.	None

Returns:

Type	Description
bool	True if the app closed successfully, False otherwise.

Source code in src/pymordialblue/bluestacks_controller.py

def close_app(
    self,
    app_name: str | PymordialAndroidApp,
    package_name: str | None = None,
    timeout: int | None = None,
    wait_time: int | None = None,
) -> bool:
    """Closes an app on the device.

    Args:
        app_name: The display name of the app or a PymordialAndroidApp instance.
        package_name: The Android package name.
        timeout: Maximum seconds to wait for closure.
        wait_time: Seconds to wait after close command.

    Returns:
        True if the app closed successfully, False otherwise.
    """
    if isinstance(app_name, PymordialAndroidApp):
        if package_name is None and hasattr(app_name, "package_name"):
            package_name = app_name.package_name
        app_name = app_name.app_name

    timeout = timeout or _CONFIG["adb"]["commands"]["timeout"]
    wait_time = wait_time or _CONFIG["adb"]["default_wait_time"]

    return self.adb.close_app(
        package_name=package_name,
        app_name=app_name,
        timeout=float(timeout),
        wait_time=float(wait_time),
    )

capture_screen()

Captures the current screen.

Returns:

Type	Description
bytes | None	Screenshot as bytes, or None if failed.

Convenience method that delegates to adb.capture_screenshot().

Source code in src/pymordialblue/bluestacks_controller.py

def capture_screen(self) -> bytes | None:
    """Captures the current screen.

    Returns:
        Screenshot as bytes, or None if failed.

    Convenience method that delegates to adb.capture_screenshot().
    """
    return self.adb.capture_screen()

disconnect()

Closes the ADB connection and performs cleanup.

Source code in src/pymordialblue/bluestacks_controller.py

def disconnect(self) -> None:
    """Closes the ADB connection and performs cleanup."""
    if self.adb.is_connected():
        self.adb.disconnect()

click_coord(coords, times=CLICK_COORD_TIMES)

Clicks specific coordinates on the screen.

Parameters:

Name	Type	Description	Default
coords	tuple[int, int]	(x, y) coordinates to click.	required
times	int	Number of times to click.	CLICK_COORD_TIMES

Returns:

Type	Description
bool	True if the click was sent successfully, False otherwise.

Source code in src/pymordialblue/bluestacks_controller.py

def click_coord(
    self, coords: tuple[int, int], times: int = CLICK_COORD_TIMES
) -> bool:
    """Clicks specific coordinates on the screen.

    Args:
        coords: (x, y) coordinates to click.
        times: Number of times to click.

    Returns:
        True if the click was sent successfully, False otherwise.
    """
    # Ensure Bluestacks is ready before trying to click coords
    match self.bluestacks.state.current_state:
        case EmulatorState.CLOSED | EmulatorState.LOADING:
            logger.warning("Cannot click coords - Bluestacks is not ready")
            return False
        case EmulatorState.READY:
            is_connected = self.adb.is_connected()
            if not is_connected:
                logger.warning(
                    "ADB device not connected. Skipping 'click_coords' method call."
                )
                return False
            single_tap = self.CMD_TAP.format(x=coords[0], y=coords[1])
            tap_command = " && ".join([single_tap] * times)

            self.adb.run_command(tap_command)
            logger.debug(
                f"Click event sent via ADB at coords x={coords[0]}, y={coords[1]}"
            )
            return True

click_element(pymordial_element, times=DEFAULT_CLICK_TIMES, screenshot_img_bytes=None, max_tries=DEFAULT_MAX_TRIES)

Clicks a UI element on the screen.

Parameters:

Name	Type	Description	Default
pymordial_element	PymordialElement	The element to click.	required
times	int	Optional number of times to click. Defaults to DEFAULT_CLICK_TIMES config.	DEFAULT_CLICK_TIMES
screenshot_img_bytes	bytes | None	Optional pre-captured screenshot to look for the element in. Defaults to None.	None
max_tries	int	Optional maximum number of retries to find the element. Defaults to DEFAULT_MAX_TRIES config.	DEFAULT_MAX_TRIES

Returns:

Type	Description
bool	True if the element was found and clicked, False otherwise.

Source code in src/pymordialblue/bluestacks_controller.py

def click_element(
    self,
    pymordial_element: PymordialElement,
    times: int = DEFAULT_CLICK_TIMES,
    screenshot_img_bytes: bytes | None = None,
    max_tries: int = DEFAULT_MAX_TRIES,
) -> bool:
    """Clicks a UI element on the screen.

    Args:
        pymordial_element: The element to click.
        times: Optional number of times to click. Defaults to DEFAULT_CLICK_TIMES config.
        screenshot_img_bytes: Optional pre-captured screenshot to look for the element in. Defaults to None.
        max_tries: Optional maximum number of retries to find the element. Defaults to DEFAULT_MAX_TRIES config.

    Returns:
        True if the element was found and clicked, False otherwise.
    """
    # Ensure Bluestacks is ready before trying to click ui
    match self.bluestacks.state.current_state:
        case EmulatorState.CLOSED | EmulatorState.LOADING:
            logger.warning("Cannot click coords - Bluestacks is not ready")
            return False
        case EmulatorState.READY:
            if not self.adb.is_connected():
                self.adb.connect()
                if not self.adb.is_connected():
                    logger.warning(
                        "ADB device not connected. Skipping 'click_element' method call."
                    )
                    return False
            coord: tuple[int, int] | None = self.find_element(
                pymordial_element=pymordial_element,
                pymordial_screenshot=screenshot_img_bytes,
                max_tries=max_tries,
            )
            if not coord:
                logger.debug(f"UI element {pymordial_element.label} not found")
                return False
            if self.click_coord(coord, times=times):
                logger.debug(
                    f"Click event sent via ADB at coords x={coord[0]}, y={coord[1]}"
                )
                return True
            return False
        case _:
            logger.warning(
                "Cannot click coords - Bluestacks state is not in a valid state."
                " Make sure it is in the 'EmulatorState.READY' state."
            )
            return False

click_elements(pymordial_elements, screenshot_img_bytes=None, max_tries=DEFAULT_MAX_TRIES)

Clicks any of the elements in the list.

Parameters:

Name	Type	Description	Default
pymordial_elements	list[PymordialElement]	List of elements to try clicking.	required
screenshot_img_bytes	bytes | None	Optional pre-captured screenshot.	None
max_tries	int	Maximum number of retries per element.	DEFAULT_MAX_TRIES

Returns:

Type	Description
bool	True if any element was clicked, False otherwise.

Source code in src/pymordialblue/bluestacks_controller.py

def click_elements(
    self,
    pymordial_elements: list[PymordialElement],
    screenshot_img_bytes: bytes | None = None,
    max_tries: int = DEFAULT_MAX_TRIES,
) -> bool:
    """Clicks any of the elements in the list.

    Args:
        pymordial_elements: List of elements to try clicking.
        screenshot_img_bytes: Optional pre-captured screenshot.
        max_tries: Maximum number of retries per element.

    Returns:
        True if any element was clicked, False otherwise.
    """
    return any(
        self.click_element(
            pymordial_element=pymordial_element,
            screenshot_img_bytes=screenshot_img_bytes,
            max_tries=max_tries,
        )
        for pymordial_element in pymordial_elements
    )

go_home()

Navigate to Android home screen.

Convenience method that delegates to adb.go_home().

Source code in src/pymordialblue/bluestacks_controller.py

def go_home(self) -> None:
    """Navigate to Android home screen.

    Convenience method that delegates to adb.go_home().
    """
    self.adb.go_home()

go_back()

Press Android back button.

Convenience method that delegates to adb.go_back().

Source code in src/pymordialblue/bluestacks_controller.py

def go_back(self) -> None:
    """Press Android back button.

    Convenience method that delegates to adb.go_back().
    """
    self.adb.go_back()

tap(x, y)

Tap at specific coordinates.

Parameters:

Name	Type	Description	Default
x	int	X coordinate.	required
y	int	Y coordinate.	required

Convenience method that delegates to adb.tap().

Source code in src/pymordialblue/bluestacks_controller.py

def tap(self, x: int, y: int) -> None:
    """Tap at specific coordinates.

    Args:
        x: X coordinate.
        y: Y coordinate.

    Convenience method that delegates to adb.tap().
    """
    return self.adb.tap(x, y)

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

Perform swipe gesture.

Parameters:

Name	Type	Description	Default
start_x	int	Starting X coordinate.	required
start_y	int	Starting Y coordinate.	required
end_x	int	Ending X coordinate.	required
end_y	int	Ending Y coordinate.	required
duration	int	Swipe duration in milliseconds.	300

Convenience method that delegates to adb.swipe().

Source code in src/pymordialblue/bluestacks_controller.py

def swipe(
    self, start_x: int, start_y: int, end_x: int, end_y: int, duration: int = 300
) -> None:
    """Perform swipe gesture.

    Args:
        start_x: Starting X coordinate.
        start_y: Starting Y coordinate.
        end_x: Ending X coordinate.
        end_y: Ending Y coordinate.
        duration: Swipe duration in milliseconds.

    Convenience method that delegates to adb.swipe().
    """
    return self.adb.swipe(start_x, start_y, end_x, end_y, duration)

find_element(pymordial_element, pymordial_screenshot=None, max_tries=DEFAULT_MAX_TRIES)

Finds the coordinates of a UI element on the screen.

Parameters:

Name	Type	Description	Default
pymordial_element	PymordialElement	The element to find.	required
pymordial_screenshot	bytes | None	Optional pre-captured screenshot.	None
max_tries	int	Maximum number of retries.	DEFAULT_MAX_TRIES

Returns:

Type	Description
tuple[int, int] | None	(x, y) coordinates if found, None otherwise.

Source code in src/pymordialblue/bluestacks_controller.py

def find_element(
    self,
    pymordial_element: PymordialElement,
    pymordial_screenshot: bytes | None = None,
    max_tries: int = DEFAULT_MAX_TRIES,
) -> tuple[int, int] | None:
    """Finds the coordinates of a UI element on the screen.

    Args:
        pymordial_element: The element to find.
        pymordial_screenshot: Optional pre-captured screenshot.
        max_tries: Maximum number of retries.

    Returns:
        (x, y) coordinates if found, None otherwise.
    """
    if isinstance(pymordial_element, PymordialImage):
        return self.ui.where_element(
            pymordial_element=pymordial_element,
            pymordial_screenshot=pymordial_screenshot,
            max_tries=max_tries,
        )
    elif isinstance(pymordial_element, PymordialText):
        return self.ui.find_text(
            text_to_find=pymordial_element.element_text,
            pymordial_screenshot=pymordial_screenshot,
            strategy=pymordial_element.extract_strategy,
        )
    elif isinstance(pymordial_element, PymordialPixel):
        # Capture screenshot if not provided (avoid 'or' with numpy arrays)
        pixel_screenshot = (
            pymordial_screenshot
            if pymordial_screenshot is not None
            else self.capture_screen()
        )
        is_match = self.ui.check_pixel_color(
            pymordial_pixel=pymordial_element,
            pymordial_screenshot=pixel_screenshot,
        )
        return pymordial_element.position if is_match else None

    raise NotImplementedError(
        f"find_element() not implemented for this element type: {type(pymordial_element)}"
    )

is_element_visible(pymordial_element, pymordial_screenshot=None, max_tries=None)

Checks if a UI element is visible on the screen.

Parameters:

Name	Type	Description	Default
pymordial_element	PymordialElement	The element to check for.	required
pymordial_screenshot	bytes | None	Optional pre-captured screenshot.	None
max_tries	int | None	Optional maximum number of retries.	None

Returns:

Type	Description
bool	True if the element is found, False otherwise.

Source code in src/pymordialblue/bluestacks_controller.py

def is_element_visible(
    self,
    pymordial_element: PymordialElement,
    pymordial_screenshot: bytes | None = None,
    max_tries: int | None = None,
) -> bool:
    """Checks if a UI element is visible on the screen.

    Args:
        pymordial_element: The element to check for.
        pymordial_screenshot: Optional pre-captured screenshot.
        max_tries: Optional maximum number of retries.

    Returns:
        True if the element is found, False otherwise.
    """
    if not isinstance(pymordial_element, PymordialElement):
        raise TypeError(
            f"pymordial_element must be an instance of PymordialElement, not {type(pymordial_element)}"
        )

    if isinstance(pymordial_element, PymordialImage):
        return (
            self.find_element(
                pymordial_element=pymordial_element,
                pymordial_screenshot=pymordial_screenshot,
                max_tries=max_tries or self.DEFAULT_MAX_TRIES,
            )
            is not None
        )
    elif isinstance(pymordial_element, PymordialText):
        # For text, we use the text controller to check existence
        # Note: This doesn't return coordinates yet, so click_element won't work for Text
        # unless find_element is implemented for Text.

        # If the element has a defined region, crop the image to that region
        if pymordial_element.region and pymordial_screenshot is not None:
            try:
                if isinstance(pymordial_screenshot, bytes):
                    pymordial_screenshot = Image.open(BytesIO(pymordial_screenshot))
                elif isinstance(pymordial_screenshot, np.ndarray):
                    pymordial_screenshot = Image.fromarray(pymordial_screenshot)
                else:
                    pymordial_screenshot = None

                if pymordial_screenshot is not None:
                    # region is (left, top, right, bottom)
                    pymordial_screenshot = pymordial_screenshot.crop(
                        pymordial_element.region
                    )
                    pymordial_screenshot = np.array(pymordial_screenshot)
            except Exception as e:
                logger.warning(f"Failed to crop image for text detection: {e}")

        return self.ui.check_text(
            text_to_find=pymordial_element.element_text,
            pymordial_screenshot=pymordial_screenshot,
            strategy=pymordial_element.extract_strategy,
            case_sensitive=False,
        )
    elif isinstance(pymordial_element, PymordialPixel):
        return (
            self.find_element(
                pymordial_element=pymordial_element,
                pymordial_screenshot=pymordial_screenshot,
                max_tries=max_tries or self.DEFAULT_MAX_TRIES,
            )
            is not None
        )
    else:
        raise NotImplementedError(
            f"is_element_visible not implemented for {type(pymordial_element)}"
        )

press_enter()

Press the Enter key.

Convenience method that delegates to adb.press_enter().

Source code in src/pymordialblue/bluestacks_controller.py

def press_enter(self) -> None:
    """Press the Enter key.

    Convenience method that delegates to adb.press_enter().
    """
    return self.adb.press_enter()

press_esc()

Press the Esc key.

Convenience method that delegates to adb.press_esc().

Source code in src/pymordialblue/bluestacks_controller.py

def press_esc(self) -> None:
    """Press the Esc key.

    Convenience method that delegates to adb.press_esc().
    """
    return self.adb.press_esc()

type_text(text, enter=False)

Send text input to the device.

Parameters:

Name	Type	Description	Default
text	str	Text to send.	required
enter	bool	Whether to press enter after typing.	False

Convenience method that delegates to adb.type_text().

Source code in src/pymordialblue/bluestacks_controller.py

def type_text(self, text: str, enter: bool = False) -> None:
    """Send text input to the device.

    Args:
        text: Text to send.
        enter: Whether to press enter after typing.

    Convenience method that delegates to adb.type_text().
    """
    return self.adb.type_text(text, enter)

run_command(command)

Execute ADB shell command.

Parameters:

Name	Type	Description	Default
command	str	Shell command to execute.	required

Returns:

Type	Description
bytes | None	Command output as bytes, or None if failed.

Convenience method that delegates to adb.run_command().

Source code in src/pymordialblue/bluestacks_controller.py

def run_command(self, command: str) -> bytes | None:
    """Execute ADB shell command.

    Args:
        command: Shell command to execute.

    Returns:
        Command output as bytes, or None if failed.

    Convenience method that delegates to adb.run_command().
    """
    return self.adb.run_command(command)

get_current_app()

Get the currently running app's package name.

Returns:

Type	Description
str | None	Package name of current app, or None if failed.

Convenience method that delegates to adb.get_current_app().

Source code in src/pymordialblue/bluestacks_controller.py

def get_current_app(self) -> str | None:
    """Get the currently running app's package name.

    Returns:
        Package name of current app, or None if failed.

    Convenience method that delegates to adb.get_current_app().
    """
    return self.adb.get_current_app()

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

Read text from an image using OCR.

Parameters:

Name	Type	Description	Default
image_path	Path | bytes | str	Path to image file, image bytes, or string path.	required
strategy	PymordialExtractStrategy | None	Optional preprocessing strategy.	None

Returns:

Type	Description
list[str]	List of detected text lines.

Convenience method that delegates to text.read_text().

Source code in src/pymordialblue/bluestacks_controller.py

def read_text(
    self,
    image_path: "Path | bytes | str",
    case_sensitive: bool = False,
    strategy: "PymordialExtractStrategy | None" = None,
) -> list[str]:
    """Read text from an image using OCR.

    Args:
        image_path: Path to image file, image bytes, or string path.
        strategy: Optional preprocessing strategy.

    Returns:
        List of detected text lines.

    Convenience method that delegates to text.read_text().
    """
    return self.ui.read_text(image_path, case_sensitive, strategy)

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

Check if specific text exists in an image.

Parameters:

Name	Type	Description	Default
text_to_find	str	Text to search for.	required
image_path	Path | bytes | str	Image to search in.	required
case_sensitive	bool	Whether search is case-sensitive.	False
strategy	PymordialExtractStrategy | None	Optional preprocessing strategy.	None

Returns:

Type	Description
bool	True if text found, False otherwise.

Convenience method that delegates to text.check_text().

Source code in src/pymordialblue/bluestacks_controller.py

def check_text(
    self,
    text_to_find: str,
    image_path: "Path | bytes | str",
    case_sensitive: bool = False,
    strategy: "PymordialExtractStrategy | None" = None,
) -> bool:
    """Check if specific text exists in an image.

    Args:
        text_to_find: Text to search for.
        image_path: Image to search in.
        case_sensitive: Whether search is case-sensitive.
        strategy: Optional preprocessing strategy.

    Returns:
        True if text found, False otherwise.

    Convenience method that delegates to text.check_text().
    """
    return self.ui.check_text(text_to_find, image_path, case_sensitive, strategy)

is_bluestacks_ready()

Check if BlueStacks is in READY state.

Returns:

Type	Description
bool	True if BlueStacks is ready, False otherwise.

Convenience method that delegates to self.bluestacks.is_ready().

Source code in src/pymordialblue/bluestacks_controller.py

def is_bluestacks_ready(self) -> bool:
    """Check if BlueStacks is in READY state.

    Returns:
        True if BlueStacks is ready, False otherwise.

    Convenience method that delegates to self.bluestacks.is_ready().
    """
    return self.bluestacks.is_ready()

is_bluestacks_loading()

Check if BlueStacks is currently loading.

Returns:

Type	Description
bool	True if BlueStacks is loading, False otherwise.

Convenience method that delegates to bluestacks.is_loading().

Source code in src/pymordialblue/bluestacks_controller.py

def is_bluestacks_loading(self) -> bool:
    """Check if BlueStacks is currently loading.

    Returns:
        True if BlueStacks is loading, False otherwise.

    Convenience method that delegates to bluestacks.is_loading().
    """
    return self.bluestacks.is_loading()

start_streaming()

Starts the screen stream using ADB device.

Returns:

Type	Description
bool	True if streaming started successfully, False otherwise.

Convenience method that delegates to adb.start_stream().

Source code in src/pymordialblue/bluestacks_controller.py

def start_streaming(self) -> bool:
    """Starts the screen stream using ADB device.

    Returns:
        True if streaming started successfully, False otherwise.

    Convenience method that delegates to adb.start_stream().
    """
    result = self.adb.start_stream()
    if result:
        self._streaming_enabled = True
    return result

get_frame()

Get the latest frame from the active stream.

Returns:

Type	Description
ndarray | None	Latest frame as numpy array (RGB), or None if unavailable.

Convenience method that delegates to adb.get_latest_frame().

Example

frame = controller.get_frame() if frame is not None: ... # Process frame (OCR, template matching, etc.) ... text = controller.read_text(frame)

Source code in src/pymordialblue/bluestacks_controller.py

def get_frame(self) -> "np.ndarray | None":
    """Get the latest frame from the active stream.

    Returns:
        Latest frame as numpy array (RGB), or None if unavailable.

    Convenience method that delegates to adb.get_latest_frame().

    Example:
        >>> frame = controller.get_frame()
        >>> if frame is not None:
        ...     # Process frame (OCR, template matching, etc.)
        ...     text = controller.read_text(frame)
    """
    return self.adb.get_latest_frame()

stop_streaming()

Stop the active video stream and disable auto-restart.

Convenience method that delegates to adb.stop_stream().

Source code in src/pymordialblue/bluestacks_controller.py

def stop_streaming(self) -> None:
    """Stop the active video stream and disable auto-restart.

    Convenience method that delegates to adb.stop_stream().
    """
    self._streaming_enabled = False
    return self.adb.stop_stream()

__repr__()

Returns a string representation of the PymordialController.

Source code in src/pymordialblue/bluestacks_controller.py

def __repr__(self) -> str:
    """Returns a string representation of the PymordialController."""
    return (
        f"PymordialController("
        f"apps={len(self._apps)}, "
        f"adb_connected={self.adb.is_connected()}, "
        f"bluestacks={self.bluestacks.state.current_state.name})"
    )