Add API wrappers for LLUI and LLWindow LEAP APIs
This commit is contained in:
Salad Dais
@@ -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)
|
||||
|
||||
@@ -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()"""
|
||||
|
||||
|
||||
Reference in New Issue
Block a user