Mmokko/java inspector backend

robocorp-code/pyproject.toml

@@ -39,6 +39,8 @@ pillow = "^9.0"
 pywin32 = { version = ">=300,<304", platform = "win32", python = "!=3.8.1" }
 psutil = "^5.9.0"
 comtypes = "^1.1"
+java-access-bridge-wrapper = "^1.1.0"
+ruff = "^0.2.1"
 
 [tool.poetry.group.dev.dependencies]
 robocorp-python-ls-core = {path = "../robocorp-python-ls-core/", develop = true}
@@ -52,7 +54,6 @@ robocorp-actions = "^0.0.7" # Just needed for testing.
 
 numpy = "<2"
 black = "^23.1.0"
-ruff = "^0.0.255"
 mypy = "^1.1.1"
 isort = { version = "^5.12.0", python = "^3.8" }
 invoke = "^2.0"
@@ -89,7 +90,7 @@ exclude = "vendored"
 [tool.ruff]
 ignore = [
     "E501", # Line-len.
-    "F541", # f-string without placeholders. 
+    "F541", # f-string without placeholders.
     "E731", # Use 'def' instead of lambda.
 ]
 

robocorp-code/src/robocorp_code/inspector/java/java_inspector.py

@@ -0,0 +1,96 @@
+from typing import List, Optional, TypedDict, cast
+
+from JABWrapper.context_tree import ContextNode
+from JABWrapper.jab_wrapper import JavaWindow
+from robocorp_ls_core.robotframework_log import get_logger
+
+from robocorp_code.inspector.java.robocorp_java._inspector import ColletedTreeTypedDict
+
+log = get_logger(__name__)
+
+JavaWindowInfoTypedDict = TypedDict(
+    "JavaWindowInfoTypedDict",
+    {
+        "pid": int,
+        "hwnd": int,
+        "title": str,
+    },
+)
+
+LocatorNodeInfoTypedDict = TypedDict(
+    "LocatorNoneInfoTypedDict",
+    {
+        "name": str,
+        "role": str,
+        "description": str,
+        "states": str,
+        "indexInParent": int,
+        "childrenCount": int,
+        "x": int,
+        "y": int,
+        "width": int,
+        "height": int,
+        "ancestry": int,
+    },
+)
+
+
+class MatchesAndHierarchyTypedDict(TypedDict):
+    # A list with the nodes matched (these are the ones that the
+    # locator matched)
+    matched_paths: List[str]
+    # This includes all the entries found along with the full hierarchy
+    # to reach the matched entries.
+    hierarchy: List[LocatorNodeInfoTypedDict]
+
+
+def to_window_info(java_window: JavaWindow) -> JavaWindowInfoTypedDict:
+    ret = {}
+    for dct_name in JavaWindowInfoTypedDict.__annotations__:
+        ret[dct_name] = getattr(java_window, dct_name)
+    return cast(JavaWindowInfoTypedDict, ret)
+
+
+def to_locator_info(context_node: ContextNode) -> LocatorNodeInfoTypedDict:
+    ret = {}
+    for dct_name in LocatorNodeInfoTypedDict.__annotations__:
+        if (dct_name) == "ancestry":
+            ret["ancestry"] = getattr(context_node, dct_name)
+        else:
+            ret[dct_name] = getattr(context_node.context_info, dct_name)
+    return cast(LocatorNodeInfoTypedDict, ret)
+
+
+def to_matches_and_hierarchy(
+    matches_and_hierarchy: ColletedTreeTypedDict,
+) -> MatchesAndHierarchyTypedDict:
+    matches = (
+        [str(matches_and_hierarchy["matches"])]
+        if type(matches_and_hierarchy["matches"]) == ContextNode
+        else [str(match) for match in matches_and_hierarchy["matches"]]
+    )
+    hierarchy = [to_locator_info(node) for node in matches_and_hierarchy["tree"]]
+    return {"matches": matches, "hierarchy": hierarchy}
+
+
+class JavaInspector:
+    def __init__(self):
+        from robocorp_code.inspector.java.robocorp_java._inspector import (
+            ElementInspector,
+        )
+
+        self._inspector = ElementInspector()
+
+    def list_windows(self) -> List[JavaWindowInfoTypedDict]:
+        windows = self._inspector.list_windows()
+        return [to_window_info(window) for window in windows]
+
+    def collect_tree(
+        self, window: str, search_depth=1, locator: Optional[str] = None
+    ) -> MatchesAndHierarchyTypedDict:
+        log.info(f"Collect tree from locator: {locator}")
+
+        matches_and_hierarchy = self._inspector.collect_tree(
+            window, search_depth, locator
+        )
+        return to_matches_and_hierarchy(matches_and_hierarchy)

robocorp-code/src/robocorp_code/inspector/java/robocorp_java/_errors.py

@@ -0,0 +1,2 @@
+class NoMatchingLocatorException(Exception):
+    """Match for locator not found."""

robocorp-code/src/robocorp_code/inspector/java/robocorp_java/_event_pump.py

@@ -0,0 +1,63 @@
+import ctypes
+import queue
+import threading
+import time
+
+from JABWrapper.jab_wrapper import JavaAccessBridgeWrapper
+from robocorp_ls_core.robotframework_log import get_logger
+
+PeekMessage = ctypes.windll.user32.PeekMessageW
+GetMessage = ctypes.windll.user32.GetMessageW
+TranslateMessage = ctypes.windll.user32.TranslateMessage
+DispatchMessage = ctypes.windll.user32.DispatchMessageW
+
+log = get_logger(__name__)
+
+
+REMOVE_FROM_QUEUE = 0x0001
+
+
+class EventPumpThread(threading.Thread):
+    def __init__(
+        self,
+    ) -> None:
+        super().__init__()
+        # Jab wrapper needs to be part of the thread that pumps the window events
+        self._jab_wrapper: JavaAccessBridgeWrapper = None
+        self._queue = queue.Queue()
+        self._quit_queue_loop = threading.Event()
+
+    def _pump_background(self) -> bool:
+        try:
+            message = ctypes.byref(ctypes.wintypes.MSG())
+            # Nonblocking API to get windows window events from the queue.
+            # https://learn.microsoft.com/en-us/windows/win32/api/winuser/nf-winuser-peekmessagea
+            # Use non blocing API here so that the thread can quit.
+            if PeekMessage(message, 0, 0, 0, REMOVE_FROM_QUEUE):
+                TranslateMessage(message)
+                log.debug("Dispatching msg={}".format(repr(message)))
+                DispatchMessage(message)
+                return True
+        except Exception as err:
+            log.error(f"Pump error: {err}")
+        finally:
+            log.info("Stopped processing events")
+        return False
+
+    def run(self) -> None:
+        self._jab_wrapper = JavaAccessBridgeWrapper(ignore_callbacks=True)
+        self._queue.put(self._jab_wrapper)
+        while not self._quit_queue_loop.is_set():
+            # The pump is non blocking. If the is no message in the queue
+            # wait for 10 milliseconds until check again to prevent too
+            # fast loop.
+            # TODO: add backoff timer
+            if not self._pump_background():
+                time.sleep(0.01)
+
+    def stop(self):
+        self._quit_queue_loop.set()
+        self._jab_wrapper = None
+
+    def get_wrapper(self) -> JavaAccessBridgeWrapper:
+        return self._queue.get()