diff --git a/hippolyzer/apps/proxy_leapreceiver.py b/hippolyzer/apps/proxy_leapreceiver.py index 62da10d..548e7fb 100644 --- a/hippolyzer/apps/proxy_leapreceiver.py +++ b/hippolyzer/apps/proxy_leapreceiver.py @@ -9,7 +9,7 @@ import logging import multiprocessing import pprint -from hippolyzer.lib.proxy.leap import LEAPBridgeServer, LEAPClient +from hippolyzer.lib.proxy.leap import LEAPBridgeServer, LEAPClient, LLWindowWrapper, UIPath, LLUIWrapper async def client_connected(client: LEAPClient): @@ -53,6 +53,24 @@ async def client_connected(client: LEAPClient): # `command()` except it internally addresses whatever the system command pump is. await client.sys_command("ping") + # Spawn the test textbox floater + client.void_command("LLFloaterReg", "showInstance", {"name": "test_textbox"}) + + window_api = LLWindowWrapper(client) + + # LEAP allows addressing UI elements by "path". We expose that through a pathlib-like interface + # to allow composing UI element paths. + textbox_path = UIPath.for_floater("floater_test_textbox") / "long_text_editor" + # Click the "long_text_editor" in the test textbox floater. + await window_api.mouse_click(button="LEFT", path=textbox_path) + + # Type some text in it + window_api.text_input("Also I can type in here pretty good.") + + # Print out the value of the textbox we just typed in + ui_api = LLUIWrapper(client) + pprint.pprint(await ui_api.get_value(textbox_path)) + def receiver_main(): logging.basicConfig(level=logging.INFO) diff --git a/hippolyzer/lib/proxy/leap.py b/hippolyzer/lib/proxy/leap.py index 22585f4..344a2c5 100644 --- a/hippolyzer/lib/proxy/leap.py +++ b/hippolyzer/lib/proxy/leap.py @@ -6,14 +6,16 @@ TODO: split this out into its own package from __future__ import annotations +from typing import * + +import abc +import asyncio import collections import contextlib import dataclasses -from typing import * - -import asyncio import enum import logging +import pathlib import uuid import weakref @@ -147,6 +149,8 @@ class LEAPClient: # The future will be cleaned up when the Future is done. fut.add_done_callback(self._cleanup_request_future) self._reply_futs[req_id] = fut + elif expect_reply: + raise ValueError(f"Must send a dict in `data` if you want a reply, you sent {data!r}") self._write_message(pump, data) return fut @@ -281,6 +285,162 @@ class LEAPClient: return +class LEAPAPIWrapper(abc.ABC): + """Base class for classes wrapping specific LEAP APIs""" + PUMP_NAME: Optional[str] = None + + def __init__(self, client: LEAPClient, pump_name: Optional[str] = None): + super().__init__() + self._client = client + self._pump_name = pump_name or self.PUMP_NAME + assert self._pump_name + + +class UIPath(pathlib.PurePosixPath): + __slots__ = [] + + @classmethod + def for_floater(cls, floater_name: str) -> UIPath: + return cls("/main_view/menu_stack/world_panel/Floater View") / floater_name + + +UI_PATH_TYPE = Optional[Union[str, UIPath]] + + +class LLWindowWrapper(LEAPAPIWrapper): + PUMP_NAME = "LLWindow" + + MASK_TYPE = Optional[Collection[str]] + KEYCODE_TYPE = Optional[int] + KEYSYM_TYPE = Optional[str] + CHAR_TYPE = Optional[str] + MOUSE_COORD_TYPE: Optional[int] + + def _convert_key_payload(self, /, *, keycode: KEYCODE_TYPE, keysym: KEYSYM_TYPE, char: CHAR_TYPE, + mask: MASK_TYPE, path: UI_PATH_TYPE) -> Dict: + if keycode is not None: + payload = {"keycode": keycode} + elif keysym is not None: + payload = {"keysym": keysym} + elif char is not None: + payload = {"char": char} + else: + raise ValueError("Didn't have one of keycode, keysym or char") + + if path: + payload["path"] = str(path) + if mask: + payload["mask"] = mask + + return payload + + def key_down(self, /, *, mask: MASK_TYPE = None, keycode: KEYCODE_TYPE = None, + keysym: KEYSYM_TYPE = None, char: CHAR_TYPE = None, path: UI_PATH_TYPE = None): + """Simulate a key being pressed down""" + payload = self._convert_key_payload(keysym=keysym, keycode=keycode, char=char, mask=mask, path=path) + self._client.void_command(self._pump_name, "keyDown", payload) + + def key_up(self, /, *, mask: MASK_TYPE = None, keycode: KEYCODE_TYPE = None, + keysym: KEYSYM_TYPE = None, char: CHAR_TYPE = None, path: UI_PATH_TYPE = None): + """Simulate a key being released""" + payload = self._convert_key_payload(keysym=keysym, keycode=keycode, char=char, mask=mask, path=path) + self._client.void_command(self._pump_name, "keyUp", payload) + + def key_press(self, /, *, mask: MASK_TYPE = None, keycode: KEYCODE_TYPE = None, + keysym: KEYSYM_TYPE = None, char: CHAR_TYPE = None, path: UI_PATH_TYPE = None): + """Simulate a key being pressed down and immediately released""" + self.key_down(mask=mask, keycode=keycode, keysym=keysym, char=char, path=path) + self.key_up(mask=mask, keycode=keycode, keysym=keysym, char=char, path=path) + + def text_input(self, text_input: str, path: Optional[str] = None): + """Simulate a user typing a string of text""" + # TODO: Uhhhhh I can't see how the key* APIs could possibly handle i18n correctly, + # what with all the U8s. Maybe I'm just dumb? + for char in text_input: + self.key_press(char=char, path=path) + + async def get_paths(self, under: str = "") -> Dict: + """Get all UI paths under the root, or under a path if specified""" + return await self._client.command(self._pump_name, "getPaths", {"under": under}) + + async def get_info(self, path: str) -> Dict: + """Get info about an element specified by path""" + return await self._client.command(self._pump_name, "getInfo", {"path": path}) + + def _build_mouse_payload(self, /, *, x: MOUSE_COORD_TYPE, y: MOUSE_COORD_TYPE, path: UI_PATH_TYPE, + mask: MASK_TYPE, button: str = None) -> Dict: + if path is not None: + payload = {"path": str(path)} + elif x is not None and y is not None: + payload = {"x": x, "y": y} + else: + raise ValueError("Didn't have one of x + y or path") + + if mask: + payload["mask"] = mask + if button: + payload["button"] = button + + return payload + + def mouse_down(self, /, *, x: MOUSE_COORD_TYPE = None, y: MOUSE_COORD_TYPE = None, + path: UI_PATH_TYPE = None, mask: MASK_TYPE = None, button: str) -> Awaitable[Dict]: + """Simulate a mouse down event occurring at a coordinate or UI element path""" + payload = self._build_mouse_payload(x=x, y=y, path=path, mask=mask, button=button) + return self._client.command(self._pump_name, "mouseDown", payload) + + def mouse_up(self, /, *, x: MOUSE_COORD_TYPE = None, y: MOUSE_COORD_TYPE = None, + path: UI_PATH_TYPE = None, mask: MASK_TYPE = None, button: str) -> Awaitable[Dict]: + """Simulate a mouse up event occurring at a coordinate or UI element path""" + payload = self._build_mouse_payload(x=x, y=y, path=path, mask=mask, button=button) + return self._client.command(self._pump_name, "mouseUp", payload) + + def mouse_click(self, /, *, x: MOUSE_COORD_TYPE = None, y: MOUSE_COORD_TYPE = None, + path: UI_PATH_TYPE = None, mask: MASK_TYPE = None, button: str) -> Awaitable[Dict]: + """Simulate a mouse down and immediately following mouse up event""" + # We're going to ignore the mouseDown response, so use void_command instead. + # Most side effects are actually executed on mouseUp. + self._client.void_command(self._pump_name, "mouseDown", + self._build_mouse_payload(x=x, y=y, path=path, mask=mask, button=button)) + return self.mouse_up(x=x, y=y, path=path, mask=mask, button=button) + + def mouse_move(self, /, *, x: MOUSE_COORD_TYPE = None, y: MOUSE_COORD_TYPE = None, + path: UI_PATH_TYPE = None) -> Awaitable[Dict]: + """Move the mouse to the coordinates or path specified""" + payload = self._build_mouse_payload(x=x, y=y, path=path, mask=None) + return self._client.command(self._pump_name, "mouseMove", payload) + + def mouse_scroll(self, clicks: int): + """Act as if the scroll wheel has been moved `clicks` amount. May be negative""" + self._client.command(self._pump_name, "mouseScroll", {"clicks": clicks}) + + +class LLUIWrapper(LEAPAPIWrapper): + PUMP_NAME = "UI" + + def call(self, function: str, parameter: Any = None) -> None: + """ + Invoke the `function` operation as if from a menu or button click, passing `parameter` + + Can call most things registered through `LLUICtrl::CommitCallbackRegistry`. + """ + self._client.void_command(self._pump_name, "call", {"function": function, "parameter": parameter}) + + def get_value(self, path: UI_PATH_TYPE) -> Awaitable[Any]: + """For the UI control identified by `path`, return the current value in `value`""" + + # We want the request to be sent immediately, without requiring `get_value()` to be `await`ed first, + # but that means that we have to return a `Coroutine` that will pull the value out of the dict + # rather than directly returning the `Future`. + # TODO: extract this out into a helper function that makes "unwrapper" Coroutines that wrap a Future. + resp_fut = self._client.command(self._pump_name, "getValue", {"path": str(path)}) + + async def _wrapper(): + return (await resp_fut)['value'] + + return _wrapper() + + class LEAPBridgeServer: """LEAP Bridge TCP server to use with asyncio.start_server()"""