Miko/Hippolyzer
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

More work on LEAP API wrappers

Browse Source
This commit is contained in:
Salad Dais
2022-09-18 07:49:18 +00:00
parent fac44a12b0
commit e6423d2f43
2 changed files with 110 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files

42
hippolyzer/apps/proxy_leapreceiver.py
View File

@@ -2,27 +2,38 @@
Simple stub for testing receiving inbound LEAP connections over TCP
To be removed at some point once this is supported by the proxy itself.
For now it's a bunch of example uses of the APIs.
"""
from typing import *
import asyncio
import logging
import multiprocessing
import pprint
from hippolyzer.lib.proxy.leap import LEAPBridgeServer, LEAPClient, LLWindowWrapper, UIPath, LLUIWrapper
from hippolyzer.lib.proxy.leap import (
LEAPBridgeServer,
LEAPClient,
UIPath,
LLCommandDispatcherWrapper,
LLWindowWrapper,
LLUIWrapper,
LLViewerControlWrapper,
)
async def client_connected(client: LEAPClient):
# Kick off a request to get ops for each API supported by the viewer
# Won't wait for a response from the viewer between each send
api_futs = {}
api_futs: Dict[Awaitable, str] = {}
for api_name in (await client.sys_command("getAPIs")).keys():
api_fut = client.sys_command("getAPI", {"api": api_name})
api_futs[api_fut] = api_name
# Wait for all of our commands to complete in parallel
# Wait for all of our getAPI commands to complete in parallel
for fut in (await asyncio.wait(api_futs.keys()))[0]:
# Figure out which API this future even relates to
# Print out which API this future even relates to
print("=" * 5, api_futs[fut], "=" * 5)
# List supported ops for this api
pprint.pprint(await fut)
@@ -53,18 +64,27 @@ async def client_connected(client: LEAPClient):
# `command()` except it internally addresses whatever the system command pump is.
await client.sys_command("ping")
# Print out all the commands supported by LLCommandDispatcher
cmd_dispatcher_api = LLCommandDispatcherWrapper(client)
pprint.pprint(await cmd_dispatcher_api.enumerate())
# 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.
window_api = LLWindowWrapper(client)
await window_api.mouse_click(button="LEFT", path=textbox_path)
# Type some text in it
# Clear out the textbox, note that this does _not_ work when path is specified!
# TODO: clearing a textbox isn't so nice. CTL+A doesn't work as expected even without a path,
# it leaves a capital "A" in the text editor. We get rid of it by doing backspace right after.
window_api.key_press(keysym="a", mask=["CTL"])
window_api.key_press(keysym="Backsp")
# Type some text
window_api.text_input("Also I can type in here pretty good.")
# Print out the value of the textbox we just typed in
@@ -73,11 +93,15 @@ async def client_connected(client: LEAPClient):
# But you don't need to explicitly give input focus like above, you can send keypresses
# directly to a path.
# TODO: clearing a textbox isn't so nice. CTL+A doesn't work as expected even without a path,
# it leaves a capital "A" in the text editor.
monospace_path = UIPath.for_floater("floater_test_textbox") / "monospace_text_editor"
window_api.text_input("I typed in here by path.", path=monospace_path)
# We can also access the viewer config to reason about viewer state.
viewer_control_api = LLViewerControlWrapper(client)
pprint.pprint(await viewer_control_api.get("Global", "StatsPilotFile"))
# Print the first ten vars in the "Global" group
pprint.pprint((await viewer_control_api.vars("Global"))[:10])
def receiver_main():
logging.basicConfig(level=logging.INFO)

99
hippolyzer/lib/proxy/leap.py
View File

@@ -2,6 +2,7 @@
Tooling for working with the SL viewer's LEAP integration: now with 100% less eventlet
TODO: split this out into its own package
TODO: Support playback of VITA event recording format snippets? Does anyone actually use those?
"""
from __future__ import annotations
@@ -16,6 +17,7 @@ import dataclasses
import enum
import logging
import pathlib
import posixpath
import uuid
import weakref
@@ -114,19 +116,19 @@ class LEAPClient:
"""Make a request to an internal LEAP method using the standard command form (op in data)"""
data = data.copy() if data else {}
data['op'] = op
return self.send_message(pump, data)
return self.post(pump, data, expect_reply=True)
def void_command(self, pump: str, op: str, data: Optional[Dict] = None) -> None:
"""Like `command()`, but we don't expect a reply."""
data = data.copy() if data else {}
data['op'] = op
self.send_message(pump, data, expect_reply=False)
self.post(pump, data, expect_reply=False)
def send_message(self, pump: str, data: Any, expect_reply: bool = True) -> Optional[Awaitable]:
def post(self, pump: str, data: Any, expect_reply: bool) -> Optional[Awaitable]:
"""
Send a message with request semantics to the other side
Post an event to the other side's `pump`.
Sending the message is done synchronously, only waiting for the reply is done async.
Post the event is done synchronously, only waiting for the reply is done async.
"""
assert self.connected
# If you don't pass in a dict for data, we have nowhere to stuff `reqid`.
@@ -303,6 +305,18 @@ class UIPath(pathlib.PurePosixPath):
def for_floater(cls, floater_name: str) -> UIPath:
return cls("/main_view/menu_stack/world_panel/Floater View") / floater_name
def __str__(self) -> str:
"""Like the base __str__ except ".." and "." segments will be resolved."""
return posixpath.normpath(super().__str__())
async def _data_unwrapper(data_fut: Awaitable[Dict], inner_elem: str) -> Any:
"""Unwraps part of the data future while allowing the request itself to remain synchronous"""
# We want the request to be sent immediately, without requiring the request 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`.
return (await data_fut)[inner_elem]
UI_PATH_TYPE = Optional[Union[str, UIPath]]
@@ -335,37 +349,37 @@ class LLWindowWrapper(LEAPAPIWrapper):
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):
keysym: KEYSYM_TYPE = None, char: CHAR_TYPE = None, path: UI_PATH_TYPE = None) -> 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):
keysym: KEYSYM_TYPE = None, char: CHAR_TYPE = None, path: UI_PATH_TYPE = None) -> 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):
keysym: KEYSYM_TYPE = None, char: CHAR_TYPE = None, path: UI_PATH_TYPE = None) -> 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: UI_PATH_TYPE = None):
def text_input(self, text_input: str, path: UI_PATH_TYPE = None) -> 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:
async def get_paths(self, under: UI_PATH_TYPE = "") -> 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})
return await self._client.command(self._pump_name, "getPaths", {"under": str(under)})
async def get_info(self, path: str) -> Dict:
async def get_info(self, path: UI_PATH_TYPE) -> Dict:
"""Get info about an element specified by path"""
return await self._client.command(self._pump_name, "getInfo", {"path": path})
return await self._client.command(self._pump_name, "getInfo", {"path": str(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:
@@ -410,7 +424,7 @@ class LLWindowWrapper(LEAPAPIWrapper):
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):
def mouse_scroll(self, clicks: int) -> None:
"""Act as if the scroll wheel has been moved `clicks` amount. May be negative"""
self._client.command(self._pump_name, "mouseScroll", {"clicks": clicks})
@@ -428,17 +442,58 @@ class LLUIWrapper(LEAPAPIWrapper):
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)})
return _data_unwrapper(resp_fut, "value")
async def _wrapper():
return (await resp_fut)['value']
return _wrapper()
class LLCommandDispatcherWrapper(LEAPAPIWrapper):
PUMP_NAME = "LLCommandDispatcher"
def dispatch(
self,
cmd: str,
/,
*,
params: Optional[Collection[str]] = None,
query: Optional[Dict[str, str]] = None,
trusted: bool = True,
) -> None:
"""Execute a command registered as an LLCommandHandler"""
return self._client.void_command(self._pump_name, "dispatch", {
"cmd": cmd,
"params": params or [],
"query": query or {},
"trusted": trusted,
})
def enumerate(self) -> Awaitable[Dict]:
"""Get map of registered LLCommandHandlers, containing name, key, and (e.g.) untrusted flag"""
return self._client.command(self._pump_name, "enumerate")
class LLViewerControlWrapper(LEAPAPIWrapper):
PUMP_NAME = "LLViewerControl"
def get(self, group: str, key: str) -> Awaitable[Any]:
"""
Get value of a Control (config) key
`group` can be one of "CrashSettings", "Global", "PerAccount", "Warnings".
"""
return self._client.command(self._pump_name, "get", {"key": key, "group": group})
def vars(self, group: str) -> Awaitable[List[Dict]]:
"""Return a list of dicts of controls in `group`"""
resp_fut = self._client.command(self._pump_name, "vars", {"group": group})
return _data_unwrapper(resp_fut, "vars")
def set(self, group: str, key: str, value: Any) -> None:
"""
Set a configuration value
TODO: error handling based on "error" field in resp?
"""
self._client.void_command(self._pump_name, "set", {"key": key, "group": group, "value": value})
class LEAPBridgeServer: