Mercurial > libervia-backend
diff libervia/frontends/tools/webrtc_remote_control.py @ 4270:0d7bb4df2343
Reformatted code base using black.
author | Goffi <goffi@goffi.org> |
---|---|
date | Wed, 19 Jun 2024 18:44:57 +0200 |
parents | 8acf46ed7f36 |
children |
line wrap: on
line diff
--- a/libervia/frontends/tools/webrtc_remote_control.py Tue Jun 18 12:06:45 2024 +0200 +++ b/libervia/frontends/tools/webrtc_remote_control.py Wed Jun 19 18:44:57 2024 +0200 @@ -37,7 +37,7 @@ ) gi.require_versions({"Gst": "1.0", "GstWebRTC": "1.0"}) -OnOpenCbType = Callable[["WebRTCRemoteController"], None|Awaitable[None]] +OnOpenCbType = Callable[["WebRTCRemoteController"], None | Awaitable[None]] MOUSE_BTN_LEFT = 0x110 MOUSE_BTN_RIGHT = 0x111 MOUSE_BTN_MIDDLE = 0x112 @@ -71,7 +71,7 @@ self.on_call_start_cb = on_call_start_cb self.end_call_cb = end_call_cb self.loop = asyncio.get_event_loop() - self.data_channel: GstWebRTC.WebRTCDataChannel|None = None + self.data_channel: GstWebRTC.WebRTCDataChannel | None = None def send_input(self, input_data: dict) -> None: """Send an input data to controlled device @@ -95,9 +95,7 @@ rc_data.update(options) remote_control_data_s = await self.bridge.remote_control_start( str(callee), - data_format.serialise( - rc_data - ), + data_format.serialise(rc_data), profile, ) remote_control_data = data_format.deserialise(remote_control_data_s) @@ -126,10 +124,7 @@ await aio.maybe_async(on_open_cb(self)) async def start( - self, - callee: jid.JID, - options: dict, - on_open_cb: OnOpenCbType + self, callee: jid.JID, options: dict, on_open_cb: OnOpenCbType ) -> None: """Start a remote control session with ``callee`` @@ -154,8 +149,11 @@ class WebRTCRemoteControlReceiver: def __init__( - self, bridge, profile: str, on_close_cb: Callable[[], Any] | None = None, - verbose: bool = False + self, + bridge, + profile: str, + on_close_cb: Callable[[], Any] | None = None, + verbose: bool = False, ) -> None: """Initializes the File Receiver. @@ -169,8 +167,8 @@ self.on_close_cb = on_close_cb self.loop = asyncio.get_event_loop() self.desktop_portal = None - self.remote_desktop_data: dict|None = None - self.stream_node_id: int|None = None + self.remote_desktop_data: dict | None = None + self.stream_node_id: int | None = None self.verbose = verbose async def do_input(self, data: dict) -> None: @@ -183,9 +181,7 @@ x, y = data["x"], data["y"] except KeyError: dx, dy = data["movementX"], data["movementY"] - await self.desktop_portal.notify_pointer_motion( - dx, dy - ) + await self.desktop_portal.notify_pointer_motion(dx, dy) else: assert self.stream_node_id is not None await self.desktop_portal.notify_pointer_motion_absolute( @@ -200,28 +196,23 @@ # see https://developer.mozilla.org/en-US/docs/Web/API/MouseEvent/buttons#value if buttons & 1: await self.desktop_portal.notify_pointer_button( - MOUSE_BTN_LEFT, - state + MOUSE_BTN_LEFT, state ) if buttons & 2: await self.desktop_portal.notify_pointer_button( - MOUSE_BTN_RIGHT, - state + MOUSE_BTN_RIGHT, state ) if buttons & 4: await self.desktop_portal.notify_pointer_button( - MOUSE_BTN_MIDDLE, - state + MOUSE_BTN_MIDDLE, state ) if buttons & 8: await self.desktop_portal.notify_pointer_button( - MOUSE_BTN_BACK, - state + MOUSE_BTN_BACK, state ) if buttons & 16: await self.desktop_portal.notify_pointer_button( - MOUSE_BTN_FORWARD, - state + MOUSE_BTN_FORWARD, state ) elif type_ == "wheel": dx = data.get("deltaX", 0) @@ -229,23 +220,14 @@ delta_mode = data["deltaMode"] if delta_mode == 0: # deltas are in pixels - await self.desktop_portal.notify_pointer_axis( - dx, - dy - ) + await self.desktop_portal.notify_pointer_axis(dx, dy) else: # deltas are in steps (see # https://developer.mozilla.org/en-US/docs/Web/API/Element/wheel_event#event_properties) if dx: - await self.desktop_portal.notify_pointer_axis( - 1, - dx - ) + await self.desktop_portal.notify_pointer_axis(1, dx) if dy: - await self.desktop_portal.notify_pointer_axis( - 0, - dy - ) + await self.desktop_portal.notify_pointer_axis(0, dy) elif type_.startswith("key"): # FIXME: this is a really naive implementation, it needs tot be improved. key = data["key"] @@ -258,18 +240,13 @@ except Exception: log.exception(f"Can't handle input {data}") - def _on_dc_message_data(self, data_channel, glib_data) -> None: """A data chunk of the file has been received.""" raw = glib_data.get_data() data = cbor2.loads(raw) if self.verbose: print(data) - aio.run_from_thread( - self.do_input, - data, - loop=self.loop - ) + aio.run_from_thread(self.do_input, data, loop=self.loop) def _on_dc_close(self, data_channel) -> None: """Data channel is closed @@ -284,9 +261,7 @@ def _on_data_channel(self, webrtcbin, data_channel) -> None: """The data channel has been opened.""" - data_channel.connect( - "on-message-data", self._on_dc_message_data - ) + data_channel.connect("on-message-data", self._on_dc_message_data) data_channel.connect("on-close", self._on_dc_close) async def request_remote_desktop(self, with_screen_sharing: bool) -> None: @@ -295,6 +270,7 @@ @param with_screen_sharing: True if screen must be shared. """ from .portal_desktop import DesktopPortal + self.desktop_portal = DesktopPortal() self.remote_desktop_data = await self.desktop_portal.request_remote_desktop( with_screen_sharing @@ -302,10 +278,7 @@ print(self.remote_desktop_data) async def start_receiving( - self, - from_jid: jid.JID, - session_id: str, - screenshare: dict + self, from_jid: jid.JID, session_id: str, screenshare: dict ) -> None: """Receives a file via WebRTC and saves it to the specified path. @@ -328,7 +301,7 @@ video_properties={ "path": str(self.stream_node_id), "do-timestamp": 1, - } + }, ) except KeyError: sources_data = SourcesNone() @@ -339,9 +312,11 @@ self.bridge, self.profile, call_data, - sources_data = sources_data, + sources_data=sources_data, sinks_data=webrtc.SinksNone(), - dc_data_list=[webrtc.SinksDataChannel( - dc_on_data_channel=self._on_data_channel, - )], + dc_data_list=[ + webrtc.SinksDataChannel( + dc_on_data_channel=self._on_data_channel, + ) + ], )