Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 7 additions & 2 deletions .github/workflows/ci.yml
Original file line number Diff line number Diff line change
Expand Up @@ -48,7 +48,7 @@ jobs:
- 'Dockerfile'
- '.github/workflows/ci.yml'

lint:
format_and_lint:
timeout-minutes: 3
runs-on: ubuntu-latest

Expand Down Expand Up @@ -95,7 +95,7 @@ jobs:
- name: Install dependencies
run: |
source .venv/bin/activate
poetry install --no-root --with format
poetry install --no-root --with format,lint

- name: Check for successful installation
run: |
Expand All @@ -106,3 +106,8 @@ jobs:
run: |
source .venv/bin/activate
poetry run python tasks/format.py --dry-run

- name: Format and lint project
run: |
source .venv/bin/activate
poetry run mypy
6 changes: 6 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -129,3 +129,9 @@ poetry run pytest -rfsxE --capture=no --log-cli-level=DEBUG --maxfail=1 -vv ./te
```bash
poetry run python ./tasks/format.py
```

#### Code Linting

```bash
poetry run mypy
```
229 changes: 226 additions & 3 deletions poetry.lock

Large diffs are not rendered by default.

12 changes: 12 additions & 0 deletions pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -24,13 +24,25 @@ test = [
format = [
"ruff (>=0.15.7,<0.16.0)"
]
lint = [
"mypy (>=1.20.1,<2.0.0)",
"types-pynput (>=1.8.1.20260408,<2.0.0.0)",
"types-psutil (>=7.2.2.20260408,<8.0.0.0)"
]


[tool.poetry]
packages = [
{ include = "visiongui", from = "src" }
]

[tool.mypy]
files = ["src"]
mypy_path = 'src'
explicit_package_bases = true
strict = true
exclude = '(\.venv|build|dist)'

[build-system]
requires = ["poetry-core>=2.0.0,<3.0.0"]
build-backend = "poetry.core.masonry.api"
Empty file added src/visiongui/__init__.py
Empty file.
21 changes: 11 additions & 10 deletions src/visiongui/driver/DesktopDriverInterface.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
import re
import subprocess
from abc import ABC, abstractmethod

Expand All @@ -11,11 +12,11 @@
class DesktopDriverInterface(ABC):
@property
@abstractmethod
def process(self) -> subprocess.Popen | None: ...
def process(self) -> subprocess.Popen[bytes] | None: ...

@process.setter
@abstractmethod
def process(self, value: subprocess.Popen | None) -> None: ...
def process(self, value: subprocess.Popen[bytes] | None) -> None: ...

@property
@abstractmethod
Expand All @@ -26,22 +27,22 @@ def window(self) -> pywinctl.Window | None: ...
def window(self, value: pywinctl.Window | None) -> None: ...

@abstractmethod
def launch_process(self, *, cmd: list[str]) -> subprocess.Popen: ...
def launch_process(self, *, cmd: list[str]) -> subprocess.Popen[bytes]: ...

@abstractmethod
def find_window(
self,
*,
title,
timeout: float,
title: re.Pattern[str],
timeout: int,
) -> pywinctl.Window: ...

@abstractmethod
def wait_for_window_to_disappear(
self,
*,
title,
timeout: float,
title: re.Pattern[str],
timeout: int,
) -> None: ...

@abstractmethod
Expand All @@ -59,10 +60,10 @@ def find_element_by_image(
self,
*,
image_path: str,
timeout: float,
timeout: int,
log_image_name: str,
margin_of_error: float,
time_held_stable_on_screen: float,
margin_of_error: int,
time_held_stable_on_screen: int,
debug_output_base_path: str,
match_with_color: bool = False,
) -> DesktopElementInterface: ...
25 changes: 13 additions & 12 deletions src/visiongui/driver/DesktopDriverWindowsImplementation.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
import re
import subprocess

import pywinctl
Expand All @@ -19,16 +20,16 @@


class DesktopDriverWindowsImplementation(DesktopDriverInterface):
def __init__(self):
self._process: subprocess.Popen | None = None
def __init__(self) -> None:
self._process: subprocess.Popen[bytes] | None = None
self._window: pywinctl.Window | None = None

@property
def process(self) -> subprocess.Popen | None:
def process(self) -> subprocess.Popen[bytes] | None:
return self._process

@process.setter
def process(self, value: subprocess.Popen | None) -> None:
def process(self, value: subprocess.Popen[bytes] | None) -> None:
self._process = value

@property
Expand All @@ -39,15 +40,15 @@ def window(self) -> pywinctl.Window | None:
def window(self, value: pywinctl.Window | None) -> None:
self._window = value

def launch_process(self, *, cmd: list[str]) -> subprocess.Popen:
def launch_process(self, *, cmd: list[str]) -> subprocess.Popen[bytes]:
self.process = launch_process(cmd=cmd)
return self.process

def find_window(
self,
*,
title,
timeout: float,
title: re.Pattern[str],
timeout: int,
) -> pywinctl.Window:
return find_window(
title=title,
Expand All @@ -57,8 +58,8 @@ def find_window(
def wait_for_window_to_disappear(
self,
*,
title,
timeout: float,
title: re.Pattern[str],
timeout: int,
) -> None:
return wait_for_window_to_disappear(
title=title,
Expand All @@ -69,10 +70,10 @@ def find_element_by_image(
self,
*,
image_path: str,
timeout: float,
timeout: int,
log_image_name: str,
margin_of_error: float,
time_held_stable_on_screen: float,
margin_of_error: int,
time_held_stable_on_screen: int,
debug_output_base_path: str,
match_with_color: bool = False,
) -> DesktopElementInterface:
Expand Down
6 changes: 5 additions & 1 deletion src/visiongui/driver/close.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,11 @@ def close(
if not isinstance(driver, DesktopDriverInterface):
raise TypeError("Expected driver to be an instance of DesktopDriver")

pid = driver.window.getPID()
window = driver.window
pid = None
if window is not None:
pid = window.getPID()

logger.debug(f"Forcefully killing process owning the window: {pid}")
proc = psutil.Process(pid)
proc.terminate()
Expand Down
36 changes: 26 additions & 10 deletions src/visiongui/driver/find_element_by_image.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
import logging
import os
import time
from typing import Callable

import cv2
import numpy as np
Expand All @@ -23,18 +24,22 @@
logger = logging.getLogger(__name__)


def _is_stable(get_current_location, timeout, time_held_stable_on_screen):
def _is_stable(
get_current_location: Callable[[], DesktopElementImplementation | bool],
timeout: int,
time_held_stable_on_screen: int,
) -> DesktopElementImplementation:
start_time = time.time()
last_location = None
last_location: DesktopElementImplementation | None = None
stable_start = None
found_once = False

while time.time() - start_time < timeout:
current_location = get_current_location()
if current_location:
if isinstance(current_location, DesktopElementImplementation):
found_once = True
if (
last_location
last_location is not None
and current_location.top_left == last_location.top_left
and current_location.bottom_right == last_location.bottom_right
):
Expand All @@ -60,7 +65,13 @@ def _is_stable(get_current_location, timeout, time_held_stable_on_screen):
raise ExceptionElementNotFound(timeout=timeout)


def _match_template(template, margin_of_error, monitor, screen_image, mask=None):
def _match_template(
template: np.ndarray,
margin_of_error: float,
monitor: dict[str, int],
screen_image: np.ndarray,
mask: np.ndarray | None = None,
) -> DesktopElementImplementation | bool:
# Ensure screen image has same number of channels as template
if len(template.shape) == 2:
if len(screen_image.shape) == 3:
Expand Down Expand Up @@ -109,18 +120,21 @@ def _match_template(template, margin_of_error, monitor, screen_image, mask=None)

def find_element_by_image(
image_path: str,
timeout: float,
timeout: int,
log_image_name: str,
debug_output_base_path: str,
margin_of_error: float,
time_held_stable_on_screen: float,
time_held_stable_on_screen: int,
match_with_color: bool = False,
) -> DesktopElementImplementation:
if not image_path or not os.path.isfile(image_path):
raise FileNotFoundError(f"Template image not found: {image_path}")

# [68566fb0-e936-48e3-8c87-c5b8735567df] If the image has an alpha channel, extract it to build a binary mask. This mask ensures that only opaque regions of the template are matched, ignoring transparent padding.
image = cv2.imread(image_path, cv2.IMREAD_UNCHANGED)
if image is None:
raise ValueError(f"Unable to read image at {image_path}")

mask = None
template = None
if len(image.shape) == 3 and image.shape[2] == 4:
Expand Down Expand Up @@ -154,7 +168,7 @@ def find_element_by_image(
with mss() as sct:
monitor = sct.monitors[1]

def get_current_location():
def get_current_location() -> DesktopElementImplementation | bool:
screenshot = sct.grab(monitor)
screen_image = np.array(screenshot)
return _match_template(
Expand All @@ -173,10 +187,12 @@ def get_current_location():
)
return result
try:
result = WebDriverWait(timeout).until(
wait_result = WebDriverWait(timeout).until(
condition=get_current_location,
)
return result
if not isinstance(wait_result, DesktopElementImplementation):
raise ExceptionElementNotFound(timeout=timeout)
return wait_result
except ExceptionTimeout:
save_debug_screenshot(
image_file_name_prefix=FileNamePrefix.FAIL,
Expand Down
8 changes: 4 additions & 4 deletions src/visiongui/driver/find_window.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,8 +10,8 @@
logger = logging.getLogger(__name__)


def _get_matching_window(title: re.Pattern) -> pywinctl.Window | None:
all_windows = pywinctl.getAllWindows()
def _get_matching_window(title: re.Pattern[str]) -> pywinctl.Window | None:
all_windows = pywinctl.getAllWindows() # type: ignore[no-untyped-call]
titles = [window.title for window in all_windows if window.title.strip()]
logger.debug(f"Checking all window titles: {titles}")

Expand All @@ -25,8 +25,8 @@ def _get_matching_window(title: re.Pattern) -> pywinctl.Window | None:


def find_window(
title: re.Pattern,
timeout: float,
title: re.Pattern[str],
timeout: int,
) -> pywinctl.Window:
def _window_check() -> pywinctl.Window | None:
return _get_matching_window(title)
Expand Down
2 changes: 1 addition & 1 deletion src/visiongui/driver/launch_process.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,7 @@
import subprocess


def launch_process(cmd: list[str]) -> subprocess.Popen:
def launch_process(cmd: list[str]) -> subprocess.Popen[bytes]:
if not cmd:
raise ValueError("Command list must not be empty")

Expand Down
2 changes: 1 addition & 1 deletion src/visiongui/driver/save_debug_screenshot.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,7 @@ def save_debug_screenshot(
image_file_name_prefix: FileNamePrefix,
log_image_name: str,
debug_output_base_path: str,
):
) -> None:
file_binary = take_screenshot()
file_name = f"{image_file_name_prefix.value}_{int(time.time())}_{os.path.basename(log_image_name)}.png"
save_file(
Expand Down
7 changes: 5 additions & 2 deletions src/visiongui/driver/switch_to.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,8 +4,11 @@


def _set_foreground_hwnd(hwnd: int) -> None:
user32 = ctypes.WinDLL("user32", use_last_error=True)
kernel32 = ctypes.WinDLL("kernel32", use_last_error=True)
if platform.system() != "Windows":
raise RuntimeError("This function is only supported on Windows")

user32 = ctypes.WinDLL("user32", use_last_error=True) # type: ignore[attr-defined]
kernel32 = ctypes.WinDLL("kernel32", use_last_error=True) # type: ignore[attr-defined]
GetForegroundWindow = user32.GetForegroundWindow
GetWindowThreadProcessId = user32.GetWindowThreadProcessId
GetCurrentThreadId = kernel32.GetCurrentThreadId
Expand Down
9 changes: 6 additions & 3 deletions src/visiongui/driver/wait.py
Original file line number Diff line number Diff line change
@@ -1,18 +1,21 @@
import time
from collections.abc import Callable
from typing import TypeVar

from visiongui.driver.exception import ExceptionTimeout

T = TypeVar("T")


class WebDriverWait:
def __init__(self, timeout: float, poll_frequency: float = 0.1):
def __init__(self, timeout: int, poll_frequency: float = 0.1):
self.timeout = timeout
self.poll_frequency = poll_frequency

def until(
self,
condition: Callable[[], object],
):
condition: Callable[[], T],
) -> T:
end_time = time.time() + self.timeout
while True:
try:
Expand Down
4 changes: 2 additions & 2 deletions src/visiongui/driver/wait_for_window_to_disappear.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,8 +11,8 @@


def wait_for_window_to_disappear(
title: re.Pattern,
timeout: float,
title: re.Pattern[str],
timeout: int,
) -> None:
def _window_check() -> bool:
return _get_matching_window(title) is None
Expand Down
Loading