attempt to add plugin events to the plugin frontend api.

unable to test right now though
This commit is contained in:
Party Wumpus
2024-04-09 15:44:38 +01:00
committed by PartyWumpus
parent de9d2144a6
commit f9ff518e6d
6 changed files with 57 additions and 21 deletions
+2 -3
View File
@@ -19,7 +19,7 @@ import subprocess
import logging
import time
from typing import TypeVar, Type
from typing import Any
"""
Constants
@@ -213,9 +213,8 @@ logger.setLevel(logging.INFO)
"""
Event handling
"""
DataType = TypeVar("DataType")
# TODO better docstring im lazy
async def emit(event: str, data: DataType | None = None, data_type: Type[DataType] | None = None) -> None:
async def emit(event: str, *args: Any) -> None:
"""
Send an event to the frontend.
"""
@@ -16,7 +16,7 @@ __version__ = '0.1.0'
import logging
from typing import TypeVar, Type
from typing import Any
"""
Constants
@@ -177,9 +177,8 @@ logger: logging.Logger
"""
Event handling
"""
DataType = TypeVar("DataType")
# TODO better docstring im lazy
async def emit(event: str, data: DataType | None = None, data_type: Type[DataType] | None = None) -> None:
async def emit(event: str, *args: Any) -> None:
"""
Send an event to the frontend.
"""
+1 -1
View File
@@ -59,7 +59,7 @@ class PluginWrapper:
if line != None:
res = loads(line)
if res["type"] == SocketMessageType.EVENT.value:
create_task(self.emitted_event_callback(res["event"], res["data"]))
create_task(self.emitted_event_callback(res["event"], res["args"]))
elif res["type"] == SocketMessageType.RESPONSE.value:
self._method_call_requests.pop(res["id"]).set_result(res)
except:
@@ -14,7 +14,7 @@ from ..localplatform.localplatform import setgid, setuid, get_username, get_home
from ..enums import UserType
from .. import helpers
from typing import List, TypeVar, Type
from typing import List, TypeVar, Any
DataType = TypeVar("DataType")
@@ -83,11 +83,11 @@ class SandboxedPlugin:
sysmodules[key.replace("decky_loader.", "")] = sysmodules[key]
from .imports import decky
async def emit(event: str, data: DataType | None = None, data_type: Type[DataType] | None = None) -> None:
async def emit(event: str, *args: Any) -> None:
await self._socket.write_single_line_server(dumps({
"type": SocketMessageType.EVENT,
"event": event,
"data": data
"args": args
}))
# copy the docstring over so we don't have to duplicate it
emit.__doc__ = decky.emit.__doc__