import abc
from time import sleep
from fortrace.core.qemu_monitor import QEMUMonitorSession
from fortrace.utility.applications.application import (
ApplicationEvent,
ApplicationType,
GenericApplication,
)
from fortrace.utility.applications.application_factory import get_application
from fortrace.utility.desktop_environments.desktop_environment import DesktopEnvironment
from fortrace.utility.distribution_constants import DesktopEnvironmentType, OSType
from fortrace.utility.exceptions import ConfigurationError, DesktopEnvironmentException
from fortrace.utility.image_processing.image_similarity import (
detect_newly_opened_window,
)
from fortrace.utility.image_processing.text_detection import (
detect_and_recognize_text,
text_line_contains,
)
from fortrace.utility.logger_helper import setup_logger
logger = setup_logger(__name__)
[docs]
class Windows(DesktopEnvironment, abc.ABC):
"""Abstract class to implement general Windows behavior."""
_windows_menu_coordinates: tuple[int, int, int, int] | None
def __init__(
self,
desktop_env: DesktopEnvironmentType,
qemu_monitor_session: QEMUMonitorSession,
):
super().__init__(OSType.WINDOWS, desktop_env, qemu_monitor_session)
self._windows_menu_coordinates = None
[docs]
def open_application(
self, application_type: ApplicationType, application_name: str, **kwargs
) -> GenericApplication:
# TODO: minimize all applications with meta_l-d ?
before = (
self._qs.take_screenshot()
) # take screenshot when application is not opened
if not self._windows_menu_coordinates:
self._qs.send_key_combination("meta_l")
sleep(2)
for _ in range(5):
windows_menu_coordinates = detect_newly_opened_window(
before, self._qs.take_screenshot()
)
if windows_menu_coordinates:
self._windows_menu_coordinates = windows_menu_coordinates
logger.debug(
"Set windows menu coordinates to %s",
self._windows_menu_coordinates,
)
break
sleep(2)
else:
raise DesktopEnvironmentException(
"Cannot determine Windows menu coordinates"
)
else:
self._qs.send_key_combination("meta_l")
for _ in range(5): # detect if Windows menu has loaded
text = detect_and_recognize_text(
self._qs.take_screenshot(), self._windows_menu_coordinates
)[1]
if text_line_contains(
text,
["Alarms & Clocks", "Cortana", "Pinned", "Recommended", "Edge", "All"],
"ignore_case",
):
logger.debug("Windows application window opened")
break
sleep(5)
else:
logger.error("Could not open Windows application window within 25 seconds")
raise DesktopEnvironmentException(
"Windows application window is not responding as expected"
)
self._qs.send_text(application_name)
for _ in range(5):
if text_line_contains(
detected := detect_and_recognize_text(
self._qs.take_screenshot(), self._windows_menu_coordinates
)[1],
["Best match", "Open"],
"ignore_case",
):
break
logger.debug(
"Search for application has not completed yet. Will wait longer."
)
sleep(10)
else:
logger.warning(
"Could not find %s with Windows' search. Maybe a typo? Or the search hang up.",
application_name,
)
logger.warning("Detected strings: %s", detected)
raise DesktopEnvironmentException(
f"Application {application_name} not found!"
)
if kwargs.get("run_as_administrator", False):
logger.info(
"Open application %s with administrative privileges", application_name
)
self._qs.send_key_combination("ctrl-shift-ret")
else:
logger.info("Open application %s", application_name)
self._qs.send_key_combination("ret")
application = get_application(
application_type, application_name, self._qs, self._on_change, **kwargs
)
# maybe there is a window asking to run program with administrative privileges
sleep(
5
) # wait for said window to come up (application can open, does not matter)
self._confirm_uac_window(
application_name, kwargs.get("run_as_administrator", False)
)
for _ in range(5):
coordinates = detect_newly_opened_window(before, self._qs.take_screenshot())
if coordinates is None:
logger.debug(
"Application %s is currently not opened. Will wait.",
application_name,
)
else:
break
sleep(5)
else:
raise DesktopEnvironmentException(
"Failed to open application %s!", application_name
)
application.coordinates = coordinates
return application
[docs]
def login(self, username: str, password: str | None):
if password:
self._qs.send_key_combination("ret")
sleep(4)
# TODO: select user in multiuser system
self._qs.send_text(password, True)
for i in range(5):
text = detect_and_recognize_text(self._qs.take_screenshot())[1]
if text_line_contains(text, "password is incorrect"):
raise ConfigurationError(
f"The provided password '{password}' is not correct"
)
sleep(2)
self._session_unlocked = True
# block until screen is static -> Windows desktop
for _ in range(5):
text = detect_and_recognize_text(self._qs.take_screenshot())[1]
if text_line_contains(text, "welcome"):
sleep(5)
self._qs.mouse.init()
[docs]
def minimize_all_unfocused(self):
"""Minimize all except the active desktop window (restores all windows on second stroke)."""
self._qs.send_key_combination("meta_l-pos1")
[docs]
def focus_application(self, application: GenericApplication):
if application.focused:
return
# traverse through list of opened applications from right to left with alt-esc
index_to_focus = self._applications.index(application)
index_in_focus = self._applications.index(self.active_application)
if index_to_focus < index_in_focus:
times = index_in_focus - index_to_focus
else:
times = index_in_focus + len(self._applications) - index_to_focus
self._qs.send_key_combination("alt-esc", times)
self._on_change(
ApplicationEvent.FOCUS_SHIFTED, application_reference=application
)
sleep(2) # wait for application to be active
[docs]
def run_command(self, command: str, run_as_administrator: bool = False):
"""Opens Windows' run dialog box and executes the provided command.
Is NOT aware of which application is potentially opened
Args:
command: command to be executed
run_as_administrator: should command be executed with administrative privileges
"""
self._qs.send_key_combination("meta_l-r")
sleep(1)
if run_as_administrator:
logger.info("Executing command %s with administrative privileges", command)
self._qs.send_text(command)
self._qs.send_key_combination("ctrl-shift-ret")
else:
logger.info("Executing command %s", command)
self._qs.send_text(command, True)
sleep(5) # wait for application to open/UAC window to come up
self._confirm_uac_window(command, run_as_administrator)
def _confirm_uac_window(self, application_name: str, blocking: bool = False):
"""Confirm the User Account Control window dialogue.
This method scans for the UAC window, and if present, confirms it.
Args:
application_name: name of the application that raises the UAC window
blocking: should this method wait for the window to come up (useful if program is executed with
administrative rights and the window is known to show up)
"""
if blocking:
while not text_line_contains(
text := detect_and_recognize_text(self._qs.take_screenshot())[1],
["User Account Control", "changes to your device?"],
"jaro",
):
sleep(3)
self._qs.send_key_combination("left") # place cursor on 'YES'
self._qs.send_key_combination("ret")
logger.debug(
"Confirmed User Account Control window of %s", application_name
)
else:
if text_line_contains(
detect_and_recognize_text(self._qs.take_screenshot())[1],
["User Account Control", "changes to your device?"],
"jaro",
):
logger.debug(
"User Account Control window of %s detected.", application_name
)
self._qs.send_key_combination("left") # place cursor on 'YES'
self._qs.send_key_combination("ret")
logger.debug(
"Confirmed User Account Control window of %s.", application_name
)
else:
logger.debug("No User Account Window of %s detected.", application_name)
[docs]
def system_power_down(self):
"""Powers down the system via the desktop environment.
This method uses the Windows left-bottom corner menu to power down the system.
"""
self._qs.send_key_combination("meta_l-x")
self._qs.send_key_combination("up", 2)
self._qs.send_key_combination("right")
self._qs.send_key_combination("down")
self._qs.send_key_combination("ret")