WebRTCInput
The server-side input authority shared by both transports.
Dispatches every client data-channel/WebSocket message: keyboard, mouse, gamepad, clipboard (including multipart transfers), cursor monitoring, and the settings/stats callbacks the streaming pipeline registers on it. One instance exists per transport service; X11 and Wayland sessions flow through the same handler so behavior stays in parity, with the backend chosen by is_wayland.
Keyboard injection follows the module-level fallback ladders. On Wayland all key work is serialized through keyboard_queue and a single worker so key ordering holds across the seat, virtual-keyboard, and clipboard rungs; on X11 injection is direct (XTEST first, xdotool fallback) with stale-key sweeping and server-side auto-repeat emulation.
Attributes
attributewayland_socket_index= wayland_socket_indexattributeapp_wayland_display= app_wayland_displaySocket of the compositor apps run under (input and
clipboard target) when it differs from the pixelflux capture
compositor; resolved lazily into _app_wl_display_cached, since a
nested session comes up after this process.
attribute_app_wl_display_cached= Noneattribute_x_reconnect_thread= Noneattribute_x11_monitor_build_lock= asyncio.Lock()attribute_x_event_wake= Noneattribute_x_watcher_fd= Noneattribute_app_wl_negcache= Noneattribute_app_wl_negcache_at= 0.0attribute_app_wl_is_separate= FalseTrue once a resolved app compositor is confirmed
distinct from the capture compositor, so
_has_separate_app_compositor answers without re-resolving.
attributeactive_shortcut_modifiers= set()attributeSHORTCUT_MODIFIER_XKEY_NAMES= {'Control_L', 'Control_R', 'Alt_L', 'Alt_R', 'Super_L', 'Super_R', 'Meta_L', 'Meta_R'}attributeactive_modifiers= set()attributeatomically_typed_keys= set()attributetranslated_keys= set()attributeACTION_MODIFIER_KEYSYMS= {65507, 65508, 65513, 65514, 65511, 65512, 65515, 65516, 65517, 65518}attributeLEVEL_MODIFIER_KEYSYMS= frozenset({65505, 65506, 65027, 65406})Shift_L/R, ISO_Level3_Shift, Mode_switch — the level-selecting modifiers whose client-held state the injectors consult in place of a per-press server query.
attributeMODIFIER_KEYSYMS= {65505, 65506, 65507, 65508, 65513, 65514, 65027, 65511, 65512, 65515, 65516, 65517, 65518}Keysyms never armed for auto-repeat or routed as ordinary keys. Super and Hyper are included because the client maps the Meta/Windows key to Super, not Meta.
attributertc_app= rtc_appattributeloop= asyncio.get_running_loop()attributejs_socket_path_prefix= js_socket_path_prefixattributenum_gamepads= 4attributegamepad_instances= {}attributeclient_gamepad_associations= {}attributegamepad_heartbeats= {}Slot to last held-state heartbeat (js,h). Only
slots that have sent one are swept: a client that never heartbeats
(an older web client) keeps the transport-close release path alone.
attributeuinput_gamepads= uinput_gamepads_enabled(uinput_gamepad)Resolved once so the decision and its reason are logged at startup rather than per slot.
attributeclipboard_running= Falseattribute_binary_clipboard_lock= asyncio.Lock()Serializes update_binary_clipboard_setting so its cancel+reassign of the monitor task is atomic under rapid toggles.
attribute_clipboard_monitor_active= FalseSingleton guard; one start_clipboard loop runs at a time.
attributeuinput_mouse_socket_path= uinput_mouse_socket_pathattributeuinput_mouse_socket= Noneattributeenable_clipboard= enable_clipboardattributeenable_binary_clipboard= enable_binary_clipboardattribute_apps_runner_okOptional[bool]= Noneattributeenable_cursors= enable_cursorsattributecursors_running= Falseattributecursor_scale= cursor_scaleattributecursor_size= cursor_sizeattributecursor_debug= cursor_debugattributemax_cursor_size= max_cursor_sizeattributesystem_dpi= 96.0attributecursor_size_cap= max_cursor_sizeattribute_cursor_msg_cache= None(serial, size cap) to encoded cursor message, see
_encode_cursor.
attributekeyboard= Noneattributemouse= Noneattributexdisplay= Noneattributebutton_mask= 0attributelast_x= -1attributelast_y= -1attributetracked_position_stale= Falseattributeping_start= Noneattributeupload_dir= upload_dirattributeupload_dir_path= Noneattributeon_video_encoder_bit_rate= _unhandled_video_bitrateattributeon_audio_encoder_bit_rate= _unhandled_audio_bitrateattributeon_mouse_pointer_visible= _unhandled_mouse_pointerattributeon_clipboard_read= self._on_clipboard_readattributeon_set_fps= lambda fps, display_id='primary': logger_webrtc_input.warning('unhandled on_set_fps')attributeon_request_keyframe= lambda display_id='primary': logger_webrtc_input.warning('unhandled on_request_keyframe')attributeon_set_enable_resize= lambda enable_resize, res: logger_webrtc_input.warning('unhandled on_set_enable_resize')attributeon_client_fps= lambda fps: logger_webrtc_input.warning('unhandled on_client_fps')attributeon_client_latency= lambda latency: logger_webrtc_input.warning('unhandled on_client_latency')attributeon_resize= lambda res, display_id='primary': logger_webrtc_input.warning('unhandled on_resize')attributeon_scaling_ratio= lambda res: logger_webrtc_input.warning('unhandled on_scaling_ratio')attributeon_ping_response= lambda latency: logger_webrtc_input.warning('unhandled on_ping_response')attributeon_cursor_change= self._on_cursor_changeattributeon_client_webrtc_stats= _unhandled_webrtc_statsattributeclipboard_monitor_task= Noneattributemultipart_clipboard_buffer= Noneattributemultipart_clipboard_mime_type= 'text/plain'attributemultipart_clipboard_total_size= 0attributemultipart_clipboard_in_progress= Falseattributemultipart_clipboard_id= Noneattributemultipart_clipboard_kind= Noneattributedata_server_instance= data_server_instanceattributeon_update_settings= lambda settings_json, display_id='primary': logger_webrtc_input.warning('unhandled update_settings')attributeis_wayland= is_waylandattributewayland_input= ScreenCapture()attribute_client_kb_layout= NoneLast client keyboardLayout hint seen (SETTINGS) on either transport.
attribute_wl_seat_client_layout= NoneThe hint the Wayland seat's base layout currently carries; None while the seat is on the deployment layout.
attribute_wl_keymap_owner= NoneKeysym policy for the seat, kept here rather than in
the compositor: built lazily from the compositor's keymap, retried
on a cooldown (_wl_keymap_retry_at) if that read fails, and
rebuilt carrying held keys once a base-layout change sets
_wl_keymap_stale.
attribute_wl_keymap_stale= Falseattribute_wl_keymap_owner_lock= asyncio.Lock()attribute_wl_keymap_retry_at= 0.0attribute_wl_typer_lock= asyncio.Lock()attribute_wl_typer_retry_at= 0.0attribute_clipboard_inject_lock= asyncio.Lock()attribute_clipboard_inject_active= Falseattribute_clipboard_last_bytes= NoneChange-detection baseline shared by the monitor and write_clipboard: content this server just wrote is never re-broadcast (client/server echo loop), and the baseline survives client reconnects so nothing is resent unchanged.
attribute_x11_clipboard_monitor= Noneattribute_x11_monitor_retry_at= 0.0attribute_x11_monitor_unavail_logged= Falseattribute_app_watch_failure= Noneattribute_app_clip_read_failure= Noneattribute_wl_native_arm_failure= NoneLast error the compositor clipboard callback failed to arm with; the monitor retries each tick and reports each distinct error once.
attribute_session_env_cache= {}attribute_session_env_empty_at= {}attribute_session_env_negcache_ttl= 5.0attribute_xclip_missing_warned= Falseattribute_last_clipboard_request_ts= {}REQUEST_CLIPBOARD (Ctrl/Cmd+C) debounce, keyed per requesting connection so a keypress storm cannot stack clipboard reads and one client's copy cannot suppress another's.
attribute_clipboard_request_debounce= 0.25attribute_bg_tasks= set()Strong refs for fire-and-forget tasks; asyncio holds running tasks weakly, so an unreferenced one can be collected mid-flight.
attributekeyboard_queue= asyncio.Queue(maxsize=4096)attributekeyboard_worker_task= Noneattribute_wl_text_routed= {}Keysyms whose kd became buffered text on the Wayland worker (nested app compositor): their ku is swallowed, not released.
attributepressed_keys= {}Keysym to last kh heartbeat; the sweep auto-releases
any key whose heartbeat stops (a key-up lost to congestion).
attributereaped_atomic_keys= set()Atomic (non-alpha) keys the sweep reaped were never physically held, so a late ku would emit a spurious keyup; tracked to swallow it, and a fresh kd clears the entry.
attributemax_pressed_keys= 1024Cap so a kd flood cannot grow pressed_keys unbounded.
attributekey_stale_window= 2.0Clients heartbeat every 100 ms but hidden tabs throttle to >= 1 s, so 2 s avoids false-releasing a backgrounded held key.
attributekey_sweep_interval= 0.1attributekey_sweep_task= Noneattributekey_repeat_enabled= not self.is_waylandServer-side auto-repeat, X11 only: XTEST/xdotool synthetic presses do not trigger the server's native repeat, so a held key would emit one character. Off on Wayland, where the focused app repeats virtual-keyboard keys itself via wl_keyboard repeat_info and a server-side repeat would double it.
attributekey_repeat_delay= 0.5attributekey_repeat_interval= 0.04attributekey_repeat_tick= 0.02attributekey_repeat_heartbeat_grace= 0.3Repeat pauses when the held key's last heartbeat is older than this (stalled stream / hidden tab); kept above ~3x the client's 100 ms heartbeat.
attributekey_repeat_state= {}Keysym to the monotonic time of its next due repeat.
attributekey_repeat_task= Noneattributekeymap_watch_task= NoneMappingNotify consumer for sessions where the cursor monitor (the normal X event consumer) is disabled.
attributeon_update_rate_control_mode= lambda mode, display_id='primary': logger_webrtc_input.warning('unhandled on_update_rate_control_mode')attributeon_update_crf= lambda value, display_id='primary': logger_webrtc_input.warning('unhandled on_update_crf')attribute_WL_RESET_WAIT_S= 5.0attributeSPARE_SCREEN_SIZE= (320, 240)Functions
func__init__(self, rtc_app, uinput_mouse_socket_path='', js_socket_path_prefix='/tmp', enable_clipboard='', enable_binary_clipboard='', enable_cursors=True, cursor_size=16, cursor_scale=1.0, cursor_debug=False, max_cursor_size=32, data_server_instance=None, upload_dir=None, is_wayland=False, wayland_socket_index=0, app_wayland_display='', uinput_gamepad='auto') -> NoneSource Code
def __init__(
self,
rtc_app: Any,
uinput_mouse_socket_path: str = "",
js_socket_path_prefix: str = "/tmp",
enable_clipboard: str = "",
enable_binary_clipboard: str = "",
enable_cursors: bool = True,
cursor_size: int = 16,
cursor_scale: float = 1.0,
cursor_debug: bool = False,
max_cursor_size: int = 32,
data_server_instance: Any = None,
upload_dir: Optional[str] = None,
is_wayland: bool = False,
wayland_socket_index: int = 0,
app_wayland_display: str = "",
uinput_gamepad: str = "auto",
) -> None:
self.wayland_socket_index = wayland_socket_index
self.app_wayland_display = app_wayland_display
self._app_wl_display_cached = None
self._x_reconnect_thread = None
self._x11_monitor_build_lock = asyncio.Lock()
self._x_event_wake = None
self._x_watcher_fd = None
self._app_wl_negcache = None
self._app_wl_negcache_at = 0.0
self._app_wl_is_separate = False
self.active_shortcut_modifiers = set()
self.SHORTCUT_MODIFIER_XKEY_NAMES = {
'Control_L', 'Control_R',
'Alt_L', 'Alt_R',
'Super_L', 'Super_R',
'Meta_L', 'Meta_R'
}
self.active_modifiers = set()
self.atomically_typed_keys = set()
self.translated_keys = set()
self.ACTION_MODIFIER_KEYSYMS = {65507, 65508, 65513, 65514, 65511, 65512,
65515, 65516, 65517, 65518}
self.LEVEL_MODIFIER_KEYSYMS = frozenset({0xFFE1, 0xFFE2, 0xFE03, 0xFF7E})
self.MODIFIER_KEYSYMS = {
# Shift_L, Shift_R
65505, 65506,
# Control_L, Control_R
65507, 65508,
# Alt_L, Alt_R
65513, 65514,
# ISO_Level3_Shift (AltGr)
65027,
# Meta_L, Meta_R
65511, 65512,
# Super_L, Super_R
65515, 65516,
# Hyper_L, Hyper_R
65517, 65518,
}
self.rtc_app = rtc_app
self.loop = asyncio.get_running_loop()
self.js_socket_path_prefix = js_socket_path_prefix
self.num_gamepads = 4
self.gamepad_instances = {}
self.client_gamepad_associations = {}
self.gamepad_heartbeats = {}
self.uinput_gamepads = uinput_gamepads_enabled(uinput_gamepad)
self.clipboard_running = False
self._binary_clipboard_lock = asyncio.Lock()
self._clipboard_monitor_active = False
self.uinput_mouse_socket_path = uinput_mouse_socket_path
self.uinput_mouse_socket = None
self.enable_clipboard = enable_clipboard
self.enable_binary_clipboard = enable_binary_clipboard
self._apps_runner_ok: Optional[bool] = None
self.enable_cursors = enable_cursors
self.cursors_running = False
self.cursor_scale = cursor_scale
self.cursor_size = cursor_size
self.cursor_debug = cursor_debug
# An explicit cursor_size raises the capture cap so the requested size
# survives the transport instead of being resized down.
if isinstance(cursor_size, int) and cursor_size > 0:
max_cursor_size = max(max_cursor_size, cursor_size)
self.max_cursor_size = max_cursor_size
self.system_dpi = 96.0
self.cursor_size_cap = max_cursor_size
self._cursor_msg_cache = None
self.keyboard = None
self.mouse = None
self.xdisplay = None
self.button_mask = 0
self.last_x = -1
self.last_y = -1
self.tracked_position_stale = False
self.ping_start = None
self.upload_dir = upload_dir
self.upload_dir_path = None
async def _unhandled_video_bitrate(bitrate, display_id="primary"):
logger_webrtc_input.warning(f"unhandled on_video_encoder_bit_rate: {bitrate}")
self.on_video_encoder_bit_rate = _unhandled_video_bitrate
async def _unhandled_audio_bitrate(bitrate):
logger_webrtc_input.warning(f"unhandled on_audio_encoder_bit_rate: {bitrate}")
self.on_audio_encoder_bit_rate = _unhandled_audio_bitrate
async def _unhandled_mouse_pointer(visible):
logger_webrtc_input.warning(f"unhandled on_mouse_pointer_visible: {visible}")
self.on_mouse_pointer_visible = _unhandled_mouse_pointer
self.on_clipboard_read = self._on_clipboard_read
self.on_set_fps = lambda fps, display_id="primary": logger_webrtc_input.warning("unhandled on_set_fps")
self.on_request_keyframe = lambda display_id="primary": logger_webrtc_input.warning("unhandled on_request_keyframe")
self.on_set_enable_resize = lambda enable_resize, res: logger_webrtc_input.warning("unhandled on_set_enable_resize")
self.on_client_fps = lambda fps: logger_webrtc_input.warning("unhandled on_client_fps")
self.on_client_latency = lambda latency: logger_webrtc_input.warning("unhandled on_client_latency")
self.on_resize = lambda res, display_id="primary": logger_webrtc_input.warning("unhandled on_resize")
self.on_scaling_ratio = lambda res: logger_webrtc_input.warning("unhandled on_scaling_ratio")
self.on_ping_response = lambda latency: logger_webrtc_input.warning("unhandled on_ping_response")
self.on_cursor_change = self._on_cursor_change
async def _unhandled_webrtc_stats(webrtc_stat_type, webrtc_stats):
logger_webrtc_input.debug(f"unhandled on_client_webrtc_stats: {webrtc_stat_type}")
self.on_client_webrtc_stats = _unhandled_webrtc_stats
self.clipboard_monitor_task = None
self.multipart_clipboard_buffer = None
self.multipart_clipboard_mime_type = "text/plain"
self.multipart_clipboard_total_size = 0
self.multipart_clipboard_in_progress = False
self.multipart_clipboard_id = None
self.multipart_clipboard_kind = None
self.data_server_instance = data_server_instance
self.on_update_settings = lambda settings_json, display_id="primary": logger_webrtc_input.warning("unhandled update_settings")
self.is_wayland = is_wayland
self.wayland_input = None
self._client_kb_layout = None
self._wl_seat_client_layout = None
self._wl_keymap_owner = None
self._wl_keymap_stale = False
self._wl_keymap_owner_lock = asyncio.Lock()
self._wl_keymap_retry_at = 0.0
self._wl_typer_lock = asyncio.Lock()
self._wl_typer_retry_at = 0.0
self._clipboard_inject_lock = asyncio.Lock()
self._clipboard_inject_active = False
self._clipboard_last_bytes = None
self._x11_clipboard_monitor = None
self._x11_monitor_retry_at = 0.0
self._x11_monitor_unavail_logged = False
self._app_watch_failure = None
self._app_clip_read_failure = None
self._wl_native_arm_failure = None
self._session_env_cache = {}
self._session_env_empty_at = {}
self._session_env_negcache_ttl = 5.0
self._xclip_missing_warned = False
self._last_clipboard_request_ts = {}
self._clipboard_request_debounce = 0.25
self._bg_tasks = set()
self.keyboard_queue = asyncio.Queue(maxsize=4096)
self.keyboard_worker_task = None
self._wl_text_routed = {}
self.pressed_keys = {}
self.reaped_atomic_keys = set()
self.max_pressed_keys = 1024
self.key_stale_window = 2.0
self.key_sweep_interval = 0.1
self.key_sweep_task = None
self.key_repeat_enabled = not self.is_wayland
self.key_repeat_delay = 0.5
self.key_repeat_interval = 0.04
self.key_repeat_tick = 0.02
self.key_repeat_heartbeat_grace = 0.3
self.key_repeat_state = {}
self.key_repeat_task = None
self.keymap_watch_task = None
self.on_update_rate_control_mode = lambda mode, display_id="primary": logger_webrtc_input.warning("unhandled on_update_rate_control_mode")
self.on_update_crf = lambda value, display_id="primary": logger_webrtc_input.warning("unhandled on_update_crf")
if self.is_wayland:
try:
if ScreenCapture is None:
raise RuntimeError("pixelflux is not installed")
self.wayland_input = ScreenCapture()
logger_webrtc_input.info("Wayland input injection initialized.")
missing = [m for m in (
"clipboard_write_app", "clipboard_unwatch_app",
"list_outputs", "create_output", "set_keymap_overlay",
"hold_spare_app_screens", "set_app_output_scale",
"set_app_screen_geometry",
"set_app_wayland_display", "type_text_wayland",
"get_keyboard_state",
) if not hasattr(self.wayland_input, m)]
if missing:
logger_webrtc_input.warning(
"Installed pixelflux is missing APIs this build "
"expects; Wayland features that depend on them "
"degrade or stay off. Update pixelflux to a "
"matching build.")
logger_webrtc_input.debug(
f"pixelflux methods absent: {', '.join(missing)}")
except Exception as e:
logger_webrtc_input.error(f"Failed to initialize Wayland input: {e}")paramselfparamrtc_appAnyparamuinput_mouse_socket_pathstr= ''paramjs_socket_path_prefixstr= '/tmp'paramenable_clipboardstr= ''paramenable_binary_clipboardstr= ''paramenable_cursorsbool= Trueparamcursor_sizeint= 16paramcursor_scalefloat= 1.0paramcursor_debugbool= Falseparammax_cursor_sizeint= 32paramdata_server_instanceAny= Noneparamupload_dirOptional[str]= Noneparamis_waylandbool= Falseparamwayland_socket_indexint= 0paramapp_wayland_displaystr= ''paramuinput_gamepadstr= 'auto'Returns
Nonefunc_on_clipboard_read(self, data, mime_type='text/plain') -> NoneSource Code
async def _on_clipboard_read(self, data: Union[str, bytes],
mime_type: str = "text/plain") -> None:
await self.send_clipboard_data(data, mime_type)paramselfparamdataUnion[str, bytes]parammime_typestr= 'text/plain'Returns
Nonefunc_on_cursor_change(self, data) -> NoneSource Code
def _on_cursor_change(self, data: dict) -> None: self.send_cursor_data(data)paramselfparamdatadictReturns
Nonefuncsend_clipboard_data(self, data, mime_type='text/plain', reply_to=None, conn_id=None) -> NoneRoute clipboard content to the transport's own chunked sender.
Each transport owns one (SelkiesStreamingApp.send_ws_clipboard_data / RTCApp.send_clipboard_data) and addresses the requester by the identity its own connections carry.
Source Code
async def send_clipboard_data(self, data: Union[str, bytes],
mime_type: str = "text/plain",
reply_to: Optional[str] = None,
conn_id: Any = None) -> None:
"""Route clipboard content to the transport's own chunked sender.
Each transport owns one (SelkiesStreamingApp.send_ws_clipboard_data /
RTCApp.send_clipboard_data) and addresses the requester by the
identity its own connections carry.
"""
if self._ws_transport():
await self.rtc_app.send_ws_clipboard_data(data, mime_type, reply_to=reply_to,
conn_id=conn_id)
else:
await self.rtc_app.send_clipboard_data(data, mime_type, reply_to=reply_to,
peer_id=conn_id)paramselfparamdataUnion[str, bytes]parammime_typestr= 'text/plain'paramreply_toOptional[str]= Noneparamconn_idAny= NoneReturns
Nonefunc_ws_transport(self) -> boolWhether the owning app is the websockets transport (it carries a
mode); the WebRTC app has no such attribute and routes the other way.
Source Code
def _ws_transport(self) -> bool:
"""Whether the owning app is the websockets transport (it carries a
`mode`); the WebRTC app has no such attribute and routes the other way."""
return getattr(self.rtc_app, "mode", None) == "websockets"paramselfReturns
boolfuncsend_cursor_data(self, data) -> NoneSource Code
def send_cursor_data(self, data: dict) -> None:
if self._ws_transport(): self.rtc_app.send_ws_cursor_data(data)
else: self.rtc_app.send_cursor_data(data)paramselfparamdatadictReturns
Nonefuncsend_command_status(self, action, conn_id=None) -> NoneRoute a command notice (command_error/command_done) to the
transport (see send_cursor_data).
Each transport carries the system action on its own wire format. Over
WebRTC conn_id is the requesting peer's id, so the notice targets
that peer's channel; the websockets transport notifies its requesting
socket in its own cmd branch and only broadcasts here.
Source Code
def send_command_status(self, action: str, conn_id: Optional[str] = None) -> None:
"""Route a command notice (``command_error``/``command_done``) to the
transport (see send_cursor_data).
Each transport carries the system action on its own wire format. Over
WebRTC ``conn_id`` is the requesting peer's id, so the notice targets
that peer's channel; the websockets transport notifies its requesting
socket in its own cmd branch and only broadcasts here.
"""
try:
if self._ws_transport():
self.rtc_app.send_system_action(action)
else:
self.rtc_app.send_system_action(action, peer_id=conn_id)
except Exception:
logger_webrtc_input.debug("command status notify failed", exc_info=True)paramselfparamactionstrparamconn_idOptional[str]= NoneReturns
Nonefunc__keyboard_connect(self) -> NoneSource Code
def __keyboard_connect(self) -> None: self.keyboard = _XTestKeyboard(self.xdisplay) if self.xdisplay else NoneparamselfReturns
Nonefunc_apply_input_x_reply_bound(self) -> NoneBound the wait for an X REPLY on the shared input connection so an unresponsive server (driver hang, a foreign client's server grab) raises ConnectionClosedError instead of freezing the event loop forever. XTEST injection (mouse motion/buttons, key press) is a no-reply request and is unaffected — only the modifier query (query_keymap) and the cursor-image fetch block, and those recover via _reconnect_xdisplay(). Event waits stay unbounded (a quiet server sending no cursor events is not an error).
Source Code
def _apply_input_x_reply_bound(self) -> None:
"""Bound the wait for an X REPLY on the shared input connection so an
unresponsive server (driver hang, a foreign client's server grab) raises
ConnectionClosedError instead of freezing the event loop forever. XTEST
injection (mouse motion/buttons, key press) is a no-reply request and is
unaffected — only the modifier query (query_keymap) and the cursor-image
fetch block, and those recover via _reconnect_xdisplay(). Event waits stay
unbounded (a quiet server sending no cursor events is not an error)."""
if self.xdisplay is None:
return
try:
self.xdisplay.display.blocking_timeout = INPUT_X_REPLY_TIMEOUT_S
except Exception:
passparamselfReturns
Nonefunc_arm_x_event_watcher(self) -> None(Re)register the event-loop reader that wakes X consumers when the input connection's socket goes readable. Idempotent; re-arms when the fd changes under us (reconnect).
Source Code
def _arm_x_event_watcher(self) -> None:
"""(Re)register the event-loop reader that wakes X consumers when the
input connection's socket goes readable. Idempotent; re-arms when the fd
changes under us (reconnect)."""
if self._x_event_wake is None:
return
fd = None
if self.xdisplay is not None:
try:
fd = self.xdisplay.fileno()
except Exception:
fd = None
if fd is not None and fd < 0:
fd = None
if fd == self._x_watcher_fd:
return
if self._x_watcher_fd is not None:
try:
self.loop.remove_reader(self._x_watcher_fd)
except Exception:
pass
self._x_watcher_fd = fd
if fd is not None:
try:
self.loop.add_reader(fd, self._x_event_wake.set)
except Exception as e:
logger_webrtc_input.debug(f"X event watcher unavailable ({e}); consumers poll.")
self._x_watcher_fd = NoneparamselfReturns
Nonefunc_disarm_x_event_watcher(self) -> NoneSource Code
def _disarm_x_event_watcher(self) -> None:
if self._x_watcher_fd is not None:
try:
self.loop.remove_reader(self._x_watcher_fd)
except Exception:
pass
self._x_watcher_fd = NoneparamselfReturns
Nonefunc_wait_x_event(self, timeout=1.0) -> NoneSleep until the X socket signals readability or the failsafe elapses. Callers clear the Event BEFORE re-checking event availability, so an event arriving between the check and the wait is never lost. With no reader armed (add_reader unsupported or failed) nothing will ever set the Event, so fall back to a short poll rather than idling out the failsafe.
Source Code
async def _wait_x_event(self, timeout: float = 1.0) -> None:
"""Sleep until the X socket signals readability or the failsafe elapses.
Callers clear the Event BEFORE re-checking event availability, so an
event arriving between the check and the wait is never lost. With no
reader armed (add_reader unsupported or failed) nothing will ever set the
Event, so fall back to a short poll rather than idling out the failsafe."""
wake = self._x_event_wake
if wake is None:
await asyncio.sleep(timeout)
return
if self._x_watcher_fd is None:
await asyncio.sleep(min(timeout, INPUT_X_EVENT_POLL_S))
return
try:
await asyncio.wait_for(wake.wait(), timeout)
except asyncio.TimeoutError:
passparamselfparamtimeoutfloat= 1.0Returns
Nonefunc_reconnect_xdisplay(self) -> NoneRebuild the input X connection after a bounded reply-wait closed it.
Fire-and-forget: the rebuild runs on a worker thread and installs via the loop (_install_reconnected_xdisplay), so xdisplay stays None — and input degraded to the xdotool fallbacks — until the attempt lands; callers just continue and the next X failure retries. The thread is needed because every caller sits on the event loop, and against a server hung under another client's grab — the main condition this reconnect exists to survive — close() and the connection setup would freeze the loop for the whole outage. The handshake itself is bounded so a permanently dead server cannot pin the thread forever.
Source Code
def _reconnect_xdisplay(self) -> None:
"""Rebuild the input X connection after a bounded reply-wait closed it.
Fire-and-forget: the rebuild runs on a worker thread and installs via
the loop (_install_reconnected_xdisplay), so xdisplay stays None — and
input degraded to the xdotool fallbacks — until the attempt lands;
callers just continue and the next X failure retries. The thread is
needed because every caller sits on the event loop, and against a
server hung under another client's grab — the main condition this
reconnect exists to survive — close() and the connection setup would
freeze the loop for the whole outage. The handshake itself is bounded
so a permanently dead server cannot pin the thread forever.
"""
self._disarm_x_event_watcher()
if self._x_reconnect_thread is not None and self._x_reconnect_thread.is_alive():
return
old = self.xdisplay
self.xdisplay = None
self.keyboard = None
self.mouse = None
def _attempt():
try:
if old is not None:
try:
old.close()
except Exception:
pass
disp = display.Display(blocking_timeout=INPUT_X_REPLY_TIMEOUT_S)
except Exception as e:
logger_webrtc_input.error(f"Could not reconnect input X display: {e}")
return
try:
self.loop.call_soon_threadsafe(self._install_reconnected_xdisplay, disp)
except RuntimeError:
try:
disp.close()
except Exception:
pass
self._x_reconnect_thread = threading.Thread(
target=_attempt, name="x-input-reconnect", daemon=True
)
self._x_reconnect_thread.start()paramselfReturns
Nonefunc_install_reconnected_xdisplay(self, disp) -> NoneLoop-side half of _reconnect_xdisplay: wire the fresh connection into every consumer in one step so nothing observes a half-initialized display.
Source Code
def _install_reconnected_xdisplay(self, disp: Any) -> None:
"""Loop-side half of _reconnect_xdisplay: wire the fresh connection into
every consumer in one step so nothing observes a half-initialized display."""
if self.xdisplay is not None:
# An earlier attempt already landed; drop this connection.
try:
disp.close()
except Exception:
pass
return
self.xdisplay = disp
self._apply_input_x_reply_bound()
self._arm_x_event_watcher()
self.__keyboard_connect()
if not self.is_wayland:
self.mouse = _XTestMouse(self.xdisplay)
if self.cursors_running:
try:
screen = self.xdisplay.screen()
self.xdisplay.xfixes_select_cursor_input(
screen.root, xfixes.XFixesDisplayCursorNotifyMask
)
except Exception as e:
logger_webrtc_input.warning(f"Could not re-arm cursor monitor after reconnect: {e}")
logger_webrtc_input.warning("Input X connection was unresponsive; reconnected.")paramselfparamdispAnyReturns
Nonefunc_is_x_conn_closed(self, exc) -> boolTrue if exc is the connection-closed error the reply bound raises.
Source Code
def _is_x_conn_closed(self, exc: BaseException) -> bool:
"""True if `exc` is the connection-closed error the reply bound raises."""
return xlib_error is not None and isinstance(exc, xlib_error.ConnectionClosedError)paramselfparamexcBaseExceptionReturns
boolfunc_load_server_autorepeat_rate(self) -> NoneBest-effort: adopt the X server's configured autorepeat delay/rate for our
synthetic server-side repeat, so a held key feels native. python-xlib in some
builds lacks the XKB controls API, so read it once at connect from xset q
(one-time, not per-event). Any failure or out-of-range value keeps the sane
defaults set in init.
Source Code
async def _load_server_autorepeat_rate(self) -> None:
"""Best-effort: adopt the X server's configured autorepeat delay/rate for our
synthetic server-side repeat, so a held key feels native. python-xlib in some
builds lacks the XKB controls API, so read it once at connect from `xset q`
(one-time, not per-event). Any failure or out-of-range value keeps the sane
defaults set in __init__."""
if self.is_wayland:
return
try:
process = await subprocess.create_subprocess_exec(
"xset", "q", stdout=subprocess.PIPE, stderr=subprocess.PIPE
)
out, _err = await self._communicate_or_kill(process, 0.5, "xset q autorepeat")
if process.returncode != 0 or not out:
return
text = out.decode("utf-8", "replace") if isinstance(out, (bytes, bytearray)) else str(out)
m = re.search(r"auto repeat delay:\s*(\d+)\s*repeat rate:\s*(\d+)", text)
if not m:
return
delay_ms = int(m.group(1))
rate_hz = int(m.group(2))
# Reject nonsense (a 0) that would busy-repeat or never repeat.
if 100 <= delay_ms <= 2000:
self.key_repeat_delay = delay_ms / 1000.0
if 1 <= rate_hz <= 100:
self.key_repeat_interval = 1.0 / rate_hz
logger_webrtc_input.info(
f"Server autorepeat: delay {self.key_repeat_delay:.3f}s, "
f"interval {self.key_repeat_interval:.3f}s."
)
except Exception as e:
logger_webrtc_input.debug(f"Could not read server autorepeat rate (using defaults): {e}")paramselfReturns
Nonefunc__mouse_connect(self) -> NoneSource Code
def __mouse_connect(self) -> None:
if self.uinput_mouse_socket_path:
logger_webrtc_input.info(f"Connecting to uinput mouse socket: {self.uinput_mouse_socket_path}")
self.uinput_mouse_socket = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)
if not self.is_wayland and self.xdisplay:
self.mouse = _XTestMouse(self.xdisplay)paramselfReturns
Nonefunc__mouse_disconnect(self) -> NoneSource Code
def __mouse_disconnect(self) -> None:
if self.mouse: del self.mouse; self.mouse = None
if self.uinput_mouse_socket is not None:
try:
self.uinput_mouse_socket.close()
except OSError:
pass
self.uinput_mouse_socket = NoneparamselfReturns
Nonefunc__mouse_emit(self, *args, **kwargs) -> NoneForward one msgpack-encoded mouse event to the uinput helper socket.
Source Code
def __mouse_emit(self, *args: Any, **kwargs: Any) -> None:
"""Forward one msgpack-encoded mouse event to the uinput helper socket."""
if self.uinput_mouse_socket_path:
cmd = {"args": args, "kwargs": kwargs}
data = msgpack.packb(cmd, use_bin_type=True)
self.uinput_mouse_socket.sendto(data, self.uinput_mouse_socket_path)paramselfparamargsAny= ()paramkwargsAny= {}Returns
Nonefunc__gamepad_connect(self, gamepad_idx, client_name, client_num_btns, client_num_axes, conn_id=None) -> NoneAssociate a client controller with a persistent gamepad slot.
A fresh association starts with no heartbeat, so the previous client's last beat cannot date-stamp it into an immediate sweep, and the kernel device is brought up before the first input so applications see a plug event rather than a controller appearing mid-press.
Source Code
async def __gamepad_connect(self, gamepad_idx: int, client_name: str,
client_num_btns: int, client_num_axes: int,
conn_id: Any = None) -> None:
"""Associate a client controller with a persistent gamepad slot.
A fresh association starts with no heartbeat, so the previous client's
last beat cannot date-stamp it into an immediate sweep, and the kernel
device is brought up before the first input so applications see a
plug event rather than a controller appearing mid-press.
"""
if not (0 <= gamepad_idx < self.num_gamepads):
logger_webrtc_input.error(f"Client association: Gamepad index {gamepad_idx} out of range (0-{self.num_gamepads-1}).")
return
if gamepad_idx not in self.gamepad_instances:
logger_webrtc_input.error(
f"Client association: No persistent gamepad instance found for index {gamepad_idx}. "
f"This should not happen if _initialize_persistent_gamepads ran correctly."
)
return
logger_webrtc_input.info(
f"Client controller '{client_name}' ({client_num_btns}b, {client_num_axes}a) "
f"is now associated with persistent virtual gamepad slot {gamepad_idx}."
)
self.client_gamepad_associations[gamepad_idx] = {
"client_name": client_name,
"client_num_btns": client_num_btns,
"client_num_axes": client_num_axes,
"association_time": time.time(),
"conn_id": conn_id,
}
self.gamepad_heartbeats.pop(gamepad_idx, None)
self.gamepad_instances[gamepad_idx].ensure_uinput()paramselfparamgamepad_idxintparamclient_namestrparamclient_num_btnsintparamclient_num_axesintparamconn_idAny= NoneReturns
Nonefuncrelease_gamepads_for_conn(self, conn_id) -> NoneDisassociate (and neutralize, via reset_state) every gamepad slot whose association was made by this transport connection. This is the ungraceful path — a tab that dies mid-press never sends 'js,d', and only the transport knows the connection is gone.
Source Code
async def release_gamepads_for_conn(self, conn_id: Any) -> None:
"""Disassociate (and neutralize, via reset_state) every gamepad slot whose
association was made by this transport connection. This is the ungraceful
path — a tab that dies mid-press never sends 'js,d', and only the transport
knows the connection is gone."""
if conn_id is None:
return
for idx, info in list(self.client_gamepad_associations.items()):
if info.get("conn_id") == conn_id:
await self.__gamepad_disconnect(idx)paramselfparamconn_idAnyReturns
Nonefunc__gamepad_disconnect(self, gamepad_idx=None) -> NoneDisassociate one slot (or all, with None), releasing anything held.
The release matters for an ungraceful client drop (a tab closed mid-press), which would otherwise leave the in-desktop app with a stuck button or deflected stick until a new client re-sends state.
Source Code
async def __gamepad_disconnect(self, gamepad_idx: Optional[int] = None) -> None:
"""Disassociate one slot (or all, with None), releasing anything held.
The release matters for an ungraceful client drop (a tab closed
mid-press), which would otherwise leave the in-desktop app with a
stuck button or deflected stick until a new client re-sends state.
"""
if gamepad_idx is None:
indices_to_disassociate = list(self.client_gamepad_associations.keys())
logger_webrtc_input.info("Disassociating all client gamepads from persistent slots.")
elif not (0 <= gamepad_idx < self.num_gamepads):
logger_webrtc_input.error(f"Client disassociation: Gamepad index {gamepad_idx} out of range.")
return
else:
indices_to_disassociate = [gamepad_idx]
for idx in indices_to_disassociate:
self.gamepad_heartbeats.pop(idx, None)
if idx in self.client_gamepad_associations:
associated_info = self.client_gamepad_associations.pop(idx)
gamepad = self.gamepad_instances.get(idx)
if gamepad is not None:
gamepad.reset_state()
logger_webrtc_input.info(
f"Client controller '{associated_info.get('client_name', 'Unknown')}' "
f"disassociated from persistent virtual gamepad slot {idx}."
)
elif gamepad_idx is not None:
logger_webrtc_input.warning(
f"Client disassociation: No active client association found for gamepad slot {idx} to disassociate."
)paramselfparamgamepad_idxOptional[int]= NoneReturns
Nonefunc__gamepad_emit_btn(self, gamepad_idx, client_btn_num, client_btn_val) -> NoneSource Code
def __gamepad_emit_btn(self, gamepad_idx: int, client_btn_num: int,
client_btn_val: float) -> None:
gamepad = self.gamepad_instances.get(gamepad_idx)
if gamepad:
gamepad.send_event(client_btn_num, client_btn_val, is_button_event=True)paramselfparamgamepad_idxintparamclient_btn_numintparamclient_btn_valfloatReturns
Nonefunc__gamepad_emit_axis(self, gamepad_idx, client_axis_num, client_axis_val) -> NoneSource Code
def __gamepad_emit_axis(self, gamepad_idx: int, client_axis_num: int,
client_axis_val: float) -> None:
gamepad = self.gamepad_instances.get(gamepad_idx)
if gamepad:
gamepad.send_event(client_axis_num, client_axis_val, is_button_event=False)paramselfparamgamepad_idxintparamclient_axis_numintparamclient_axis_valfloatReturns
Nonefuncconnect(self) -> NoneBring the input backends up: X/Wayland connections, DPI detection, keyboard reset, persistent gamepads, and the background key tasks.
Source Code
async def connect(self) -> None:
"""Bring the input backends up: X/Wayland connections, DPI detection,
keyboard reset, persistent gamepads, and the background key tasks."""
if not self.is_wayland and X11_LIBS_AVAILABLE:
# Bounded handshake: a server hung under another client's grab must
# surface as a failure, not freeze the loop.
try: self.xdisplay = display.Display(blocking_timeout=INPUT_X_REPLY_TIMEOUT_S)
except Exception as e: logger_webrtc_input.error(f"Failed to connect to X display: {e}"); self.xdisplay = None
self._apply_input_x_reply_bound()
if self._x_event_wake is None:
self._x_event_wake = asyncio.Event()
self._arm_x_event_watcher()
if self.xdisplay:
try:
screen = self.xdisplay.screen()
width_mm = screen.width_in_mms
height_mm = screen.height_in_mms
if width_mm > 0 and height_mm > 0:
dpi_x = (screen.width_in_pixels * 25.4) / width_mm
dpi_y = (screen.height_in_pixels * 25.4) / height_mm
self.system_dpi = (dpi_x + dpi_y) / 2.0
dpi_scale_factor = self.system_dpi / 96.0
self.cursor_size_cap = int(self.max_cursor_size * dpi_scale_factor)
logger_webrtc_input.info(
f"System DPI detected as ~{self.system_dpi:.0f}. "
f"Cursor size cap set to {self.cursor_size_cap}x{self.cursor_size_cap}px."
)
except Exception as e:
logger_webrtc_input.warning(f"Could not determine system DPI, using default 96. Error: {e}")
if not self.is_wayland and X11_LIBS_AVAILABLE:
self.__keyboard_connect()
if self.xdisplay:
await self._load_server_autorepeat_rate()
await self.reset_keyboard()
self.__mouse_connect()
await self._initialize_persistent_gamepads()
if self.is_wayland:
self.keyboard_worker_task = asyncio.create_task(self._keyboard_worker())
await self._push_wayland_base_layout()
# After the worker starts: a Lock left engaged by a previous session
# inverts every letter, and the reset rides the fresh worker.
await self.reset_keyboard()
if self.key_sweep_task is None:
self.key_sweep_task = asyncio.create_task(self._key_stale_sweep())
if self.key_repeat_enabled and self.key_repeat_task is None:
self.key_repeat_task = asyncio.create_task(self._key_repeat_loop())
if self.xdisplay is not None and self.keymap_watch_task is None:
self.keymap_watch_task = asyncio.create_task(self._keymap_watch_loop())paramselfReturns
Nonefunc_initialize_persistent_gamepads(self) -> NoneAdopt live process-wide gamepad instances or create and start new ones.
A live instance's sockets are what already-running apps hold open, so a service restart (transport mode switch) reuses them; rebinding would orphan those apps' fds.
Source Code
async def _initialize_persistent_gamepads(self) -> None:
"""Adopt live process-wide gamepad instances or create and start new ones.
A live instance's sockets are what already-running apps hold open, so
a service restart (transport mode switch) reuses them; rebinding would
orphan those apps' fds.
"""
logger_webrtc_input.info(f"Initializing {self.num_gamepads} persistent gamepad instances...")
if not os.path.exists(self.js_socket_path_prefix):
try:
os.makedirs(self.js_socket_path_prefix, exist_ok=True)
logger_webrtc_input.info(f"Created directory for gamepad sockets: {self.js_socket_path_prefix}")
except OSError as e:
logger_webrtc_input.error(f"Failed to create directory {self.js_socket_path_prefix} for gamepad sockets: {e}")
return
for i in range(self.num_gamepads):
if i in self.gamepad_instances:
logger_webrtc_input.warning(f"Gamepad instance for index {i} already exists. Skipping re-initialization.")
continue
existing = _persistent_gamepads.get(i)
if existing is not None and existing.running:
self.gamepad_instances[i] = existing
logger_webrtc_input.info(
f"Adopted live persistent gamepad instance for index {i} (JS: {existing.js_sock_path})."
)
continue
js_ip_sock_path = os.path.join(self.js_socket_path_prefix, f"selkies_js{i}.sock")
evdev_ip_sock_path = os.path.join(self.js_socket_path_prefix, f"selkies_event{1000+i}.sock")
gamepad = SelkiesGamepad(
js_ip_sock_path, evdev_ip_sock_path, self.loop,
uinput_enabled=self.uinput_gamepads,
)
gamepad_name_for_interposer = STANDARD_XPAD_CONFIG.get("name", f"Selkies Virtual Gamepad {i}")
std_num_btns = len(STANDARD_XPAD_CONFIG["btn_map"])
std_num_axes = len(STANDARD_XPAD_CONFIG["axes_map"])
gamepad.set_config(gamepad_name_for_interposer, std_num_btns, std_num_axes)
self._spawn_task(gamepad.run_servers())
_persistent_gamepads[i] = gamepad
self.gamepad_instances[i] = gamepad
logger_webrtc_input.info(f"Initialized and started persistent gamepad instance for index {i} (Name: '{gamepad_name_for_interposer}', JS: {js_ip_sock_path}, EVDEV: {evdev_ip_sock_path}).")paramselfReturns
Nonefuncdisconnect(self) -> NoneTear down this handler's own resources; persistent gamepads stay up.
Only the per-session client-to-slot associations are this handler's (see SelkiesGamepad). The input X connection is closed, not just dropped: the keyboard shim holds the same Display and python-xlib's root-window back-reference makes the graph a cycle with no finalizer, so a dropped connection would keep its X client slot until a cyclic collection happens to run — and every transport switch builds a new handler.
Source Code
async def disconnect(self) -> None:
"""Tear down this handler's own resources; persistent gamepads stay up.
Only the per-session client-to-slot associations are this handler's
(see SelkiesGamepad). The input X connection is closed, not just
dropped: the keyboard shim holds the same Display and python-xlib's
root-window back-reference makes the graph a cycle with no finalizer,
so a dropped connection would keep its X client slot until a cyclic
collection happens to run — and every transport switch builds a new
handler.
"""
logger_webrtc_input.info("Releasing gamepad associations (persistent instances stay up).")
await self.__gamepad_disconnect()
self.gamepad_instances = {}
self.gamepad_heartbeats.clear()
# Before the pointer backends go away, or a held button stays pressed for good.
await self.release_mouse_buttons()
self.__mouse_disconnect()
self._disarm_x_event_watcher()
if self.keyboard is not None:
self.keyboard.release_group_lock()
old_display, self.xdisplay, self.keyboard = self.xdisplay, None, None
if old_display is not None:
try:
await asyncio.to_thread(old_display.close)
except Exception as e:
logger_webrtc_input.debug(f"closing the input X connection failed: {e}")
if self.keyboard_worker_task:
self.keyboard_worker_task.cancel()
self.keyboard_worker_task = None
if self.key_sweep_task:
self.key_sweep_task.cancel()
self.key_sweep_task = None
if self.key_repeat_task:
self.key_repeat_task.cancel()
self.key_repeat_task = None
if self.keymap_watch_task:
self.keymap_watch_task.cancel()
self.keymap_watch_task = None
self.pressed_keys.clear()
self.reaped_atomic_keys.clear()
self.key_repeat_state.clear()
self._reset_multipart_clipboard()paramselfReturns
Nonefunc_key_stale_sweep(self) -> NoneAuto-release keys and neutralize gamepads whose heartbeats stopped, so no input stays stuck held when a key-up is lost to congestion or the client vanishes without a transport close.
On Wayland a release rides the serialized keyboard queue. X11 injection is not queue-serialized, so the modifier/atomic state discard is deferred until after the release and a concurrent kd is checked for on both sides of the await: the keysym is popped first, so a non-None entry means a kd re-pressed it and already injected its own keydown, and the sweep then abandons its release (a second down would double-press) and leaves that kd's state intact.
Source Code
async def _key_stale_sweep(self) -> None:
"""Auto-release keys and neutralize gamepads whose heartbeats stopped,
so no input stays stuck held when a key-up is lost to congestion or the
client vanishes without a transport close.
On Wayland a release rides the serialized keyboard queue. X11
injection is not queue-serialized, so the modifier/atomic state
discard is deferred until after the release and a concurrent kd is
checked for on both sides of the await: the keysym is popped first, so
a non-None entry means a kd re-pressed it and already injected its own
keydown, and the sweep then abandons its release (a second down would
double-press) and leaves that kd's state intact.
"""
try:
while True:
await asyncio.sleep(self.key_sweep_interval)
if not self.pressed_keys and not self.gamepad_heartbeats:
continue
now = time.monotonic()
stale = [k for k, seen in self.pressed_keys.items() if now - seen > self.key_stale_window]
for keysym in stale:
# A heartbeat or re-press during a prior await may have
# refreshed this key.
seen = self.pressed_keys.get(keysym)
if seen is None or time.monotonic() - seen <= self.key_stale_window:
continue
was_atomic = keysym in self.atomically_typed_keys
self.pressed_keys.pop(keysym, None)
self.key_repeat_state.pop(keysym, None)
logger_webrtc_input.warning(f"Auto-releasing key {keysym} (heartbeat lost).")
# An atomically-typed key was never physically held on X11.
if was_atomic and not self.is_wayland:
self.atomically_typed_keys.discard(keysym)
if len(self.reaped_atomic_keys) < self.max_pressed_keys:
self.reaped_atomic_keys.add(keysym)
continue
try:
if self.is_wayland:
self.active_modifiers.discard(keysym)
self.atomically_typed_keys.discard(keysym)
self._keyboard_enqueue(("ku", keysym))
else:
if self.pressed_keys.get(keysym) is not None:
# A kd raced us: it owns the key now.
continue
await self.send_x11_keypress(keysym, down=False)
if self.pressed_keys.get(keysym) is not None:
# A kd raced the keyup await: leave its state intact.
continue
self.active_modifiers.discard(keysym)
self.atomically_typed_keys.discard(keysym)
except Exception as e:
logger_webrtc_input.warning(f"Failed to auto-release key {keysym}: {e}")
for idx, seen in list(self.gamepad_heartbeats.items()):
if now - seen <= self.key_stale_window:
continue
self.gamepad_heartbeats.pop(idx, None)
gamepad = self.gamepad_instances.get(idx)
if gamepad is not None and gamepad._held_controls:
logger_webrtc_input.warning(
f"Neutralizing gamepad slot {idx} (heartbeat lost).")
gamepad.reset_state()
except asyncio.CancelledError:
passparamselfReturns
Nonefunc_key_repeat_loop(self) -> NoneX11 server-side key auto-repeat for held keys.
XTEST/xdotool synthetic presses don't trigger the X server's native auto-repeat, so without this a held key types a single character. We re-emit the most-recently-pressed still-held repeatable key at key_repeat_interval after an initial key_repeat_delay -- exactly like a physical keyboard, where only the last key pressed repeats and releasing it resumes the previously-held one. Modifiers are never armed, so the repeated key carries whatever modifiers are currently held (Shift+Arrow selection, Ctrl+Backspace word-delete, Ctrl+Z, etc. all repeat like native -- no special shortcut suppression); atomically-typed keys (digits/punctuation) are armed and repeat through the atomic path below. Repeats are KeyPress-only (no synthetic KeyRelease), matching X11 detectable auto-repeat, so state-based games keep the key held with no movement stutter and ignore the extra presses. Wayland is excluded (the focused app repeats held virtual-keyboard keys itself via wl_keyboard repeat_info, so a server-side repeat would double it).
key_repeat_state is insertion-ordered and arming moves a key to the end, so its last entry is the newest held key. An atomic key repeats as a self-contained XTEST press+release at its shift level (the 'ku' path injects no key-up for atomic keys, so a lone press would stick), falling to the co,end path when it has no keycode in the layout. Repeat pauses while the key's heartbeats have stopped (stalled stream, hidden tab), bounding run-on to the grace rather than the stale window.
Source Code
async def _key_repeat_loop(self) -> None:
"""X11 server-side key auto-repeat for held keys.
XTEST/xdotool synthetic presses don't trigger the X server's native
auto-repeat, so without this a held key types a single character. We re-emit the
most-recently-pressed still-held repeatable key at key_repeat_interval after an
initial key_repeat_delay -- exactly like a physical keyboard, where only the last
key pressed repeats and releasing it resumes the previously-held one. Modifiers
are never armed, so the repeated key carries whatever modifiers are currently
held (Shift+Arrow selection, Ctrl+Backspace word-delete, Ctrl+Z, etc. all repeat
like native -- no special shortcut suppression); atomically-typed keys
(digits/punctuation) are armed and repeat through the atomic path below.
Repeats are KeyPress-only (no synthetic KeyRelease),
matching X11 detectable auto-repeat, so state-based games keep the key held with
no movement stutter and ignore the extra presses. Wayland is excluded (the
focused app repeats held virtual-keyboard keys itself via wl_keyboard
repeat_info, so a server-side repeat would double it).
key_repeat_state is insertion-ordered and arming moves a key to the
end, so its last entry is the newest held key. An atomic key repeats
as a self-contained XTEST press+release at its shift level (the 'ku'
path injects no key-up for atomic keys, so a lone press would stick),
falling to the co,end path when it has no keycode in the layout.
Repeat pauses while the key's heartbeats have stopped (stalled stream,
hidden tab), bounding run-on to the grace rather than the stale window.
"""
try:
while True:
await asyncio.sleep(self.key_repeat_tick)
if not self.key_repeat_state:
continue
# Keys released/reaped since arming: never inject a down after the key-up.
for k in [k for k in self.key_repeat_state if k not in self.pressed_keys]:
self.key_repeat_state.pop(k, None)
if not self.key_repeat_state:
continue
keysym = next(reversed(self.key_repeat_state))
now = time.monotonic()
if now < self.key_repeat_state[keysym]:
continue
last_seen = self.pressed_keys.get(keysym)
if last_seen is None or (now - last_seen) > self.key_repeat_heartbeat_grace:
continue
try:
if keysym in self.atomically_typed_keys:
injected = False
if self.keyboard is not None:
try:
self.keyboard.press(
keysym,
held_keysyms=(self.active_modifiers
& self.LEVEL_MODIFIER_KEYSYMS))
self.keyboard.release(keysym)
injected = True
except Exception as e:
logger_webrtc_input.debug(
f"XTEST atomic repeat failed for keysym {keysym}; falling back: {e}"
)
if self._is_x_conn_closed(e):
self._reconnect_xdisplay()
if not injected:
unicode_codepoint = (keysym & 0x00FFFFFF
if (keysym & 0xFF000000) == 0x01000000 else keysym)
char_to_type = chr(unicode_codepoint)
await self.on_message(f"co,end,{char_to_type}")
else:
await self.send_x11_keypress(keysym, down=True)
except Exception as e:
logger_webrtc_input.warning(f"Key auto-repeat failed for {keysym}: {e}")
self.key_repeat_state.pop(keysym, None)
continue
# A 'ku' during the await released the key; the extra down is
# healed by the real key-up / stale sweep.
if keysym in self.pressed_keys:
self.key_repeat_state[keysym] = time.monotonic() + self.key_repeat_interval
else:
self.key_repeat_state.pop(keysym, None)
except asyncio.CancelledError:
passparamselfReturns
Nonefuncreset_keyboard(self) -> NoneRelease every held key/modifier and normalize a stuck Caps Lock.
Runs on client 'kr' (blur/visibility loss), at connect, and when a client holding input departs, on both backends: the client resolves letter case itself, so an engaged Lock modifier on the server would invert every letter it types. On Wayland every key goes through the serialized keyboard worker, so the reset is queued behind the key work in flight like a client 'kr' is (a press still queued behind a text batch would otherwise land after the reset, untracked and held for good); it runs in place only when no worker is draining the queue.
On X11 every still-held key is released, not just the modifier and hotkey list: once pressed_keys is cleared the stale sweep (the only other releaser) never runs for them, so a held 'w' in a game would stay pressed forever. Atomically-typed keys were never physically held and are skipped to avoid a spurious keyup; translated (Cyrillic) keys go through send_x11_keypress's translation so the injected QWERTY key comes up. Caps Lock is not forwarded from the browser, but a prior session or the desktop's own startup can leave Lock engaged, so it is toggled back off.
Source Code
async def reset_keyboard(self) -> None:
"""Release every held key/modifier and normalize a stuck Caps Lock.
Runs on client 'kr' (blur/visibility loss), at connect, and when a
client holding input departs, on both backends: the client resolves
letter case itself, so an engaged Lock modifier on the server would
invert every letter it types. On Wayland every key goes through the
serialized keyboard worker, so the reset is queued behind the key work
in flight like a client 'kr' is (a press still queued behind a text
batch would otherwise land after the reset, untracked and held for
good); it runs in place only when no worker is draining the queue.
On X11 every still-held key is released, not just the modifier and
hotkey list: once pressed_keys is cleared the stale sweep (the only
other releaser) never runs for them, so a held 'w' in a game would
stay pressed forever. Atomically-typed keys were never physically
held and are skipped to avoid a spurious keyup; translated (Cyrillic)
keys go through send_x11_keypress's translation so the injected QWERTY
key comes up. Caps Lock is not forwarded from the browser, but a prior
session or the desktop's own startup can leave Lock engaged, so it is
toggled back off.
"""
if self.is_wayland:
worker = self.keyboard_worker_task
if (worker is None or worker.done()
or asyncio.current_task() is worker):
await self._reset_keyboard_wayland()
return
done = asyncio.get_running_loop().create_future()
self._keyboard_enqueue(("kr", done))
await asyncio.wait({done, worker}, timeout=self._WL_RESET_WAIT_S,
return_when=asyncio.FIRST_COMPLETED)
if done.done():
return
if worker.done():
# The worker died with the reset queued; nothing else injects now.
await self._reset_keyboard_wayland()
else:
logger_webrtc_input.warning(
"Keyboard reset still queued behind pending key work.")
return
if not self.keyboard or not self.xdisplay :
logger_webrtc_input.warning("Cannot reset keyboard, X display or keyboard controller not available.")
return
logger_webrtc_input.info("Resetting keyboard modifiers.")
lctrl, lshift, lalt, altgr = 65507, 65505, 65513, 65027
rctrl, rshift, ralt = 65508, 65506, 65514
lmeta, rmeta, keyf, keyF, keym, keyM, escape = 65511, 65512, 102, 70, 109, 77, 65307
# Super/Hyper included: the client maps the Meta/Windows key to Super.
lsuper, rsuper, lhyper, rhyper = 65515, 65516, 65517, 65518
for k in [lctrl, lshift, lalt, altgr, rctrl, rshift, ralt, lmeta, rmeta,
lsuper, rsuper, lhyper, rhyper, keyf, keyF, keym, keyM, escape]:
try: await self.send_x11_keypress(k, down=False)
except Exception as e: logger_webrtc_input.warning(f"Error resetting key {k}: {e}")
for keysym in list(self.pressed_keys):
if keysym in self.atomically_typed_keys:
continue
try: await self.send_x11_keypress(keysym, down=False)
except Exception as e: logger_webrtc_input.warning(f"Error releasing held key {keysym}: {e}")
for k in list(self.translated_keys):
try: await self.send_x11_keypress(k, down=False)
except Exception as e: logger_webrtc_input.warning(f"Error releasing translated key {k}: {e}")
try:
if self.xdisplay.screen().root.query_pointer().mask & Xlib.X.LockMask:
caps_kc = self.xdisplay.keysym_to_keycode(0xffe5)
if caps_kc:
xtest.fake_input(self.xdisplay, Xlib.X.KeyPress, caps_kc)
xtest.fake_input(self.xdisplay, Xlib.X.KeyRelease, caps_kc)
self.xdisplay.flush()
except Exception as e:
logger_webrtc_input.warning(f"Could not normalize Lock modifier: {e}")
# Every key is up, so a group lock of ours goes back at once, not after its linger.
if self.keyboard is not None:
self.keyboard.release_group_lock()
# After the release loops, which consume this state.
self.active_modifiers.clear()
self.active_shortcut_modifiers.clear()
self.atomically_typed_keys.clear()
self.translated_keys.clear()
# Or the stale sweep would auto-release keys the reset just cleared.
self.pressed_keys.clear()
self.reaped_atomic_keys.clear()
self.key_repeat_state.clear()paramselfReturns
Nonefunc_reset_keyboard_wayland(self) -> NoneThe Wayland reset itself — see reset_keyboard for when it runs and why every held and translated key is released before the maps clear.
Caps Lock is read back through get_keyboard_state — a compositor round-trip answered by the thread that also renders, so it waits off the loop — and toggled off with a virtual Caps_Lock press+release.
Source Code
async def _reset_keyboard_wayland(self) -> None:
"""The Wayland reset itself — see reset_keyboard for when it runs and
why every held and translated key is released before the maps clear.
Caps Lock is read back through get_keyboard_state — a compositor
round-trip answered by the thread that also renders, so it waits off
the loop — and toggled off with a virtual Caps_Lock press+release.
"""
if self.wayland_input:
# Ctrl, Shift, Alt, AltGr, Meta, Super, Hyper (the client maps the
# Meta/Windows key to Super).
modifiers = [65507, 65505, 65513, 65508, 65506, 65514, 65027,
65511, 65512, 65515, 65516, 65517, 65518]
# A stale owner (rebuild backing off) still knows what is down.
owner = await self._ensure_wayland_keymap_owner() or self._wl_keymap_owner
if owner is not None:
for k in modifiers:
try:
owner.release(k)
except Exception:
pass
try:
owner.reset()
except Exception:
pass
for k in list(self.translated_keys):
try: await self.send_x11_keypress(k, down=False)
except Exception: pass
for keysym in list(self.pressed_keys):
if keysym in self.atomically_typed_keys:
continue
try: await self.send_x11_keypress(keysym, down=False)
except Exception as e: logger_webrtc_input.warning(f"Error releasing held key {keysym}: {e}")
self.active_modifiers.clear()
self.active_shortcut_modifiers.clear()
self.atomically_typed_keys.clear()
self.translated_keys.clear()
# Or the stale sweep would auto-release keys the reset just cleared.
self.pressed_keys.clear()
self.reaped_atomic_keys.clear()
try:
if self.wayland_input and hasattr(self.wayland_input, 'get_keyboard_state'):
_pressed, mods = await asyncio.to_thread(self.wayland_input.get_keyboard_state)
# mods bit 4 is caps_lock.
if mods & 0x10:
keymap_owner = await self._ensure_wayland_keymap_owner()
if keymap_owner is not None:
keymap_owner.press(0xffe5)
await asyncio.sleep(0.02)
keymap_owner.release(0xffe5)
except Exception as e:
logger_webrtc_input.debug(f"Wayland caps-lock normalization skipped: {e}")paramselfReturns
Nonefuncrelease_mouse_buttons(self) -> NoneRelease every pointer button still held server-side.
The ungraceful pointer path, mirroring release_gamepads_for_conn and the key stale-sweep: a client that dies mid-drag never sends the mask with the button cleared, so the desktop would keep the drag or selection alive until some future client happens to diff the bit away. The releases ride the normal mask-diff loop, so X11 and Wayland both heal; the zero-delta relative move leaves the pointer exactly where it is (and injects no motion at all on Wayland).
Source Code
async def release_mouse_buttons(self) -> None:
"""Release every pointer button still held server-side.
The ungraceful pointer path, mirroring release_gamepads_for_conn and the
key stale-sweep: a client that dies mid-drag never sends the mask with the
button cleared, so the desktop would keep the drag or selection alive until
some future client happens to diff the bit away. The releases ride the
normal mask-diff loop, so X11 and Wayland both heal; the zero-delta
relative move leaves the pointer exactly where it is (and injects no motion
at all on Wayland)."""
if not self.button_mask:
return
try:
await self.send_x11_mouse(0, 0, 0, 0, relative=True)
except Exception as e:
logger_webrtc_input.warning(f"Failed to release held mouse buttons: {e}")paramselfReturns
Nonefuncsend_mouse(self, action, data) -> NoneRoute one MOUSE_* action to the uinput socket or XTEST backend.
Source Code
def send_mouse(self, action: int, data: Any) -> None:
"""Route one MOUSE_* action to the uinput socket or XTEST backend.
Args:
action: A MOUSE_* constant.
data: Action-dependent — an (x, y) pair for position/move, a
(press/release, button-id) pair for MOUSE_BUTTON, None for
scroll actions.
"""
if action == MOUSE_POSITION:
if self.mouse: self.mouse.position = data
elif action == MOUSE_MOVE:
x, y = data
if self.uinput_mouse_socket_path:
self.__mouse_emit(UINPUT_REL_X, x, syn=False)
self.__mouse_emit(UINPUT_REL_Y, y)
elif self.xdisplay:
xtest.fake_input(self.xdisplay, Xlib.X.MotionNotify, detail=True, root=Xlib.X.NONE, x=x, y=y)
# flush(), not sync(): XTEST needs no reply and sync() is a round trip per move.
self.xdisplay.flush()
elif action == MOUSE_SCROLL_UP:
# MOUSE_SCROLL_* are named for the client button, not the physical
# direction: this is wheel-down (X button 5), so REL_WHEEL is -1.
if self.uinput_mouse_socket_path: self.__mouse_emit(UINPUT_REL_WHEEL, -1)
elif self.mouse: self.mouse.scroll(0, -1)
elif action == MOUSE_SCROLL_DOWN:
if self.uinput_mouse_socket_path: self.__mouse_emit(UINPUT_REL_WHEEL, 1)
elif self.mouse: self.mouse.scroll(0, 1)
elif action == MOUSE_SCROLL_LEFT:
# REL_HWHEEL is signed as the client names it (negative = left), like
# X buttons 6/7, so no flip here.
if self.uinput_mouse_socket_path: self.__mouse_emit(UINPUT_REL_HWHEEL, -1)
elif self.mouse: self.mouse.scroll(-1, 0)
elif action == MOUSE_SCROLL_RIGHT:
if self.uinput_mouse_socket_path: self.__mouse_emit(UINPUT_REL_HWHEEL, 1)
elif self.mouse: self.mouse.scroll(1, 0)
elif action == MOUSE_BUTTON:
btn_map_key = "uinput" if self.uinput_mouse_socket_path else "x11"
btn_uinput_or_x11 = MOUSE_BUTTON_MAP[data[1]][btn_map_key]
if data[0] == MOUSE_BUTTON_PRESS:
if self.uinput_mouse_socket_path: self.__mouse_emit(btn_uinput_or_x11, 1)
elif self.mouse: self.mouse.press(btn_uinput_or_x11)
else:
if self.uinput_mouse_socket_path: self.__mouse_emit(btn_uinput_or_x11, 0)
elif self.mouse: self.mouse.release(btn_uinput_or_x11)paramselfparamactionintA MOUSE_* constant.
paramdataAnyAction-dependent — an (x, y) pair for position/move, a (press/release, button-id) pair for MOUSE_BUTTON, None for scroll actions.
Returns
Nonefuncsend_x11_keypress(self, keysym, down=True, neutralize=None) -> NoneInject one key transition on whichever backend this session uses.
Despite the name this is the shared key injector: Wayland routes through the seat keymap owner (virtual-keyboard/clipboard on error), X11 through XTEST with xdotool fallbacks. Cyrillic keysyms chorded with an action modifier are translated to the QWERTY keysym on the same physical key so shortcuts (Ctrl+C on a ЙЦУКЕН layout) reach the application as the app expects.
On X11 a keysym with a keycode in the current keymap is injected through XTEST on the already-open display, which spares a ~15 ms xdotool fork per shortcut, arrow or function key; a keysym the layout lacks is overlay-bound once by the shim and reused, and one only a later layout group carries is injected under that group's lock — never a per-key xdotool fork, whose transient rebind floods MappingNotify and lags the whole input queue behind real typing. xdotool remains for a keysym with no keycode at all that it can still synthesize.
Source Code
async def send_x11_keypress(self, keysym: int, down: bool = True,
neutralize: Optional[bool] = None) -> None:
"""Inject one key transition on whichever backend this session uses.
Despite the name this is the shared key injector: Wayland routes
through the seat keymap owner (virtual-keyboard/clipboard on error),
X11 through XTEST with xdotool fallbacks. Cyrillic keysyms chorded
with an action modifier are translated to the QWERTY keysym on the
same physical key so shortcuts (Ctrl+C on a ЙЦУКЕН layout) reach the
application as the app expects.
On X11 a keysym with a keycode in the current keymap is injected
through XTEST on the already-open display, which spares a ~15 ms
xdotool fork per shortcut, arrow or function key; a keysym the layout
lacks is overlay-bound once by the shim and reused, and one only a
later layout group carries is injected under that group's lock —
never a per-key xdotool fork, whose transient rebind floods
MappingNotify and lags the whole input queue behind real typing.
xdotool remains for a keysym with no keycode at all that it can still
synthesize.
Args:
neutralize: Whether a conflicting held Shift/AltGr is lifted around
the key. None derives it from the client's held modifiers:
lifted around plain keystrokes only, since while a chord
modifier (Ctrl/Alt/Super/...) is down every held modifier is
part of the chord. A server-synthesized chord passes False,
since a Shift it pressed through this injector is not in
active_modifiers and would be lifted for the very key it
modifies.
"""
if down:
if (self.active_modifiers & self.ACTION_MODIFIER_KEYSYMS) and keysym in CYRILLIC_TO_QWERTY_KEYSYM:
self.translated_keys.add(keysym)
keysym = CYRILLIC_TO_QWERTY_KEYSYM[keysym]
else:
if keysym in self.translated_keys:
self.translated_keys.discard(keysym)
keysym = CYRILLIC_TO_QWERTY_KEYSYM[keysym]
if neutralize is None:
neutralize = (keysym not in self.MODIFIER_KEYSYMS
and not (self.active_modifiers & self.ACTION_MODIFIER_KEYSYMS))
held_level_mods = frozenset(self.active_modifiers & self.LEVEL_MODIFIER_KEYSYMS)
if self.is_wayland and self.wayland_input:
owner = await self._ensure_wayland_keymap_owner()
if owner is not None:
try:
if down:
owner.press(keysym, neutralize=neutralize)
else:
owner.release(keysym)
return
except Exception as e:
logger_webrtc_input.warning(
f"Wayland keymap injection failed for keysym {keysym}; falling back: {e}"
)
await self._type_keysym_fallback(keysym, down)
return
is_printable = (0x20 <= keysym <= 0xFF) or ((keysym & 0xFF000000) == 0x01000000)
action = "keydown" if down else "keyup"
command = None
use_keyboard_for_printable = False
allow_xtest = False
if is_printable:
unicode_codepoint = keysym & 0x00FFFFFF if (keysym & 0xFF000000) == 0x01000000 else keysym
try:
char = chr(unicode_codepoint)
if char.isalpha():
use_keyboard_for_printable = True
else:
xdotool_arg = f"U{unicode_codepoint:04X}"
if not self.active_shortcut_modifiers:
use_keyboard_for_printable = True
else:
command = ["xdotool", action, xdotool_arg]
allow_xtest = True
except ValueError:
use_keyboard_for_printable = True
else:
map_entry = X11_KEYSYM_MAP.get(keysym)
if map_entry:
xdotool_arg = map_entry.get('xkey_name')
if xdotool_arg:
command = ["xdotool", action, xdotool_arg]
allow_xtest = True
if xdotool_arg in self.SHORTCUT_MODIFIER_XKEY_NAMES:
if down:
self.active_shortcut_modifiers.add(xdotool_arg)
else:
self.active_shortcut_modifiers.discard(xdotool_arg)
if command:
if allow_xtest and xtest is not None and self.xdisplay is not None:
try:
keycode = self.xdisplay.keysym_to_keycode(keysym)
if self.keyboard and self.keyboard.outside_base_group(keysym):
# A bare keycode would type the group-1 glyph; the shim
# locks the group around the injection instead.
keycode = 0
if keycode:
xtest.fake_input(
self.xdisplay,
X.KeyPress if down else X.KeyRelease,
keycode,
)
self.xdisplay.flush()
return
if self.keyboard:
if down:
self.keyboard.press(keysym, neutralize=neutralize,
held_keysyms=held_level_mods)
else:
self.keyboard.release(keysym)
return
except Exception as e:
logger_webrtc_input.debug(
f"XTEST inject failed for keysym {keysym}; falling back to xdotool: {e}"
)
try:
process = await subprocess.create_subprocess_exec(
*command, stdout=subprocess.PIPE, stderr=subprocess.PIPE
)
await self._communicate_or_kill(process, 0.5, "xdotool key")
if process.returncode == 0:
return
logger_webrtc_input.warning(
f"xdotool {action} failed (rc={process.returncode}) for keysym {keysym}")
except Exception:
pass
if use_keyboard_for_printable or not command:
try:
if not self.keyboard:
await self._type_keysym_fallback(keysym, down)
return
if down:
self.keyboard.press(keysym, neutralize=neutralize,
held_keysyms=held_level_mods)
else:
self.keyboard.release(keysym)
except Exception as e:
if self._is_x_conn_closed(e):
self._reconnect_xdisplay()
await self._type_keysym_fallback(keysym, down)paramselfparamkeysymintparamdownbool= TrueparamneutralizeOptional[bool]= NoneWhether a conflicting held Shift/AltGr is lifted around the key. None derives it from the client's held modifiers: lifted around plain keystrokes only, since while a chord modifier (Ctrl/Alt/Super/...) is down every held modifier is part of the chord. A server-synthesized chord passes False, since a Shift it pressed through this injector is not in active_modifiers and would be lifted for the very key it modifies.
Returns
Nonefunc_type_text_xtest(self, text, neutralize=False) -> boolType a string in-process via the XTEST shim.
Each char is a press+release of its keysym (mapped chars with shift synthesis, unmapped ones via the spare-keycode overlay). A char prefers its canonical layout keysym, as the Wayland owner's type_text does, so a layout that carries the script types on its own keys (Cyrillic on ru, or on us,ru under a group lock) instead of spending overlay slots on every letter. With neutralize, conflicting held Shift/AltGr are lifted around the whole run (one keymap query, not one per char). Unmapped chars are bound in one batch (O(1) MappingNotify broadcasts instead of one per char), and nothing is typed on failure.
Source Code
def _type_text_xtest(self, text: str, neutralize: bool = False) -> bool:
"""Type a string in-process via the XTEST shim.
Each char is a press+release of its keysym (mapped chars with shift
synthesis, unmapped ones via the spare-keycode overlay). A char prefers
its canonical layout keysym, as the Wayland owner's type_text does, so
a layout that carries the script types on its own keys (Cyrillic on
ru, or on us,ru under a group lock) instead of spending overlay slots
on every letter. With neutralize, conflicting held Shift/AltGr are
lifted around the whole run (one keymap query, not one per char).
Unmapped chars are bound in one batch (O(1) MappingNotify broadcasts
instead of one per char), and nothing is typed on failure.
Returns:
True on full success; False (having typed nothing) if the shim is
unavailable or any char can't be resolved, so the caller can fall
back to xdotool without double-typing.
"""
if not self.keyboard or not text:
return False
# Pre-resolve every char so a mid-string failure types no partial line.
keysyms = []
for ch in text:
ks = character_to_layout_keysym(ch)
if not self.keyboard.layout_carries(ks):
cp = ord(ch)
ks = cp if 0x20 <= cp <= 0xFF else (0x01000000 | cp)
keysyms.append(ks)
try:
if not self.keyboard.prebind(keysyms):
return False
lifted = []
if neutralize:
down = self.keyboard._down_mod_keycodes(
self.active_modifiers & self.LEVEL_MODIFIER_KEYSYMS)
lifted = self.keyboard._mods_to_lift(set(), down)
for m in lifted:
xtest.fake_input(self.keyboard._d, Xlib.X.KeyRelease, m)
try:
for ks in keysyms:
self.keyboard.press(ks)
self.keyboard.release(ks)
finally:
for m in reversed(lifted):
xtest.fake_input(self.keyboard._d, Xlib.X.KeyPress, m)
if lifted:
self.keyboard._d.flush()
return True
except Exception as e:
logger_webrtc_input.debug(f"in-process type failed ({e}); falling back to xdotool")
return Falseparamselfparamtextstrparamneutralizebool= FalseReturns
boolTrue on full success; False (having typed nothing) if the shim is
func_spawn_task(self, coro, name=None) -> asyncio.Taskcreate_task with a keep-alive reference and error logging.
Source Code
def _spawn_task(self, coro: Any, name: Optional[str] = None) -> asyncio.Task:
"""create_task with a keep-alive reference and error logging."""
task = asyncio.create_task(coro, name=name)
self._bg_tasks.add(task)
def _done(t):
self._bg_tasks.discard(t)
if not t.cancelled() and t.exception() is not None:
logger_webrtc_input.error(f"Background task {t.get_name()} failed: {t.exception()}")
task.add_done_callback(_done)
return taskparamselfparamcoroAnyparamnameOptional[str]= NoneReturns
asyncio.asyncio.Taskfunc_invalidate_wayland_keymap_owner(self) -> NoneThe seat's base layout changed: the owner's keycode resolution is against the old base, so it is rebuilt — on the keyboard worker, where nothing injects concurrently — carrying what it holds down.
Source Code
def _invalidate_wayland_keymap_owner(self) -> None:
"""The seat's base layout changed: the owner's keycode resolution is
against the old base, so it is rebuilt — on the keyboard worker, where
nothing injects concurrently — carrying what it holds down."""
self._wl_keymap_stale = True
self._wl_keymap_retry_at = 0.0paramselfReturns
Nonefunc_push_wayland_base_layout(self, restore=False) -> NoneSet the compositor seat's BASE xkb layout from the deployment's XKB
env config (XKB_DEFAULT_LAYOUT et al.): at session start, so common
non-US keysyms resolve as base keys instead of overlay binds, and again
(restore) when a nested session compositor turns up after a client
hint moved the seat off that layout — the nested compositor translates
keycodes with the keymap it built from the same env, so the seat must
resolve keysyms against it; with no env layout the xkbcommon default is
what both run on. The compositor re-splices its overlay binds on top
with unchanged keycodes; the keymap owner rebuilds from the new base.
Source Code
async def _push_wayland_base_layout(self, restore: bool = False) -> None:
"""Set the compositor seat's BASE xkb layout from the deployment's XKB
env config (XKB_DEFAULT_LAYOUT et al.): at session start, so common
non-US keysyms resolve as base keys instead of overlay binds, and again
(`restore`) when a nested session compositor turns up after a client
hint moved the seat off that layout — the nested compositor translates
keycodes with the keymap it built from the same env, so the seat must
resolve keysyms against it; with no env layout the xkbcommon default is
what both run on. The compositor re-splices its overlay binds on top
with unchanged keycodes; the keymap owner rebuilds from the new base."""
layout = os.environ.get("XKB_DEFAULT_LAYOUT", "")
setter = getattr(self.wayland_input, 'set_xkb_layout', None)
if setter is None or not (layout or restore):
return
variant = os.environ.get("XKB_DEFAULT_VARIANT", "")
options = os.environ.get("XKB_DEFAULT_OPTIONS", "")
model = os.environ.get("XKB_DEFAULT_MODEL", "")
rules = os.environ.get("XKB_DEFAULT_RULES", "")
try:
ok = await asyncio.to_thread(
setter, layout, variant, options, model, rules)
except Exception as e:
logger_webrtc_input.warning(f"Wayland base layout push failed: {e}")
return
if ok:
self._wl_seat_client_layout = None
self._invalidate_wayland_keymap_owner()
logger_webrtc_input.info(
f"Wayland base layout set to '{layout or 'default'}'"
+ (f" ({variant})" if variant else "")
+ (" (restored for the session compositor)" if restore else ""))
else:
logger_webrtc_input.warning(
f"Wayland base layout '{layout}' rejected by the compositor; "
"keeping the default.")paramselfparamrestorebool= FalseReturns
Nonefunc_schedule_seat_layout_restore(self) -> NoneA nested session compositor was just adopted while the seat carries a client layout hint: put the seat back on the deployment layout the session translates keycodes with.
Source Code
def _schedule_seat_layout_restore(self) -> None:
"""A nested session compositor was just adopted while the seat carries
a client layout hint: put the seat back on the deployment layout the
session translates keycodes with."""
if getattr(self, "_wl_seat_client_layout", None) is None:
return
try:
asyncio.get_running_loop()
except RuntimeError:
return
self._spawn_task(self._push_wayland_base_layout(restore=True))paramselfReturns
Nonefuncapply_client_keyboard_layout(self, layout_hint) -> NoneClient SETTINGS 'keyboardLayout' hint ("de", "ch(fr)", ...). On pixelflux Wayland with the apps on the capture compositor it becomes the seat's BASE xkb layout (set_xkb_layout), so the client's physical layout resolves as base keys instead of per-keysym overlay binds. Under a nested session compositor it is informational, as on X11: the session translates keycodes with its own keymap, so a seat moved to the client layout transposes every key the two layouts place apart. Idempotent per value — clients re-assert SETTINGS on reconnects and broadcasts — and a hint only noted under a session compositor is applied once the apps are back on the capture compositor.
Source Code
async def apply_client_keyboard_layout(self, layout_hint: Any) -> None:
"""Client SETTINGS 'keyboardLayout' hint ("de", "ch(fr)", ...). On
pixelflux Wayland with the apps on the capture compositor it becomes
the seat's BASE xkb layout (set_xkb_layout), so the client's physical
layout resolves as base keys instead of per-keysym overlay binds.
Under a nested session compositor it is informational, as on X11: the
session translates keycodes with its own keymap, so a seat moved to
the client layout transposes every key the two layouts place apart.
Idempotent per value — clients re-assert SETTINGS on reconnects and
broadcasts — and a hint only noted under a session compositor is
applied once the apps are back on the capture compositor."""
hint = str(layout_hint or "").strip()
m = re.fullmatch(r"([A-Za-z0-9_,\- ]{1,32})(?:\(([A-Za-z0-9_,\- ]{1,32})\))?", hint)
if not m:
if hint:
logger_webrtc_input.warning(
f"Ignoring malformed keyboardLayout hint: {hint[:48]!r}")
return
layout, variant = m.group(1).strip(), (m.group(2) or "").strip()
noted = hint == self._client_kb_layout
self._client_kb_layout = hint
if not self.is_wayland:
if not noted:
logger_webrtc_input.info(
f"Client keyboard layout hint '{hint}' noted (X11 keymap is "
"deployment-owned; not applied).")
return
if self._has_separate_app_compositor():
if not noted:
logger_webrtc_input.info(
f"Client keyboard layout hint '{hint}' noted (the session "
"compositor owns the keymap; not applied).")
return
if hint == self._wl_seat_client_layout:
return
setter = getattr(self.wayland_input, 'set_xkb_layout', None) if self.wayland_input else None
if setter is None:
logger_webrtc_input.warning(
f"Client keyboard layout '{hint}' not applied: compositor "
"keymap control unavailable.")
return
try:
ok = await asyncio.to_thread(setter, layout, variant, "", "", "")
except Exception as e:
logger_webrtc_input.warning(f"Wayland base layout push failed: {e}")
return
if ok:
self._wl_seat_client_layout = hint
self._invalidate_wayland_keymap_owner()
logger_webrtc_input.info(
f"Wayland base layout set to '{layout}'"
+ (f" ({variant})" if variant else "") + " from client hint")
else:
logger_webrtc_input.warning(
f"Wayland base layout '{hint}' rejected by the compositor; "
"keeping the current layout.")paramselfparamlayout_hintAnyReturns
Nonefunc_ensure_wayland_keymap_owner(self) -> Optional[_WaylandKeymapOwner]Get-or-build the keymap owner; a stale one (base layout changed) is rebuilt the same way, adopting the keys it holds down. Reading the compositor keymap blocks (bounded) and compiling it costs milliseconds, so both run off the loop; after that, press/release are sync dict work
- channel sends. None while a failed build backs off (callers fall to the next injection rung).
Source Code
async def _ensure_wayland_keymap_owner(self) -> Optional[_WaylandKeymapOwner]:
"""Get-or-build the keymap owner; a stale one (base layout changed) is
rebuilt the same way, adopting the keys it holds down. Reading the
compositor keymap blocks (bounded) and compiling it costs milliseconds,
so both run off the loop; after that, press/release are sync dict work
+ channel sends. None while a failed build backs off (callers fall to
the next injection rung)."""
if self._wl_keymap_owner is not None and not self._wl_keymap_stale:
return self._wl_keymap_owner
if not hasattr(self.wayland_input, 'set_keymap_string'):
return None
now = time.monotonic()
if now < self._wl_keymap_retry_at:
return None
async with self._wl_keymap_owner_lock:
if self._wl_keymap_owner is not None and not self._wl_keymap_stale:
return self._wl_keymap_owner
loop = asyncio.get_running_loop()
previous = self._wl_keymap_owner
try:
def _build():
text = self.wayland_input.get_xkb_keymap_string()
owner = _WaylandKeymapOwner(self.wayland_input, text)
if previous is not None:
owner.adopt_held(previous)
return owner
self._wl_keymap_owner = await loop.run_in_executor(None, _build)
self._wl_keymap_stale = False
logger_webrtc_input.info(
f"Wayland keymap owner ready ({len(self._wl_keymap_owner._map)} keysyms"
+ (f", {len(previous._pressed)} held keys carried)" if previous is not None
else ")"))
except Exception as e:
self._wl_keymap_retry_at = time.monotonic() + 5.0
logger_webrtc_input.warning(
f"Wayland keymap owner unavailable ({e}); retrying in 5s.")
return None
return self._wl_keymap_ownerparamselfReturns
typing.Optional[selkies.input_handler._WaylandKeymapOwner]func_wl_type_text(self, text) -> NoneInject text through the app compositor's zwp_virtual_keyboard_manager_v1 via pixelflux's one-shot in-process client: the first fallback rung under the seat keymap, above the clipboard paste. Raises when the compositor lacks the protocol or the injection fails, so callers can drop to the next rung the same way they would on any injection error. A failure against an auto-detected app compositor also drops the cached socket, so the next attempt re-detects a compositor that died or restarted under another name rather than aiming at a dead socket.
Source Code
async def _wl_type_text(self, text: str) -> None:
"""Inject text through the app compositor's zwp_virtual_keyboard_manager_v1
via pixelflux's one-shot in-process client: the first fallback rung under
the seat keymap, above the clipboard paste. Raises when the compositor
lacks the protocol or the injection fails, so callers can drop to the
next rung the same way they would on any injection error. A failure
against an auto-detected app compositor also drops the cached socket,
so the next attempt re-detects a compositor that died or restarted
under another name rather than aiming at a dead socket."""
async with self._wl_typer_lock:
if time.monotonic() < self._wl_typer_retry_at:
raise RuntimeError(
"compositor does not advertise zwp_virtual_keyboard_manager_v1"
" (retry pending)")
display = self._app_wayland_display()
try:
typer = getattr(self.wayland_input, 'type_keysyms_wayland', None)
if typer is not None:
await asyncio.get_running_loop().run_in_executor(
None, typer, display, text_to_wayland_keysyms(text))
else:
await asyncio.get_running_loop().run_in_executor(
None, self.wayland_input.type_text_wayland, display, text)
except PixelfluxVkUnavailable:
self._wl_typer_retry_at = time.monotonic() + 30.0
raise
except Exception:
if self._app_wl_is_separate:
self._invalidate_app_wl_display()
raiseparamselfparamtextstrReturns
Nonefunc_inject_text_via_clipboard(self, text) -> boolType text by replacing the clipboard with it, pasting via
Shift+Insert, and restoring what was copied before. The route for
compositors with no zwp_virtual_keyboard (KWin): the chord's two
keysyms exist in every base layout, so this needs nothing beyond the
data-control clipboard and ordinary key events. Held modifiers are
lifted around the chord so they cannot corrupt the paste, and
write_clipboard's baseline keeps the monitor from echoing the injected
text back to clients. Returns True once the paste chord is sent.
Source Code
async def _inject_text_via_clipboard(self, text: str) -> bool:
"""Type `text` by replacing the clipboard with it, pasting via
Shift+Insert, and restoring what was copied before. The route for
compositors with no zwp_virtual_keyboard (KWin): the chord's two
keysyms exist in every base layout, so this needs nothing beyond the
data-control clipboard and ordinary key events. Held modifiers are
lifted around the chord so they cannot corrupt the paste, and
write_clipboard's baseline keeps the monitor from echoing the injected
text back to clients. Returns True once the paste chord is sent."""
if self._clipboard_inject_active:
return False
async with self._clipboard_inject_lock:
self._clipboard_inject_active = True
shift_keysym = 0xFFE1
insert_keysym = 0xFF63
held_modifiers = list(self.active_modifiers)
try:
for mod_keysym in held_modifiers:
await self.send_x11_keypress(mod_keysym, down=False)
old_data, old_mime = await self.read_clipboard(use_binary=True)
if not await self.write_clipboard(text):
return False
# The sleeps are the focused app's margins to receive the new
# offer and to finish fetching it before the selection is restored.
await asyncio.sleep(0.02)
await self.send_x11_keypress(shift_keysym, down=True)
await self.send_x11_keypress(insert_keysym, down=True, neutralize=False)
await self.send_x11_keypress(insert_keysym, down=False)
await self.send_x11_keypress(shift_keysym, down=False)
await asyncio.sleep(0.05)
if old_data is not None:
await self.write_clipboard(old_data, old_mime or "text/plain")
elif self.is_wayland:
await self._clear_injected_clipboard()
return True
except Exception as e:
logger_webrtc_input.error(f"Clipboard text injection failed: {e}")
return False
finally:
for mod_keysym in held_modifiers:
if mod_keysym in self.active_modifiers:
await self.send_x11_keypress(mod_keysym, down=True)
self._clipboard_inject_active = FalseparamselfparamtextstrReturns
boolfunc_clear_injected_clipboard(self) -> NoneDrop the selection the injection left behind when there was nothing to restore (an empty clipboard stays empty for the user).
Source Code
async def _clear_injected_clipboard(self) -> None:
"""Drop the selection the injection left behind when there was nothing
to restore (an empty clipboard stays empty for the user)."""
try:
if self._has_separate_app_compositor():
clear_fn = getattr(self.wayland_input, 'clipboard_clear_app', None)
if clear_fn is not None:
await asyncio.get_running_loop().run_in_executor(
None, clear_fn, self._app_wayland_display())
else:
self.wayland_input.set_clipboard("text/plain", b"")
except Exception as e:
logger_webrtc_input.debug(f"post-injection clipboard clear failed: {e}")paramselfReturns
Nonefunc_type_keysym_fallback(self, keysym_number, down=True) -> NoneDeliver a keysym the primary injector could not, resolving the newest mechanism first and degrading rung by rung. Wayland is subprocess-free: the keysym becomes text and goes through the in-process virtual-keyboard client, then the clipboard paste. X11 falls from the in-process XTEST shim to an xdotool key, then an xdotool type of the plain character.
Source Code
async def _type_keysym_fallback(self, keysym_number: int, down: bool = True) -> None:
"""Deliver a keysym the primary injector could not, resolving the newest
mechanism first and degrading rung by rung. Wayland is subprocess-free:
the keysym becomes text and goes through the in-process virtual-keyboard
client, then the clipboard paste. X11 falls from the in-process XTEST
shim to an xdotool key, then an xdotool type of the plain character."""
if self.is_wayland:
if not down:
return
char_to_type = keysym_to_character(keysym_number)
if char_to_type:
try:
await self._wl_type_text(char_to_type)
except Exception as e:
if not await self._inject_text_via_clipboard(char_to_type):
logger_webrtc_input.warning(f"virtual-keyboard fallback failed: {e}")
return
if not self.xdisplay:
return
xdotool_key_arg = None
char_for_type_cmd_fallback = None
keysym_name_from_xlib = None
if (keysym_number & 0xFF000000) == 0x01000000:
unicode_codepoint = keysym_number & 0x00FFFFFF
if 0 <= unicode_codepoint <= 0x10FFFF:
xdotool_key_arg = f"U{unicode_codepoint:04X}"
try:
char_for_type_cmd_fallback = chr(unicode_codepoint)
except ValueError:
pass
else:
return
else:
keysym_name_from_xlib = XK.keysym_to_string(keysym_number)
if keysym_name_from_xlib is None:
# Decoded, not chr()'d: a keysym is a codepoint only in Latin-1,
# and chr() typed Gujarati for the publishing block.
char = keysym_to_character(keysym_number)
if char is None:
return
keysym_name_from_xlib = char
char_for_type_cmd_fallback = char
else:
if len(keysym_name_from_xlib) == 1:
char_for_type_cmd_fallback = keysym_name_from_xlib
xdotool_key_arg = keysym_name_from_xlib
if len(keysym_name_from_xlib) == 1:
char_code = ord(keysym_name_from_xlib)
if char_code >= 0x80 or (char_code == keysym_number and char_code != 0x00):
xdotool_key_arg = f"U{char_code:04X}"
# XK_sterling.
elif keysym_number == 0x00a3:
xdotool_key_arg = "sterling"
if not char_for_type_cmd_fallback:
try: char_for_type_cmd_fallback = chr(0xA3)
except ValueError: pass
if xdotool_key_arg is None:
return
action = "keydown" if down else "keyup"
command_key = ["xdotool", action, xdotool_key_arg]
try:
process_key = await subprocess.create_subprocess_exec(
*command_key,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE
)
stdout_key, stderr_key = await self._communicate_or_kill(process_key, 1.0, "xdotool keydown")
if process_key.returncode != 0 or (stderr_key and (b"No such key name" in stderr_key or b"Error:" in stderr_key.lower())):
char_to_type = char_for_type_cmd_fallback
if not char_to_type and keysym_name_from_xlib and len(keysym_name_from_xlib) == 1:
char_to_type = keysym_name_from_xlib
if down and char_to_type and (0x20 <= ord(char_to_type) <= 0x7E or ord(char_to_type) >= 0xA0) and char_to_type.isprintable():
command_type = ["xdotool", "type", "--clearmodifiers", "--", char_to_type]
try:
process_type = await subprocess.create_subprocess_exec(
*command_type,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE
)
await self._communicate_or_kill(process_type, 1.0, "xdotool type")
except (asyncio.TimeoutError, FileNotFoundError, Exception):
pass
except (FileNotFoundError, asyncio.TimeoutError, Exception):
passparamselfparamkeysym_numberintparamdownbool= TrueReturns
Nonefuncsend_x11_mouse(self, x, y, button_mask, scroll_magnitude, relative=False, display_id='primary') -> NoneApply one client pointer message on whichever backend this session uses.
Moves the pointer (absolute coordinates are offset into the named display's region of the combined layout), then diffs button_mask against the held state bit by bit, emitting press/release, scroll clicks, or the Alt+Arrow back/forward chords.
A delta is injected as a delta; the tracked position follows it only so that an absolute message has something to compare against, and it is bounded by the laid-out screen because the X server and the compositor bound the pointer the same way. The tracking stays an estimate even so — an application can move the pointer, and the framebuffer can be a few pixels wider than the layout — so the first absolute position after any delta warps unconditionally rather than trusting that comparison. Where no relative injection exists (a capture backend without it), the bound matters directly: the estimate is what gets injected.
Source Code
async def send_x11_mouse(self, x: int, y: int, button_mask: int,
scroll_magnitude: int, relative: bool = False,
display_id: str = 'primary') -> None:
"""Apply one client pointer message on whichever backend this session uses.
Moves the pointer (absolute coordinates are offset into the named
display's region of the combined layout), then diffs button_mask
against the held state bit by bit, emitting press/release, scroll
clicks, or the Alt+Arrow back/forward chords.
A delta is injected as a delta; the tracked position follows it only so
that an absolute message has something to compare against, and it is
bounded by the laid-out screen because the X server and the compositor
bound the pointer the same way. The tracking stays an estimate even so
— an application can move the pointer, and the framebuffer can be a few
pixels wider than the layout — so the first absolute position after any
delta warps unconditionally rather than trusting that comparison. Where
no relative injection exists (a capture backend without it), the bound
matters directly: the estimate is what gets injected.
Args:
x: Absolute X, or the X delta when relative.
y: Absolute Y, or the Y delta when relative.
button_mask: Client button bitmask (Pointer Events numbering). The
eraser bit (button 5) folds into the primary button: neither
the X core pointer nor wl_pointer has an eraser button, so the
eraser clicks and drags like the pen tip, and OR-ing it keeps
press/release balanced when the tip bit is set at the same time.
scroll_magnitude: Wheel repeat count; 0 marks bits 3/4 as
back/forward buttons instead of wheel ticks.
relative: Interpret x/y as deltas.
display_id: Display whose layout offset absolute coordinates use.
"""
# Client-controlled; unbounded, the X11 scroll loop would block the event loop.
try:
scroll_magnitude = max(0, min(int(scroll_magnitude), 64))
except (TypeError, ValueError):
scroll_magnitude = 0
if button_mask & MOUSE_MASK_BIT_ERASER:
button_mask = (button_mask | MOUSE_MASK_BIT_PRIMARY) & ~MOUSE_MASK_BIT_ERASER
was_stale = self.tracked_position_stale
if relative:
# XTEST carries the delta in an Int16; an overlarge value would fail
# the request and take the button transitions on this message with it.
x = max(-32768, min(32767, x))
y = max(-32768, min(32767, y))
final_x = self.last_x + x
final_y = self.last_y + y
edge_x, edge_y = 0, 0
if self.data_server_instance and hasattr(self.data_server_instance, 'display_layouts'):
edge_x, edge_y = layout_extent(self.data_server_instance.display_layouts)
if edge_x > 0:
final_x = max(0, min(final_x, edge_x - 1))
if edge_y > 0:
final_y = max(0, min(final_y, edge_y - 1))
self.tracked_position_stale = True
else:
offset_x = 0
offset_y = 0
if self.data_server_instance and hasattr(self.data_server_instance, 'display_layouts'):
# A socket with no registered display (shared viewer, input
# handoff) renders the primary, whose offset is non-zero in
# left/up arrangements.
lookup_id = display_id or 'primary'
layout = self.data_server_instance.display_layouts.get(lookup_id)
if layout:
offset_x = layout.get('x', 0)
offset_y = layout.get('y', 0)
elif lookup_id != 'primary':
# A secondary with no laid-out region must not inject at
# offset zero (its clicks would land on the primary); a held
# button self-heals on the next mask diff.
return
final_x = x + offset_x
final_y = y + offset_y
self.tracked_position_stale = False
position_changed = (was_stale or final_x != self.last_x or final_y != self.last_y)
self.last_x = final_x
self.last_y = final_y
is_static_relative = relative and x == 0 and y == 0
if self.wayland_input:
if not is_static_relative:
if relative:
if hasattr(self.wayland_input, 'inject_relative_mouse_move'):
self.wayland_input.inject_relative_mouse_move(float(x), float(y))
else:
self.wayland_input.inject_mouse_move(float(final_x), float(final_y))
else:
self.wayland_input.inject_mouse_move(float(final_x), float(final_y))
if button_mask != self.button_mask:
for bit_index in range(8):
current_button_bit_value = (1 << bit_index)
button_state_changed = ((self.button_mask & current_button_bit_value) != \
(button_mask & current_button_bit_value))
if button_state_changed:
is_pressed_now = (button_mask & current_button_bit_value) != 0
state = 1 if is_pressed_now else 0
mag = float(max(1, scroll_magnitude))
# evdev BTN_LEFT, BTN_MIDDLE, BTN_RIGHT.
if bit_index == 0:
self.wayland_input.inject_mouse_button(272, state)
elif bit_index == 1:
self.wayland_input.inject_mouse_button(274, state)
elif bit_index == 2:
self.wayland_input.inject_mouse_button(273, state)
elif bit_index == 3:
if scroll_magnitude > 0:
if is_pressed_now:
self.wayland_input.inject_mouse_scroll(0.0, 10.0 * mag)
else:
if is_pressed_now:
# Queued behind pending keys like any key event:
# direct injection could land between a kd and its ku.
self._keyboard_enqueue_chord((
(KEYSYM_ALT_L, True), (KEYSYM_LEFT_ARROW, True),
(KEYSYM_LEFT_ARROW, False), (KEYSYM_ALT_L, False)))
elif bit_index == 4:
if scroll_magnitude > 0:
if is_pressed_now:
self.wayland_input.inject_mouse_scroll(0.0, -10.0 * mag)
else:
if is_pressed_now:
self._keyboard_enqueue_chord((
(KEYSYM_ALT_L, True), (KEYSYM_RIGHT_ARROW, True),
(KEYSYM_RIGHT_ARROW, False), (KEYSYM_ALT_L, False)))
elif bit_index == 6:
if scroll_magnitude > 0 and is_pressed_now:
self.wayland_input.inject_mouse_scroll(-10.0 * mag, 0.0)
elif bit_index == 7:
if scroll_magnitude > 0 and is_pressed_now:
self.wayland_input.inject_mouse_scroll(10.0 * mag, 0.0)
self.button_mask = button_mask
return
if relative:
if not is_static_relative:
self.send_mouse(MOUSE_MOVE, (x, y))
elif position_changed or button_mask != self.button_mask:
# Button transitions warp unconditionally: an application may have
# moved the pointer, and a press must land where the client aims.
self.send_mouse(MOUSE_POSITION, (final_x, final_y))
if button_mask != self.button_mask:
for bit_index in range(8):
current_button_bit_value = (1 << bit_index)
button_state_changed = ((self.button_mask & current_button_bit_value) != \
(button_mask & current_button_bit_value))
if button_state_changed:
is_pressed_now = (button_mask & current_button_bit_value) != 0
action_to_send = None
data_to_send = None
is_scroll_action = False
performed_keyboard_combo = False
if bit_index == 0:
action_to_send = MOUSE_BUTTON
data_to_send = (MOUSE_BUTTON_PRESS if is_pressed_now else MOUSE_BUTTON_RELEASE, MOUSE_BUTTON_LEFT_ID)
elif bit_index == 1:
action_to_send = MOUSE_BUTTON
data_to_send = (MOUSE_BUTTON_PRESS if is_pressed_now else MOUSE_BUTTON_RELEASE, MOUSE_BUTTON_MIDDLE_ID)
elif bit_index == 2:
action_to_send = MOUSE_BUTTON
data_to_send = (MOUSE_BUTTON_PRESS if is_pressed_now else MOUSE_BUTTON_RELEASE, MOUSE_BUTTON_RIGHT_ID)
elif bit_index == 3:
if scroll_magnitude > 0:
if is_pressed_now:
action_to_send = MOUSE_SCROLL_UP
is_scroll_action = True
else:
if is_pressed_now:
if self.keyboard:
logger_webrtc_input.debug("Sending Alt+Left Arrow for Back")
await self.send_x11_keypress(KEYSYM_ALT_L, down=True)
await self.send_x11_keypress(KEYSYM_LEFT_ARROW, down=True)
await self.send_x11_keypress(KEYSYM_LEFT_ARROW, down=False)
await self.send_x11_keypress(KEYSYM_ALT_L, down=False)
performed_keyboard_combo = True
else:
logger_webrtc_input.warning("Keyboard not available for Alt+Left.")
elif bit_index == 4:
if scroll_magnitude > 0:
if is_pressed_now:
action_to_send = MOUSE_SCROLL_DOWN
is_scroll_action = True
else:
if is_pressed_now:
if self.keyboard:
logger_webrtc_input.debug("Sending Alt+Right Arrow for Forward")
await self.send_x11_keypress(KEYSYM_ALT_L, down=True)
await self.send_x11_keypress(KEYSYM_RIGHT_ARROW, down=True)
await self.send_x11_keypress(KEYSYM_RIGHT_ARROW, down=False)
await self.send_x11_keypress(KEYSYM_ALT_L, down=False)
performed_keyboard_combo = True
else:
logger_webrtc_input.warning("Keyboard not available for Alt+Right.")
elif bit_index == 6:
if scroll_magnitude > 0 and is_pressed_now:
action_to_send = MOUSE_SCROLL_LEFT
is_scroll_action = True
elif bit_index == 7:
if scroll_magnitude > 0 and is_pressed_now:
action_to_send = MOUSE_SCROLL_RIGHT
is_scroll_action = True
if not performed_keyboard_combo and action_to_send is not None:
if is_scroll_action:
for _ in range(max(1, scroll_magnitude)):
self.send_mouse(action_to_send, None)
else:
self.send_mouse(action_to_send, data_to_send)
self.button_mask = button_mask
if not relative and self.xdisplay:
# flush(), not sync(): a round trip per mouse event otherwise.
self.xdisplay.flush()paramselfparamxintAbsolute X, or the X delta when relative.
paramyintAbsolute Y, or the Y delta when relative.
parambutton_maskintClient button bitmask (Pointer Events numbering). The eraser bit (button 5) folds into the primary button: neither the X core pointer nor wl_pointer has an eraser button, so the eraser clicks and drags like the pen tip, and OR-ing it keeps press/release balanced when the tip bit is set at the same time.
paramscroll_magnitudeintWheel repeat count; 0 marks bits 3/4 as back/forward buttons instead of wheel ticks.
paramrelativebool= FalseInterpret x/y as deltas.
paramdisplay_idstr= 'primary'Display whose layout offset absolute coordinates use.
Returns
Nonefuncupdate_binary_clipboard_setting(self, enabled) -> NoneUpdate the binary clipboard setting and restart the monitor if it is running.
Source Code
async def update_binary_clipboard_setting(self, enabled: bool) -> None:
"""Update the binary clipboard setting and restart the monitor if it is running."""
async with self._binary_clipboard_lock:
new_setting_str = "true" if enabled else "false"
if self.enable_binary_clipboard == new_setting_str:
return
logger_webrtc_input.info(f"Binary clipboard setting changing to: {enabled}. Restarting monitor.")
self.enable_binary_clipboard = new_setting_str
if self.clipboard_monitor_task and not self.clipboard_monitor_task.done():
self.stop_clipboard()
self.clipboard_monitor_task.cancel()
try:
await self.clipboard_monitor_task
except asyncio.CancelledError:
pass
self.clipboard_monitor_task = asyncio.create_task(self.start_clipboard())paramselfparamenabledboolReturns
Nonefunc_wayland_display_name(self) -> strThe compositor's REAL socket name. The pixelflux compositor auto-picks the first free wayland-N socket, so the running backend is authoritative; the process env is next (stream_server mirrors the name there at bring-up) and --wayland-socket-index survives only as a legacy hint.
Source Code
def _wayland_display_name(self) -> str:
"""The compositor's REAL socket name. The pixelflux compositor auto-picks
the first free wayland-N socket, so the running backend is authoritative;
the process env is next (stream_server mirrors the name there at bring-up)
and --wayland-socket-index survives only as a legacy hint."""
try:
from pixelflux import get_wayland_display_name
name = get_wayland_display_name()
if name:
return name
except Exception:
pass
return (os.environ.get("WAYLAND_DISPLAY")
or f"wayland-{self.wayland_socket_index}")paramselfReturns
strfunc_app_wayland_display(self) -> strSocket of the compositor applications run under — the target for input injection and clipboard. It equals the capture compositor for a plain pixelflux session, but a nested session that pixelflux captures owns the apps on its own socket, so input and clipboard aimed at the capture compositor never reach them. Resolution order: the explicit app_wayland_display setting; else the single other wayland-* socket in XDG_RUNTIME_DIR besides the capture compositor's; else the capture socket. A distinct result is cached permanently; the capture fallback is negative- cached with a short TTL, so a nested compositor that appears after startup is still picked up within a couple seconds without relisting per call.
Source Code
def _app_wayland_display(self) -> str:
"""Socket of the compositor applications run under — the target for input
injection and clipboard. It equals the capture compositor for a plain
pixelflux session, but a nested session that pixelflux
captures owns the apps on its own socket, so input and clipboard aimed at
the capture compositor never reach them. Resolution order: the explicit
app_wayland_display setting; else the single other wayland-* socket in
XDG_RUNTIME_DIR besides the capture compositor's; else the capture socket.
A distinct result is cached permanently; the capture fallback is negative-
cached with a short TTL, so a nested compositor that appears after startup
is still picked up within a couple seconds without relisting per call."""
if self._app_wl_display_cached is not None:
return self._app_wl_display_cached
now = time.monotonic()
if (self._app_wl_negcache is not None
and (now - self._app_wl_negcache_at) < 2.0):
return self._app_wl_negcache
self._app_wl_negcache_at = now
capture = self._wayland_display_name()
override = (self.app_wayland_display or "").strip()
if override:
return self._adopt_app_wl_display(override, capture, "configured")
resolved = None
try:
import stat as _stat
runtime = os.environ.get("XDG_RUNTIME_DIR")
cap_base = os.path.basename(capture)
if runtime and os.path.isdir(runtime):
others = sorted(
n for n in os.listdir(runtime)
if n.startswith("wayland-") and not n.endswith(".lock")
and n != cap_base
and _stat.S_ISSOCK(os.stat(os.path.join(runtime, n)).st_mode))
# A stale wayland-* file left by a dead compositor must not be
# adopted; it would silently break clipboard and input.
others = [n for n in others
if self._wl_socket_live(os.path.join(runtime, n))]
if len(others) == 1:
resolved = others[0]
elif len(others) > 1:
# wayland-<N> is a compositor; a differently named socket is a
# relay a session listens on (see the module docstring).
numbered = [n for n in others
if n[len("wayland-"):].isdigit()]
if len(numbered) == 1:
resolved = numbered[0]
else:
logger_webrtc_input.warning(
"Multiple candidate app-compositor sockets %s; set "
"app_wayland_display to choose. Using capture compositor.",
others)
except Exception as e:
logger_webrtc_input.debug(f"App-compositor autodetect failed: {e}")
if resolved and resolved != capture:
return self._adopt_app_wl_display(resolved, capture, "auto-detected")
self._app_wl_negcache = capture
return captureparamselfReturns
strfunc_wl_socket_live(path) -> boolTrue if a Wayland socket accepts a connection right now (rejecting a stale socket file with no listener).
Source Code
@staticmethod
def _wl_socket_live(path: str) -> bool:
"""True if a Wayland socket accepts a connection right now (rejecting a
stale socket file with no listener)."""
try:
s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
s.settimeout(0.2)
try:
s.connect(path)
return True
finally:
s.close()
except OSError:
return FalseparampathstrReturns
boolfunc_adopt_app_wl_display(self, resolved, capture, how) -> strCache the resolved app compositor and, when it is distinct from the capture compositor, hand it to pixelflux over the Python ABI so its Computer-Use backend targets the same session. pixelflux keeps its own PIXELFLUX_APP_WAYLAND_DISPLAY env fallback for standalone use without selkies.
Source Code
def _adopt_app_wl_display(self, resolved: str, capture: str, how: str) -> str:
"""Cache the resolved app compositor and, when it is distinct from the
capture compositor, hand it to pixelflux over the Python ABI so its
Computer-Use backend targets the same session. pixelflux keeps its own
PIXELFLUX_APP_WAYLAND_DISPLAY env fallback for standalone use without
selkies."""
self._app_wl_display_cached = resolved
self._app_wl_negcache = None
self._app_wl_is_separate = resolved != capture
if self._app_wl_is_separate:
logger_webrtc_input.info(
f"Wayland app compositor '{resolved}' ({how}); routing input + "
f"clipboard there, capture stays on '{capture}'.")
try:
self.wayland_input.set_app_wayland_display(resolved)
except Exception as e:
logger_webrtc_input.debug(
f"pixelflux set_app_wayland_display failed: {e}")
self._schedule_session_scale()
self._schedule_spare_screen_hold()
self._schedule_seat_layout_restore()
return resolvedparamselfparamresolvedstrparamcapturestrparamhowstrReturns
strfuncapp_session(self) -> dictWhere the session's applications run: x11_display (the X server
they connect to, if any), wayland_display (the compositor socket,
when they are Wayland clients) and type ("x11" or "wayland").
X11 backend: the server's own DISPLAY. Wayland backend: apps under a nested session compositor use its socket and the Xwayland it spawned (the one live X server; $DISPLAY when that is live); with no nested compositor a rootful Xwayland on $DISPLAY hosts an X11 desktop — its apps are X clients and must not be offered the capture compositor's socket, which would take them out of the desktop as fullscreen toplevels. The server on $DISPLAY counts only when it is that Xwayland (not a leftover Xvfb/Xorg holding the number); otherwise the apps are Wayland clients of the capture compositor itself.
Source Code
def app_session(self) -> dict:
"""Where the session's applications run: ``x11_display`` (the X server
they connect to, if any), ``wayland_display`` (the compositor socket,
when they are Wayland clients) and ``type`` ("x11" or "wayland").
X11 backend: the server's own DISPLAY. Wayland backend: apps under a
nested session compositor use its socket and the Xwayland it spawned (the
one live X server; $DISPLAY when that is live); with no nested compositor
a rootful Xwayland on $DISPLAY hosts an X11 desktop — its apps are X
clients and must not be offered the capture compositor's socket, which
would take them out of the desktop as fullscreen toplevels. The server
on $DISPLAY counts only when it is that Xwayland (not a leftover
Xvfb/Xorg holding the number); otherwise the apps are Wayland clients of
the capture compositor itself.
"""
env_display = os.environ.get("DISPLAY", "").strip() or None
if not self.is_wayland:
return {"x11_display": env_display, "wayland_display": None, "type": "x11"}
wayland = self._app_wayland_display()
if self._has_separate_app_compositor():
live = live_x_displays()
x11 = env_display if env_display in live else (live[0] if live else None)
return {"x11_display": x11, "wayland_display": wayland, "type": "wayland"}
if env_display and x_display_live(env_display) and x_display_is_xwayland(env_display):
return {"x11_display": env_display, "wayland_display": None, "type": "x11"}
return {"x11_display": None, "wayland_display": wayland, "type": "wayland"}paramselfReturns
dictfuncapp_launch_env(self) -> dictEnvironment for a client-requested command: the server's, with DISPLAY / WAYLAND_DISPLAY / XDG_SESSION_TYPE set for the session the applications run in (app_session) and the session bus plus desktop identity adopted from that session's processes when the server has none of its own. The adopted subset is cached per session and refreshed when its bus stops answering; an empty scan is negative-cached briefly so a command burst on a session with no bus does not rescan /proc (a synchronous walk on the loop) each time.
Source Code
def app_launch_env(self) -> dict:
"""Environment for a client-requested command: the server's, with
DISPLAY / WAYLAND_DISPLAY / XDG_SESSION_TYPE set for the session the
applications run in (app_session) and the session bus plus desktop
identity adopted from that session's processes when the server has
none of its own. The adopted subset is cached per session and refreshed
when its bus stops answering; an empty scan is negative-cached briefly
so a command burst on a session with no bus does not rescan /proc (a
synchronous walk on the loop) each time."""
session = self.app_session()
env = dict(os.environ)
for key in ("DISPLAY", "WAYLAND_DISPLAY"):
env.pop(key, None)
if session["x11_display"]:
env["DISPLAY"] = session["x11_display"]
if session["wayland_display"]:
env["WAYLAND_DISPLAY"] = session["wayland_display"]
env["XDG_SESSION_TYPE"] = session["type"]
key = (session["x11_display"], session["wayland_display"])
adopted = self._session_env_cache.get(key)
if adopted is None or not dbus_address_live(
adopted.get("DBUS_SESSION_BUS_ADDRESS", "")):
now = time.monotonic()
if adopted is None and (now - self._session_env_empty_at.get(key, -1e9)
) < self._session_env_negcache_ttl:
adopted = {}
else:
adopted = session_environment(
session["x11_display"], session["wayland_display"])
if adopted:
self._session_env_cache[key] = adopted
self._session_env_empty_at.pop(key, None)
logger_webrtc_input.info(
f"Application launches adopt the session on {key}: "
f"{', '.join(sorted(adopted))}")
else:
self._session_env_cache.pop(key, None)
self._session_env_empty_at[key] = now
for name, value in adopted.items():
env.setdefault(name, value)
return envparamselfReturns
dictfuncapp_terminal(self) -> Optional[str]Terminal command prefix clients launch proot-apps under (the
terminal plus its run-this flag, e.g. xterm -e): the first installed
one for the windowing system the session's applications use (foot on a
Wayland session, st on an X11 one), published as app_terminal.
Source Code
def app_terminal(self) -> Optional[str]:
"""Terminal command prefix clients launch proot-apps under (the
terminal plus its run-this flag, e.g. `xterm -e`): the first installed
one for the windowing system the session's applications use (foot on a
Wayland session, st on an X11 one), published as app_terminal."""
session = self.app_session()
return first_installed(WAYLAND_APP_TERMINALS if session["type"] == "wayland"
else X11_APP_TERMINALS)paramselfReturns
typing.Optional[str]funcapps_available(self) -> boolWhether the apps panel can do anything in this session.
Its buttons are the runner wrapper over the command channel, so all of the channel, the wrapper and an environment the wrapper can work in have to hold; where one does not, dashboards are told to drop the panel rather than offer buttons that cannot succeed. The environment answer comes from probe_apps_runner() and is assumed good until that probe has spoken, so a working session never flickers the panel.
Source Code
def apps_available(self) -> bool:
"""Whether the apps panel can do anything in this session.
Its buttons are the runner wrapper over the command channel, so all of
the channel, the wrapper and an environment the wrapper can work in
have to hold; where one does not, dashboards are told to drop the
panel rather than offer buttons that cannot succeed. The environment
answer comes from probe_apps_runner() and is assumed good until that
probe has spoken, so a working session never flickers the panel.
"""
if not settings.command_enabled[0]:
return False
if shutil.which(APP_RUNNER) is None:
return False
return self._apps_runner_ok is not FalseparamselfReturns
boolfuncprobe_apps_runner(self) -> NoneAsk the runner whether apps can install and launch here, once.
proot emulates its chroot by tracing every process it starts, so a host that denies ptrace (Yama restricted, no CAP_SYS_PTRACE) leaves the panel with nothing that can work. Only the wrapper knows how to decide that for the runner it ships, so it is asked rather than reimplemented.
Source Code
async def probe_apps_runner(self) -> None:
"""Ask the runner whether apps can install and launch here, once.
proot emulates its chroot by tracing every process it starts, so a
host that denies ptrace (Yama restricted, no CAP_SYS_PTRACE) leaves the
panel with nothing that can work. Only the wrapper knows how to decide
that for the runner it ships, so it is asked rather than reimplemented.
"""
runner = shutil.which(APP_RUNNER)
if runner is None:
return
try:
proc = await subprocess.create_subprocess_exec(
runner, "check",
stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
stdout, _ = await self._communicate_or_kill(
proc, APP_RUNNER_CHECK_TIMEOUT_S, f"{APP_RUNNER} check")
self._apps_runner_ok = proc.returncode == 0
except Exception as e:
logger_webrtc_input.warning(f"{APP_RUNNER} check failed to run: {e}")
self._apps_runner_ok = False
return
if self._apps_runner_ok:
logger_webrtc_input.info("Apps panel enabled: %s can run here.", APP_RUNNER)
else:
detail = (stdout or b"").decode("utf-8", "replace").strip().splitlines()
logger_webrtc_input.warning(
"Apps panel hidden: %s cannot run here. %s",
APP_RUNNER, detail[0] if detail else "")
for line in detail[1:]:
logger_webrtc_input.warning(" %s", line.strip())paramselfReturns
Nonefunc_invalidate_app_wl_display(self) -> NoneDrop the cached app-compositor resolution so the next call re-detects it — used when a connection to it fails (a nested compositor that died or restarted on a different socket name).
Source Code
def _invalidate_app_wl_display(self) -> None:
"""Drop the cached app-compositor resolution so the next call re-detects
it — used when a connection to it fails (a nested compositor that died or
restarted on a different socket name)."""
if self._app_wl_display_cached:
try:
self.wayland_input.clipboard_unwatch_app(self._app_wl_display_cached)
except Exception:
pass
self._app_wl_display_cached = None
self._app_wl_is_separate = False
self._app_wl_negcache = None
self._app_wl_negcache_at = 0.0paramselfReturns
Nonefunc_invalidate_app_wl_display_if_dead(self, display) -> NoneRe-detect the app compositor when its socket no longer answers (a nested compositor that died or restarted under another name); a live socket that merely failed one request is kept.
Source Code
def _invalidate_app_wl_display_if_dead(self, display: str) -> None:
"""Re-detect the app compositor when its socket no longer answers (a
nested compositor that died or restarted under another name); a live
socket that merely failed one request is kept."""
if not display:
return
runtime = os.environ.get("XDG_RUNTIME_DIR") or ""
path = display if display.startswith("/") else os.path.join(runtime, display)
if not self._wl_socket_live(path):
logger_webrtc_input.info(
f"Wayland app compositor '{display}' no longer answers; re-detecting.")
self._invalidate_app_wl_display()paramselfparamdisplaystrReturns
Nonefunc_has_separate_app_compositor(self) -> boolTrue when apps live under a compositor distinct from pixelflux's own, so pixelflux's keymap overlay and its selection never reach them. Resolves (throttled) then reads the cached flag, so it costs no per-call FFI once settled.
Source Code
def _has_separate_app_compositor(self) -> bool:
"""True when apps live under a compositor distinct from pixelflux's own,
so pixelflux's keymap overlay and its selection never reach them. Resolves
(throttled) then reads the cached flag, so it costs no per-call FFI once
settled."""
self._app_wayland_display()
return self._app_wl_is_separateparamselfReturns
boolfunc_size_session_screen(self, display, display_index, scale, size) -> boolGive the session compositor's screen its scale, and its mode too when the caller knows the size the screen is about to carry. Blocking.
A session lays its desktop out once per applied configuration, so a scale that arrives on its own leaves the screen at the old mode under the new scale — a fraction of the size it ends at, which is what a client that does not lay out again keeps. Older pixelflux builds have no combined call and take the scale alone.
Source Code
def _size_session_screen(self, display: str, display_index: int, scale: float,
size: Optional[Tuple[int, int]]) -> bool:
"""Give the session compositor's screen its scale, and its mode too when
the caller knows the size the screen is about to carry. Blocking.
A session lays its desktop out once per applied configuration, so a scale
that arrives on its own leaves the screen at the old mode under the new
scale — a fraction of the size it ends at, which is what a client that
does not lay out again keeps. Older pixelflux builds have no combined
call and take the scale alone.
"""
geometry = getattr(self.wayland_input, "set_app_screen_geometry", None)
if geometry is not None and size and size[0] > 0 and size[1] > 0:
return bool(geometry(display, display_index,
int(size[0]), int(size[1]), scale))
return bool(self.wayland_input.set_app_output_scale(
display, display_index, scale))paramselfparamdisplaystrparamdisplay_indexintparamscalefloatparamsizeOptional[Tuple[int, int]]Returns
boolfuncrealize_wayland_dpi(self, dpi, display_index=0, size=None) -> floatApply a DPI on the Wayland backend and return the capture output scale it leaves behind.
Applications draw larger when the compositor they are on scales its own output, so a nested session is scaled through its output management and the capture keeps 1.0: scaling the capture instead would halve the logical size the session is handed and upscale the whole desktop. A session that manages no outputs for clients (KWin) takes the capture output's scale, which it follows, and so does a plain pixelflux session, where the capture output is the only screen there is. XWayland applications need nothing merged: they run in the compositor's logical space and are scaled with it.
Source Code
async def realize_wayland_dpi(self, dpi: Any, display_index: int = 0,
size: Optional[Tuple[int, int]] = None) -> float:
"""Apply a DPI on the Wayland backend and return the capture output
scale it leaves behind.
Applications draw larger when the compositor they are on scales its own
output, so a nested session is scaled through its output management and
the capture keeps 1.0: scaling the capture instead would halve the
logical size the session is handed and upscale the whole desktop. A
session that manages no outputs for clients (KWin) takes the capture
output's scale, which it follows, and so does a plain pixelflux session,
where the capture output is the only screen there is. XWayland
applications need nothing merged: they run in the compositor's logical
space and are scaled with it.
Args:
dpi: The desktop DPI to realize; 96 is unity.
display_index: Which of the session's screens backs this display.
size: The pixel size that screen is about to carry, when the caller
already knows it, so the mode and the scale land together.
Returns:
The scale left for the capture output: 1.0 once a session absorbed
it, the full scale otherwise.
"""
try:
scale = max(0.1, float(dpi) / 96.0)
except (TypeError, ValueError):
return 1.0
try:
if not self._has_separate_app_compositor():
return scale
display = self._app_wayland_display()
applied = await asyncio.to_thread(
self._size_session_screen, display, display_index, scale, size)
except Exception as e:
logger_webrtc_input.debug(f"Session output scale failed: {e}")
return scale
if applied:
logger_webrtc_input.info(
f"Session compositor screen {display_index} scaled to {scale}.")
return 1.0
return scaleparamselfparamdpiAnyThe desktop DPI to realize; 96 is unity.
paramdisplay_indexint= 0Which of the session's screens backs this display.
paramsizeOptional[Tuple[int, int]]= NoneThe pixel size that screen is about to carry, when the caller already knows it, so the mode and the scale land together.
Returns
floatThe scale left for the capture output: 1.0 once a session absorbed
funcresync_session_screens(self) -> NoneRe-hold the session's spare screens after the output set changed.
The nested session opens the screens it was started with, and which of them a capture drives changes as displays come and go, so the hold is recomputed whenever a layout pass creates or destroys an output.
Source Code
def resync_session_screens(self) -> None:
"""Re-hold the session's spare screens after the output set changed.
The nested session opens the screens it was started with, and which of
them a capture drives changes as displays come and go, so the hold is
recomputed whenever a layout pass creates or destroys an output.
"""
if self.wayland_input is None:
return
self._schedule_spare_screen_hold()paramselfReturns
Nonefunc_schedule_spare_screen_hold(self) -> NoneA nested session opens the screens it was started with, whether or not the capture drives that many: the extra ones stretch its desktop onto a screen nobody sees, which is where a client that centres itself then lands. Hold them small until a display arrives for them — pixelflux resizes one to its full size the moment it gets an output, and back when it loses one.
Source Code
def _schedule_spare_screen_hold(self) -> None:
"""A nested session opens the screens it was started with, whether or
not the capture drives that many: the extra ones stretch its desktop
onto a screen nobody sees, which is where a client that centres itself
then lands. Hold them small until a display arrives for them —
pixelflux resizes one to its full size the moment it gets an output,
and back when it loses one."""
try:
asyncio.get_running_loop()
except RuntimeError:
return
self._spawn_task(self._hold_spare_screens())paramselfReturns
Nonefunc_hold_spare_screens(self) -> NoneHold every session screen without a capture output at SPARE_SCREEN_SIZE.
Source Code
async def _hold_spare_screens(self) -> None:
"""Hold every session screen without a capture output at SPARE_SCREEN_SIZE."""
display = self._app_wayland_display()
try:
keep = max(1, len(await asyncio.to_thread(self.wayland_input.list_outputs)))
held = await asyncio.to_thread(
self.wayland_input.hold_spare_app_screens, display, keep,
*self.SPARE_SCREEN_SIZE)
except Exception as e:
logger_webrtc_input.debug(f"Holding spare session screens failed: {e}")
return
if held:
logger_webrtc_input.info(
f"Session compositor has {held} screen(s) with no capture output; "
f"held at {self.SPARE_SCREEN_SIZE[0]}x{self.SPARE_SCREEN_SIZE[1]}.")paramselfReturns
Nonefunc_schedule_session_scale(self) -> NoneA session compositor was just adopted: hand it the effective DPI as its output scale. A scale applied before it existed landed on the capture output, which the session does not follow. An operator-set DPI governs the desktop (client syncs never reach it then); otherwise the last client-synced DPI does. 96 is unity, so nothing to apply.
Source Code
def _schedule_session_scale(self) -> None:
"""A session compositor was just adopted: hand it the effective DPI as
its output scale. A scale applied before it existed landed on the
capture output, which the session does not follow. An operator-set DPI
governs the desktop (client syncs never reach it then); otherwise the
last client-synced DPI does. 96 is unity, so nothing to apply."""
try:
if settings._overridden.get("scaling_dpi", False):
dpi = int(float(settings.scaling_dpi))
else:
dpi = int(float(getattr(self, "system_dpi", 96) or 96))
except (TypeError, ValueError, AttributeError):
return
if dpi == 96:
return
try:
asyncio.get_running_loop()
except RuntimeError:
return
self._spawn_task(self.realize_wayland_dpi(dpi))paramselfReturns
Nonefunc_get_file(self, file_path, target_mime) -> tupleRead a clipboard-referenced file, bounded to 10MB; (bytes, mime) or (None, None).
Source Code
async def _get_file(self, file_path: str, target_mime: str) -> tuple:
"""Read a clipboard-referenced file, bounded to 10MB; (bytes, mime) or (None, None)."""
max_clipboard_file_size = 10 * 1024 * 1024
try:
file_size = await asyncio.to_thread(os.path.getsize, file_path)
if file_size > max_clipboard_file_size:
logger_webrtc_input.warning(
"Skipping clipboard file %s: %d bytes exceeds 10MB limit",
file_path, file_size
)
return None, None
async with aiofiles.open(file_path, 'rb') as f:
file_data = await f.read(max_clipboard_file_size + 1)
if len(file_data) > max_clipboard_file_size:
logger_webrtc_input.warning(
"Skipping clipboard file %s: file grew beyond 10MB limit during read (%d bytes)",
file_path, len(file_data)
)
return None, None
return file_data, target_mime
except OSError as e:
logger_webrtc_input.warning("Failed to access clipboard file %s: %s", file_path, e)
return None, Noneparamselfparamfile_pathstrparamtarget_mimestrReturns
tuplefunc_kill_and_reap_process(self, proc, description) -> NoneKill a timed-out helper process and wait briefly so it is reaped.
Source Code
async def _kill_and_reap_process(self, proc: Any, description: str) -> None:
"""Kill a timed-out helper process and wait briefly so it is reaped."""
logger_webrtc_input.warning(
"Timed out waiting for clipboard command '%s' pid=%s; killing it.",
description,
getattr(proc, "pid", "unknown"),
)
try:
proc.kill()
except ProcessLookupError:
pass
try:
await asyncio.wait_for(proc.wait(), timeout=1.0)
except (asyncio.TimeoutError, ProcessLookupError):
logger_webrtc_input.warning(
"Timed-out clipboard command '%s' pid=%s could not be reaped promptly.",
description,
getattr(proc, "pid", "unknown"),
)paramselfparamprocAnyparamdescriptionstrReturns
Nonefunc_communicate_or_kill(self, proc, timeout, description, input=None) -> tupleproc.communicate with a deadline; on timeout the process is killed and reaped before TimeoutError is re-raised.
Source Code
async def _communicate_or_kill(self, proc: Any, timeout: float,
description: str,
input: Optional[bytes] = None) -> tuple:
"""proc.communicate with a deadline; on timeout the process is killed and
reaped before TimeoutError is re-raised."""
try:
return await asyncio.wait_for(proc.communicate(input=input), timeout=timeout)
except asyncio.TimeoutError:
await self._kill_and_reap_process(proc, description)
raiseparamselfparamprocAnyparamtimeoutfloatparamdescriptionstrparaminputOptional[bytes]= NoneReturns
tuplefunc_clipboard_has_consumers(self) -> boolWhether any connected client would receive an outbound clipboard send.
Source Code
def _clipboard_has_consumers(self) -> bool:
"""Whether any connected client would receive an outbound clipboard send."""
if getattr(self.rtc_app, "mode", None) != "websockets":
return True
server = self.data_server_instance or getattr(self.rtc_app, "data_streaming_server", None)
return bool(server and getattr(server, "clients", None))paramselfReturns
boolfunc_app_clipboard_read(self, use_binary) -> tupleRead the selection of the compositor the apps use over the pixelflux data-control ABI; (None, None) when it is empty or unreadable.
Source Code
async def _app_clipboard_read(self, use_binary: bool) -> tuple:
"""Read the selection of the compositor the apps use over the pixelflux
data-control ABI; (None, None) when it is empty or unreadable."""
read_fn = getattr(self.wayland_input, 'clipboard_read_app', None)
types_fn = getattr(self.wayland_input, 'clipboard_types_app', None)
if read_fn is None or types_fn is None:
# A pixelflux without the data-control ABI reports no data, not failure.
return None, None
display = self._app_wayland_display()
loop = asyncio.get_running_loop()
try:
available_types = await loop.run_in_executor(
None, types_fn, display)
self._app_clip_read_failure = None
if use_binary:
image_mimes = ['image/png', 'image/jpeg', 'image/bmp', 'image/webp',
'image/svg+xml', 'image/svg']
target_mime = next((m for m in image_mimes if m in available_types), None)
if target_mime:
data = await loop.run_in_executor(
None, read_fn, display, target_mime)
if data:
return bytes(data), target_mime
text_mimes = ['text/plain;charset=utf-8', 'text/plain',
'UTF8_STRING', 'STRING', 'TEXT']
source_mime = next((m for m in text_mimes if m in available_types), None)
if source_mime:
data = await loop.run_in_executor(None, read_fn, display, source_mime)
if data is not None:
return data.decode('utf-8', errors='replace'), 'text/plain'
except Exception as e:
failure = (display, str(e))
log = (logger_webrtc_input.debug if failure == self._app_clip_read_failure
else logger_webrtc_input.warning)
self._app_clip_read_failure = failure
log(f"data-control clipboard read failed: {e}")
self._invalidate_app_wl_display_if_dead(display)
return None, Noneparamselfparamuse_binaryboolReturns
tuplefuncread_clipboard(self, use_binary=False) -> tupleRead the session clipboard.
Wayland is fully native. A rootful Xwayland on the capture compositor bridges no selection of its own, so the X11 desktop's copies are read from it first (a selection still owned there is our own write, for which the compositor side is authoritative); then the compositor callback's cache, which holds the capture compositor's selection and is the wrong session under a separate app compositor, where the data-control client reads whichever socket the apps use instead. X11 uses the XFixes monitor with an xclip fallback.
Source Code
async def read_clipboard(self, use_binary: bool = False) -> tuple:
"""Read the session clipboard.
Wayland is fully native. A rootful Xwayland on the capture compositor
bridges no selection of its own, so the X11 desktop's copies are read
from it first (a selection still owned there is our own write, for
which the compositor side is authoritative); then the compositor
callback's cache, which holds the capture compositor's selection and
is the wrong session under a separate app compositor, where the
data-control client reads whichever socket the apps use instead. X11
uses the XFixes monitor with an xclip fallback.
Args:
use_binary: Prefer image targets (and file-manager uri-lists)
before falling back to text.
Returns:
(data, mime): text as str with mime 'text/plain', images as bytes
with their mime, or (None, None) when nothing is readable.
"""
if self.is_wayland:
monitor = await self._ensure_x11_clipboard_monitor_async()
if monitor is not None and not monitor.owns_selection():
try:
loop = asyncio.get_running_loop()
data, mime = await loop.run_in_executor(None, monitor.read, use_binary)
if data is not None:
return data, mime
except Exception as e:
logger_webrtc_input.warning(f"X11 clipboard read on the Wayland session failed: {e}")
cached = (None if self._has_separate_app_compositor()
else getattr(self, '_wl_native_last', None))
if cached is not None:
raw, native_mime = cached
if native_mime.startswith('image/'):
if use_binary:
return bytes(raw), native_mime
else:
return bytes(raw).decode('utf-8', errors='replace'), 'text/plain'
return await self._app_clipboard_read(use_binary)
monitor = await self._ensure_x11_clipboard_monitor_async()
if monitor is not None:
try:
loop = asyncio.get_running_loop()
return await loop.run_in_executor(None, monitor.read, use_binary)
except Exception as e:
logger_webrtc_input.warning(f"native X11 clipboard read failed, using xclip: {e}")
try:
proc_targets = await subprocess.create_subprocess_exec(
"xclip", "-selection", "clipboard", "-o", "-t", "TARGETS",
stdout=subprocess.PIPE, stderr=subprocess.PIPE
)
stdout_targets, _ = await self._communicate_or_kill(proc_targets, 1, "xclip TARGETS")
if proc_targets.returncode != 0:
return None, None
targets = stdout_targets.decode().strip().split('\n')
if use_binary:
for mime_type in ['image/png', 'image/jpeg', 'image/bmp', 'image/webp',
'image/svg+xml', 'image/svg']:
if mime_type in targets:
proc_data = await subprocess.create_subprocess_exec(
"xclip", "-selection", "clipboard", "-o", "-t", mime_type,
stdout=subprocess.PIPE, stderr=subprocess.PIPE
)
stdout_data, _ = await self._communicate_or_kill(proc_data, 3, f"xclip {mime_type}")
if proc_data.returncode == 0 and stdout_data:
return stdout_data, mime_type
# File-manager copy: a text/uri-list of file:// URIs.
if 'text/uri-list' in targets:
proc_data = await subprocess.create_subprocess_exec(
"xclip", "-selection", "clipboard", "-o", "-t", "text/uri-list",
stdout=subprocess.PIPE, stderr=subprocess.PIPE
)
stdout_data, _ = await self._communicate_or_kill(proc_data, 1, "xclip text/uri-list")
if proc_data.returncode == 0 and stdout_data:
lines = stdout_data.decode("utf-8", errors="replace").splitlines()
for line in lines:
line = line.strip()
if not line or line.startswith("#"):
continue
parsed_uri = urllib.parse.urlparse(line)
if parsed_uri.scheme == 'file':
file_path = urllib.request.url2pathname(parsed_uri.path)
if os.path.isfile(file_path):
ext = os.path.splitext(file_path)[1].lower()
mime_map = {
'.png': 'image/png', '.jpg': 'image/jpeg',
'.jpeg': 'image/jpeg', '.bmp': 'image/bmp',
'.webp': 'image/webp', '.svg': 'image/svg+xml'
}
if ext in mime_map:
target_mime = mime_map[ext]
return await self._get_file(file_path, target_mime)
if 'UTF8_STRING' in targets:
proc_text = await subprocess.create_subprocess_exec(
"xclip", "-selection", "clipboard", "-o", "-t", "UTF8_STRING",
stdout=subprocess.PIPE, stderr=subprocess.PIPE
)
stdout_text, _ = await self._communicate_or_kill(proc_text, 1, "xclip UTF8_STRING")
if proc_text.returncode == 0:
return stdout_text.decode(), 'text/plain'
return None, None
except FileNotFoundError:
if not self._xclip_missing_warned:
self._xclip_missing_warned = True
logger_webrtc_input.warning(
"xclip is not installed; the clipboard polling rung has "
"nothing to read with.")
return None, None
except Exception as e:
logger_webrtc_input.warning(f"Error reading clipboard with xclip: {e}", exc_info=True)
return None, Noneparamselfparamuse_binarybool= FalsePrefer image targets (and file-manager uri-lists) before falling back to text.
Returns
(data, mime)text as str with mime 'text/plain', images as bytes
funcwrite_clipboard(self, data, mime_type='text/plain') -> boolSet the session clipboard, native first with forked fallbacks.
Wayland sets pixelflux's own selection in-process (or writes through the data-control client to a separate app compositor); X11 offers through the XFixes monitor's connection, falling back to an xclip fork. The written bytes become the monitor baseline BEFORE the write so the ownership-change event cannot echo a client's own content back (that echo loop saturates the transport). On the capture compositor the payload is also offered on the unbridged X server (a rootful Xwayland sees no Wayland selection) so the X11 desktop can paste it.
Source Code
async def write_clipboard(self, data: Union[str, bytes],
mime_type: str = "text/plain") -> bool:
"""Set the session clipboard, native first with forked fallbacks.
Wayland sets pixelflux's own selection in-process (or writes through
the data-control client to a separate app compositor); X11 offers
through the XFixes monitor's connection, falling back to an xclip
fork. The written bytes become the monitor baseline BEFORE the write
so the ownership-change event cannot echo a client's own content back
(that echo loop saturates the transport). On the capture compositor
the payload is also offered on the unbridged X server (a rootful
Xwayland sees no Wayland selection) so the X11 desktop can paste it.
Returns:
True when the clipboard was set (an empty payload is a no-op True).
"""
if not data:
return True
input_bytes = data if isinstance(data, bytes) else data.encode('utf-8')
self._clipboard_last_bytes = input_bytes
if self.is_wayland:
if not self._has_separate_app_compositor():
try:
self.wayland_input.set_clipboard(mime_type, input_bytes)
# The compositor does not echo its own selection back; a
# later read (another client joining) must still see it.
self._wl_native_last = (input_bytes, mime_type)
ok = True
except Exception as e:
logger_webrtc_input.warning(f"native wayland clipboard set failed: {e}")
ok = False
monitor = await self._ensure_x11_clipboard_monitor_async()
if monitor is not None:
try:
loop = asyncio.get_running_loop()
if await loop.run_in_executor(None, monitor.offer, input_bytes, mime_type):
ok = True
except Exception as e:
logger_webrtc_input.warning(f"X11 clipboard offer on the Wayland session failed: {e}")
return ok
# Text is offered under every conventional target; apps pick their own.
if mime_type == "text/plain":
entries = [(m, input_bytes) for m in (
"text/plain;charset=utf-8", "text/plain",
"UTF8_STRING", "STRING", "TEXT")]
else:
entries = [(mime_type, input_bytes)]
try:
await asyncio.get_running_loop().run_in_executor(
None, self.wayland_input.clipboard_write_app,
self._app_wayland_display(), entries)
return True
except Exception as e:
logger_webrtc_input.warning(f"data-control clipboard write failed: {e}")
self._invalidate_app_wl_display_if_dead(self._app_wayland_display())
return False
env = os.environ.copy()
if 'LANG' not in env or env['LANG'] == 'C':
env['LANG'] = 'C.UTF-8'
monitor = await self._ensure_x11_clipboard_monitor_async()
if monitor is not None:
try:
loop = asyncio.get_running_loop()
ok = await loop.run_in_executor(None, monitor.offer, input_bytes, mime_type)
if ok:
return True
except Exception as e:
logger_webrtc_input.warning(f"native X11 clipboard offer failed, using xclip: {e}")
try:
is_text = mime_type == "text/plain"
target_mime = "UTF8_STRING" if is_text else mime_type
process = await subprocess.create_subprocess_exec(
"xclip", "-selection", "clipboard", "-i", "-t", target_mime,
stdin=subprocess.PIPE,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
env=env
)
# One deadline for the stdin write plus exit: a payload larger than
# the pipe buffer cannot wedge on a stalled xclip.
await self._communicate_or_kill(
process, 2.0, f"xclip -i {target_mime}", input_bytes)
return_code = process.returncode
if return_code == 0:
return True
else:
logger_webrtc_input.warning(f"xclip process exited with non-zero code: {return_code}")
return False
except asyncio.TimeoutError:
logger_webrtc_input.warning("Timeout waiting for xclip process to terminate.")
return False
except FileNotFoundError:
if not self._xclip_missing_warned:
self._xclip_missing_warned = True
logger_webrtc_input.warning(
"xclip is not installed; the clipboard polling rung has "
"nothing to write with.")
return False
except Exception:
logger_webrtc_input.warning("Error writing to clipboard with xclip", exc_info=True)
return FalseparamselfparamdataUnion[str, bytes]parammime_typestr= 'text/plain'Returns
boolTrue when the clipboard was set (an empty payload is a no-op True).
func_x11_session_display(self) -> Optional[str]Display of an X server whose selections nothing bridges into the Wayland session: on the Wayland backend without a nested session compositor, a live server on $DISPLAY is a rootful (or unmanaged) Xwayland client of the capture compositor hosting an X11 desktop, and Xwayland bridges no selection by itself. A nested compositor's own XWM bridges its Xwayland into the selection selkies already watches, and on the X11 backend the display is the session itself, so both answer None. The server must be an Xwayland (not a leftover Xvfb/Xorg that merely holds the display number), or nothing on it belongs to this session.
Source Code
def _x11_session_display(self) -> Optional[str]:
"""Display of an X server whose selections nothing bridges into the
Wayland session: on the Wayland backend without a nested session
compositor, a live server on $DISPLAY is a rootful (or unmanaged)
Xwayland client of the capture compositor hosting an X11 desktop, and
Xwayland bridges no selection by itself. A nested compositor's own XWM
bridges its Xwayland into the selection selkies already watches, and on
the X11 backend the display is the session itself, so both answer None.
The server must be an Xwayland (not a leftover Xvfb/Xorg that merely
holds the display number), or nothing on it belongs to this session."""
if not self.is_wayland or self._has_separate_app_compositor():
return None
name = os.environ.get("DISPLAY", "").strip()
if not name or not x_display_live(name) or not x_display_is_xwayland(name):
return None
return nameparamselfReturns
typing.Optional[str]func_ensure_x11_clipboard_monitor(self, display_name=None) -> Optional[_X11ClipboardMonitor]Get-or-create the event-driven X11 monitor; None if unavailable. On
the Wayland backend display_name names the unbridged X server to
watch (see _x11_session_display) and None means there is none.
Source Code
def _ensure_x11_clipboard_monitor(self, display_name: Optional[str] = None) -> Optional[_X11ClipboardMonitor]:
"""Get-or-create the event-driven X11 monitor; None if unavailable. On
the Wayland backend ``display_name`` names the unbridged X server to
watch (see _x11_session_display) and None means there is none."""
if self._x11_clipboard_monitor is not None:
return self._x11_clipboard_monitor
if not X11_LIBS_AVAILABLE or (self.is_wayland and display_name is None):
return None
try:
self._x11_clipboard_monitor = _X11ClipboardMonitor(display_name)
self._x11_monitor_unavail_logged = False
where = f" on {display_name} (unbridged X server of the Wayland session)" if self.is_wayland else ""
logger_webrtc_input.info(f"X11 clipboard: XFixes event monitor active{where}.")
except Exception as e:
log = (logger_webrtc_input.debug if self._x11_monitor_unavail_logged
else logger_webrtc_input.info)
# The Wayland backend has no X poll rung (the compositor feed carries
# the selection), so the monitor is simply retried.
fallback = "retrying" if self.is_wayland else "falling back to polling"
log(f"X11 clipboard: XFixes monitor unavailable ({e}); {fallback}.")
self._x11_monitor_unavail_logged = True
self._x11_clipboard_monitor = None
return self._x11_clipboard_monitorparamselfparamdisplay_nameOptional[str]= NoneReturns
typing.Optional[selkies.input_handler._X11ClipboardMonitor]func_ensure_x11_clipboard_monitor_async(self) -> Optional[_X11ClipboardMonitor]Off-loop get-or-create: construction opens its own X connection, and a server disrupted mid-session (the respawn case) would stall the event loop for the whole bounded handshake. The lock keeps concurrent callers from racing two monitors into existence. A failed build backs off before the next attempt, so pollers do not hammer a dead display with connection attempts — and a display that comes up later is still re-probed.
Source Code
async def _ensure_x11_clipboard_monitor_async(self) -> Optional[_X11ClipboardMonitor]:
"""Off-loop get-or-create: construction opens its own X connection, and a
server disrupted mid-session (the respawn case) would stall the event loop
for the whole bounded handshake. The lock keeps concurrent callers from
racing two monitors into existence. A failed build backs off before the
next attempt, so pollers do not hammer a dead display with connection
attempts — and a display that comes up later is still re-probed."""
if self._x11_clipboard_monitor is not None:
return self._x11_clipboard_monitor
if time.monotonic() < self._x11_monitor_retry_at:
return None
async with self._x11_monitor_build_lock:
if self._x11_clipboard_monitor is not None:
return self._x11_clipboard_monitor
display_name = self._x11_session_display() if self.is_wayland else None
if self.is_wayland and display_name is None:
# No unbridged X server right now; no backoff, since the socket
# probe is cheap and an Xwayland starting later is caught within a tick.
return None
monitor = await asyncio.to_thread(self._ensure_x11_clipboard_monitor, display_name)
if monitor is None:
self._x11_monitor_retry_at = time.monotonic() + 10.0
return monitorparamselfReturns
typing.Optional[selkies.input_handler._X11ClipboardMonitor]func_wait_x11_or_compositor_change(x11_monitor, queue, timeout) -> tupleWait for whichever signals first: the X server's selection-owner change or a compositor clipboard delivery. Returns (changed, item) where item is the compositor delivery, if that is what fired.
Source Code
@staticmethod
async def _wait_x11_or_compositor_change(x11_monitor: _X11ClipboardMonitor,
queue: "asyncio.Queue", timeout: float) -> tuple:
"""Wait for whichever signals first: the X server's selection-owner
change or a compositor clipboard delivery. Returns (changed, item) where
item is the compositor delivery, if that is what fired."""
x_wait = asyncio.ensure_future(x11_monitor.wait_change(timeout))
q_get = asyncio.ensure_future(queue.get())
try:
done, _ = await asyncio.wait({x_wait, q_get}, timeout=timeout + 0.5,
return_when=asyncio.FIRST_COMPLETED)
item = q_get.result() if q_get in done else None
x_changed = bool(x_wait.result()) if x_wait in done else False
return (item is not None or x_changed), item
finally:
# Whichever did not fire (both on cancellation), so no queue.get() lingers.
for fut in (x_wait, q_get):
if not fut.done():
fut.cancel()paramx11_monitor_X11ClipboardMonitorparamqueueasyncio.QueueparamtimeoutfloatReturns
tuplefunc_arm_wayland_native_clipboard(self) -> Optional[asyncio.Queue]Register the compositor clipboard callback (fork-free watch+read); returns the delivery queue, or None when the API is unavailable.
Registered once per monitor start: the compositor keeps the callback across capture stops and starts, and every registration stages a fresh read of the whole current selection (so a copy made before the monitor ran is delivered), which must not recur while idle.
Source Code
def _arm_wayland_native_clipboard(self) -> Optional[asyncio.Queue]:
"""Register the compositor clipboard callback (fork-free watch+read);
returns the delivery queue, or None when the API is unavailable.
Registered once per monitor start: the compositor keeps the callback
across capture stops and starts, and every registration stages a
fresh read of the whole current selection (so a copy made before the
monitor ran is delivered), which must not recur while idle."""
if not (self.wayland_input is not None
and hasattr(self.wayland_input, 'set_clipboard_callback')):
return None
try:
loop = asyncio.get_running_loop()
queue = asyncio.Queue(maxsize=4)
def _on_clip(mime, data):
# Cache for on-demand reads (cr/REQUEST_CLIPBOARD).
self._wl_native_last = (bytes(data), mime)
def _put():
if queue.full():
queue.get_nowait()
queue.put_nowait((data, mime))
loop.call_soon_threadsafe(_put)
self.wayland_input.set_clipboard_callback(_on_clip)
self._wl_native_arm_failure = None
logger_webrtc_input.info("Wayland clipboard: native compositor callback active (no polling).")
return queue
except Exception as e:
if str(e) != getattr(self, '_wl_native_arm_failure', None):
self._wl_native_arm_failure = str(e)
logger_webrtc_input.warning(
f"Wayland clipboard: native callback failed to arm ({e}); retrying.")
return NoneparamselfReturns
typing.Optional[asyncio.asyncio.Queue]func_arm_app_compositor_watch(self) -> Optional[asyncio.Queue]Selection-change signals from the app compositor over the pixelflux data-control ABI (fork-free); returns the signal queue, or None when no watch could be armed — the monitor loop then polls that compositor and retries the arm each tick. The watch itself only reports a failed data-control handshake on its own thread, so the same handshake is made here first (one-shot, off the loop): a compositor that offers neither ext- nor zwlr-data-control is known at arm time instead of leaving a watch that never fires. Each failure is logged once.
Source Code
async def _arm_app_compositor_watch(self) -> Optional[asyncio.Queue]:
"""Selection-change signals from the app compositor over the pixelflux
data-control ABI (fork-free); returns the signal queue, or None when no
watch could be armed — the monitor loop then polls that compositor and
retries the arm each tick. The watch itself only reports a failed
data-control handshake on its own thread, so the same handshake is
made here first (one-shot, off the loop): a compositor that offers
neither ext- nor zwlr-data-control is known at arm time instead of
leaving a watch that never fires. Each failure is logged once."""
watch_fn = getattr(self.wayland_input, 'clipboard_watch_app', None)
if watch_fn is None:
# Without the data-control ABI the native callback monitor covers this display.
return None
display = self._app_wayland_display()
try:
probe = getattr(self.wayland_input, 'clipboard_types_app', None)
if probe is not None:
await asyncio.to_thread(probe, display)
loop = asyncio.get_running_loop()
queue = asyncio.Queue(maxsize=4)
def _on_change(mimes):
def _put():
if queue.full():
queue.get_nowait()
queue.put_nowait(mimes)
loop.call_soon_threadsafe(_put)
watch_fn(display, _on_change)
self._app_watch_failure = None
logger_webrtc_input.info(
"Wayland clipboard: app-compositor data-control watch active (no forks).")
return queue
except Exception as e:
failure = (display, str(e))
if failure != self._app_watch_failure:
self._app_watch_failure = failure
logger_webrtc_input.warning(
f"Wayland clipboard: no selection watch on app compositor "
f"'{display}' ({e}); polling it instead.")
self._invalidate_app_wl_display_if_dead(display)
return NoneparamselfReturns
typing.Optional[asyncio.asyncio.Queue]funcstart_clipboard(self) -> NoneRun the outbound clipboard monitor until stop_clipboard.
Event-driven on every rung that offers events (XFixes monitor, compositor callback, app-compositor data-control watch), with xclip polling as the last X11 fallback; each pass reads the selection, compares against the echo baseline, and broadcasts real changes to clients. The first consumer after a consumer-less stretch gets the current selection once, since change events during that stretch were skipped and a copy made before any client connected would otherwise never arrive.
The compositor callback watches the capture compositor's selection; with a separate app compositor the apps' copies land on its selection instead, watched through the data-control client, which is (re)armed inside the loop so a nested session appearing after startup — or restarting on a new socket — is picked up. A mode switch stops the running monitor and starts the replacement immediately, so the singleton guard waits briefly for the stopped loop to unwind instead of refusing, which would leave the session with no outbound clipboard until the setting is toggled.
Source Code
async def start_clipboard(self) -> None:
"""Run the outbound clipboard monitor until stop_clipboard.
Event-driven on every rung that offers events (XFixes monitor,
compositor callback, app-compositor data-control watch), with xclip
polling as the last X11 fallback; each pass reads the selection,
compares against the echo baseline, and broadcasts real changes to
clients. The first consumer after a consumer-less stretch gets the
current selection once, since change events during that stretch were
skipped and a copy made before any client connected would otherwise
never arrive.
The compositor callback watches the capture compositor's selection;
with a separate app compositor the apps' copies land on its selection
instead, watched through the data-control client, which is (re)armed
inside the loop so a nested session appearing after startup — or
restarting on a new socket — is picked up. A mode switch stops the
running monitor and starts the replacement immediately, so the
singleton guard waits briefly for the stopped loop to unwind instead
of refusing, which would leave the session with no outbound clipboard
until the setting is toggled.
"""
if self.enable_clipboard not in ["true", "out"]:
logger_webrtc_input.info("Skipping outbound clipboard service."); return
for _ in range(50):
if not self._clipboard_monitor_active:
break
await asyncio.sleep(0.1)
if self._clipboard_monitor_active:
logger_webrtc_input.info("Clipboard monitor already running; not starting a second instance.")
return
self._clipboard_monitor_active = True
logger_webrtc_input.info(f"Clipboard monitor running (binary mode: {self.enable_binary_clipboard in ['true', 'out']})")
self.clipboard_running = True
x11_monitor = await self._ensure_x11_clipboard_monitor_async()
wl_native_queue = (self._arm_wayland_native_clipboard()
if self.is_wayland and not self._has_separate_app_compositor()
else None)
wl_native_item = None
app_watch_queue = None
app_watch_display = None
# Primed so the first pass publishes the current content once.
first_pass = True
had_consumers = False
try:
while self.clipboard_running:
try:
wl_native_item = None
if self.is_wayland and self._has_separate_app_compositor():
if x11_monitor is not None:
# A nested session compositor appeared: its XWM bridges
# its own Xwayland, so the app-compositor watch takes over.
try:
x11_monitor.close()
except Exception:
pass
self._x11_clipboard_monitor = None
x11_monitor = None
disp = self._app_wayland_display()
if app_watch_queue is None or disp != app_watch_display:
q = await self._arm_app_compositor_watch()
if q is not None:
app_watch_queue, app_watch_display = q, disp
elif self.is_wayland and wl_native_queue is None:
# Direct mode reached only now (a nested compositor died
# or never appeared): arm the compositor callback late.
wl_native_queue = self._arm_wayland_native_clipboard()
if self.is_wayland and x11_monitor is None:
# An X11 desktop's Xwayland comes up after its compositor;
# watched from the moment the server answers.
x11_monitor = await self._ensure_x11_clipboard_monitor_async()
if first_pass:
changed = True
first_pass = False
elif x11_monitor is not None:
if not x11_monitor.alive():
# Rebuilt in place so outbound clipboard heals on its own
# instead of staying dead until a setting is toggled.
logger_webrtc_input.warning(
"X11 clipboard monitor thread exited; respawning.")
try:
x11_monitor.close()
except Exception:
pass
self._x11_clipboard_monitor = None
x11_monitor = await self._ensure_x11_clipboard_monitor_async()
if x11_monitor is None:
await asyncio.sleep(0.5)
changed = False
else:
# Republish on the fresh monitor; the baseline still dedupes.
changed = True
elif wl_native_queue is not None:
changed, wl_native_item = await self._wait_x11_or_compositor_change(
x11_monitor, wl_native_queue, 2.0)
else:
changed = await x11_monitor.wait_change(2.0)
elif wl_native_queue is not None and not self._has_separate_app_compositor():
# Not re-armed on idle: re-registering would stage a full
# selection read every tick.
try:
wl_native_item = await asyncio.wait_for(wl_native_queue.get(), 2.0)
changed = True
except asyncio.TimeoutError:
changed = False
elif app_watch_queue is not None:
try:
await asyncio.wait_for(app_watch_queue.get(), 2.0)
changed = True
except asyncio.TimeoutError:
changed = False
# A watch on a dead compositor never fires again; a dead
# socket drops the detection so the loop re-arms.
self._invalidate_app_wl_display_if_dead(app_watch_display)
if self._app_wl_display_cached is None:
app_watch_queue, app_watch_display = None, None
elif self.is_wayland and self._has_separate_app_compositor():
# Poll rung (no data-control watch); the arm above is
# retried every tick and takes over the moment one holds.
await asyncio.sleep(2.0)
changed = True
elif self.is_wayland:
# Neither watch armed yet (compositor briefly absent).
await asyncio.sleep(0.5)
changed = False
else:
# Poll rung; the XFixes rung is re-probed on its cooldown,
# so an X server that answers later upgrades back to events.
x11_monitor = await self._ensure_x11_clipboard_monitor_async()
await asyncio.sleep(0.5)
changed = True
has_consumers = self._clipboard_has_consumers()
if has_consumers and not had_consumers:
changed = True
had_consumers = has_consumers
if not changed:
continue
if not has_consumers:
continue
use_binary = self.enable_binary_clipboard in ["true", "out"]
if wl_native_item is not None:
raw, native_mime = wl_native_item
if native_mime.startswith('image/') and use_binary:
curr_data, curr_mime = bytes(raw), native_mime
elif not native_mime.startswith('image/'):
curr_data = bytes(raw).decode('utf-8', errors='replace')
curr_mime = 'text/plain'
else:
curr_data, curr_mime = None, None
elif x11_monitor is not None:
loop = asyncio.get_running_loop()
curr_data, curr_mime = await loop.run_in_executor(
None, x11_monitor.read, use_binary)
else:
curr_data, curr_mime = await self.read_clipboard(use_binary=use_binary)
if curr_data is None:
curr_data_bytes = None
else:
curr_data_bytes = curr_data.encode('utf-8') if isinstance(curr_data, str) else curr_data
if curr_data_bytes is not None and curr_data_bytes != self._clipboard_last_bytes:
logger_webrtc_input.info(f"Clipboard changed. Sending content ({curr_mime})")
self._clipboard_last_bytes = curr_data_bytes
await self.on_clipboard_read(curr_data, curr_mime)
except asyncio.CancelledError:
logger_webrtc_input.info("Clipboard monitor task cancelled.")
break
except Exception as e:
logger_webrtc_input.error(f"Error in clipboard monitor loop: {e}", exc_info=True)
await asyncio.sleep(2)
finally:
self.clipboard_running = False
self._clipboard_monitor_active = False
logger_webrtc_input.info("Clipboard monitor stopped")paramselfReturns
Nonefuncstop_clipboard(self) -> NoneStop the monitor loop and release the X11 monitor's connection and thread; a mode switch builds a fresh handler and must not leak one monitor per transition.
Source Code
def stop_clipboard(self) -> None:
"""Stop the monitor loop and release the X11 monitor's connection and
thread; a mode switch builds a fresh handler and must not leak one
monitor per transition."""
self.clipboard_running = False
if self._x11_clipboard_monitor is not None:
self._x11_clipboard_monitor.close()
self._x11_clipboard_monitor = None
logger_webrtc_input.info("Stopping clipboard monitor")paramselfReturns
Nonefunc_handle_mapping_notify(self, event) -> NoneKeep the python-xlib keymap cache and the XTEST overlay coherent with server-side keymap changes. Every in-session layout switch (setxkbmap, desktop layout applets, fcitx-xkb) lands here as a MappingNotify; without the refresh, keysym_to_keycode resolves against the dead layout and the overlay trusts bindings the switch wiped.
A modifier remap only re-resolves the Shift/AltGr keycodes: it never touches the overlay, and this shim's own overlay binds surface as Modifier notifies on some servers, so clearing the overlay there would loop bind, notify, clear forever.
Source Code
def _handle_mapping_notify(self, event: Any) -> None:
"""Keep the python-xlib keymap cache and the XTEST overlay coherent with
server-side keymap changes. Every in-session layout switch (setxkbmap,
desktop layout applets, fcitx-xkb) lands here as a MappingNotify; without
the refresh, keysym_to_keycode resolves against the dead layout and the
overlay trusts bindings the switch wiped.
A modifier remap only re-resolves the Shift/AltGr keycodes: it never
touches the overlay, and this shim's own overlay binds surface as
Modifier notifies on some servers, so clearing the overlay there would
loop bind, notify, clear forever.
"""
kb = self.keyboard
if event.request == X.MappingModifier:
if kb is not None:
kb.refresh_modifier_keycodes()
return
if event.request != X.MappingKeyboard:
return
try:
self.xdisplay.refresh_keyboard_mapping(event)
except Exception as e:
logger_webrtc_input.warning(f"keymap cache refresh failed: {e}")
if kb is None:
return
kb.note_mapping_change(event.first_keycode, event.count)
if kb.bindings_intact():
# Our own bind, or a change that left the overlay alone.
return
logger_webrtc_input.info(
"Foreign keymap change detected (request=%d, keycodes %d+%d): "
"invalidating XTEST overlay state.",
event.request, event.first_keycode, event.count)
kb.invalidate_mapping()paramselfparameventAnyReturns
Nonefunc_dispatch_keymap_event(self, event) -> boolHand a keymap change on the input connection to _handle_mapping_notify.
A core MappingNotify goes as is. A whole-keyboard replacement (setxkbmap, a desktop layout switcher) reaches the XKB-aware input connection only as XkbNewKeyboardNotify, which is handed on as the full-range keyboard MappingNotify it stands for.
Source Code
def _dispatch_keymap_event(self, event: Any) -> bool:
"""Hand a keymap change on the input connection to _handle_mapping_notify.
A core MappingNotify goes as is. A whole-keyboard replacement
(setxkbmap, a desktop layout switcher) reaches the XKB-aware input
connection only as XkbNewKeyboardNotify, which is handed on as the
full-range keyboard MappingNotify it stands for.
Returns:
True when the event was a keymap change and has been handled.
"""
if event.type == X.MappingNotify:
self._handle_mapping_notify(event)
return True
kb = self.keyboard
span = kb.keyboard_replaced(event) if kb is not None else None
if span is None:
return False
lo, hi = span
self._handle_mapping_notify(xevent.MappingNotify(
sequence_number=event.sequence_number, request=X.MappingKeyboard,
first_keycode=lo, count=hi - lo + 1))
return TrueparamselfparameventAnyReturns
boolTrue when the event was a keymap change and has been handled.
func_keymap_watch_loop(self) -> NoneDrain X events for MappingNotify when the cursor monitor is not the event consumer (pixelflux delivers cursors natively then). Two consumers must never race next_event(), so this loop idles while cursors_running.
Source Code
async def _keymap_watch_loop(self) -> None:
"""Drain X events for MappingNotify when the cursor monitor is not the
event consumer (pixelflux delivers cursors natively then). Two consumers
must never race next_event(), so this loop idles while cursors_running."""
while True:
if self.cursors_running or self.xdisplay is None:
# Idle poll only: the event wake is not armed for this consumer
# when nothing drives it, and a foreign remap is not urgent here.
await asyncio.sleep(0.5)
continue
wake = self._x_event_wake
if wake is not None:
wake.clear()
if self.xdisplay.pending_events() == 0:
self._arm_x_event_watcher()
await self._wait_x_event(timeout=2.0)
if self.cursors_running or self.xdisplay is None:
continue
try:
while self.xdisplay.pending_events():
self._dispatch_keymap_event(self.xdisplay.next_event())
except asyncio.CancelledError:
raise
except Exception as e:
if self._is_x_conn_closed(e):
self._reconnect_xdisplay()
else:
logger_webrtc_input.debug(f"keymap watch: {e}")paramselfReturns
Nonefuncstart_cursor_monitor(self) -> NoneWatch XFixes cursor-change events and push encoded cursors to clients.
Runs only when pixelflux does not already deliver cursors natively; this loop is then the session's single X event consumer (MappingNotify included), so it never races _keymap_watch_loop on next_event(). The X fetch stays on this thread (python-xlib connections are not thread-safe and the loop also injects input on this display), while the PIL resize and PNG encode, pure CPU, run off the loop.
Source Code
async def start_cursor_monitor(self) -> None:
"""Watch XFixes cursor-change events and push encoded cursors to clients.
Runs only when pixelflux does not already deliver cursors natively;
this loop is then the session's single X event consumer (MappingNotify
included), so it never races _keymap_watch_loop on next_event(). The
X fetch stays on this thread (python-xlib connections are not
thread-safe and the loop also injects input on this display), while
the PIL resize and PNG encode, pure CPU, run off the loop.
"""
if self.is_wayland:
logger_webrtc_input.info("Wayland mode: Cursor monitor disabled (handled by compositor callback).")
return
if pixelflux_x11_cursor():
logger_webrtc_input.info(
"X11 cursor monitor disabled (pixelflux cursor callback active)."
)
return
if not self.xdisplay.has_extension("XFIXES"):
if self.xdisplay.query_extension("XFIXES") is None:
logger_webrtc_input.error(
"XFIXES extension not supported, cannot watch cursor changes"
)
return
xfixes_version = self.xdisplay.xfixes_query_version()
logger_webrtc_input.info(
"Found XFIXES version %s.%s",
xfixes_version.major_version,
xfixes_version.minor_version,
)
logger_webrtc_input.info("starting cursor monitor")
self.cursors_running = True
screen = self.xdisplay.screen()
self.xdisplay.xfixes_select_cursor_input(
screen.root, xfixes.XFixesDisplayCursorNotifyMask
)
logger_webrtc_input.info("watching for cursor changes")
try:
cursor_image = self.xdisplay.xfixes_get_cursor_image(screen.root)
cursor_data = await asyncio.to_thread(self._encode_cursor, cursor_image)
self.on_cursor_change(cursor_data)
except Exception as e:
logger_webrtc_input.warning("exception from fetching initial cursor image: %s", e)
if self._is_x_conn_closed(e):
self._reconnect_xdisplay()
while self.cursors_running:
if self.xdisplay is None:
# A background reconnect is in flight; the fresh connection
# was xfixes-armed at install, so only screen needs rebinding.
await asyncio.sleep(0.5)
if self.xdisplay is not None:
screen = self.xdisplay.screen()
continue
wake = self._x_event_wake
if wake is not None:
wake.clear()
if self.xdisplay.pending_events() == 0:
# The 1 s failsafe bounds the wait if the loop reader was lost to a reconnect.
self._arm_x_event_watcher()
await self._wait_x_event(timeout=1.0)
continue
event = self.xdisplay.next_event()
if self._dispatch_keymap_event(event):
continue
if (event.type, 0) == self.xdisplay.extension_event.DisplayCursorNotify:
try:
cursor_image = self.xdisplay.xfixes_get_cursor_image(screen.root)
cursor_data = await asyncio.to_thread(self._encode_cursor, cursor_image)
self.on_cursor_change(cursor_data)
except Exception as e:
logger_webrtc_input.warning(
"exception from fetching cursor image on change: %s", e
)
if self._is_x_conn_closed(e):
# The None guard above rebinds screen once the reconnect lands.
self._reconnect_xdisplay()
logger_webrtc_input.info("cursor monitor stopped")paramselfReturns
Nonefuncstop_cursor_monitor(self) -> NoneSource Code
def stop_cursor_monitor(self) -> None:
logger_webrtc_input.info("stopping cursor monitor")
self.cursors_running = FalseparamselfReturns
Nonefuncget_current_cursor_data(self) -> Optional[dict]One-shot fetch of the current X cursor as a client message, for seeding a newly connected client; None when unavailable.
Source Code
def get_current_cursor_data(self) -> Optional[dict]:
"""One-shot fetch of the current X cursor as a client message, for
seeding a newly connected client; None when unavailable."""
if self.is_wayland:
return None
if not self.enable_cursors or not self.xdisplay:
return None
try:
if not self.xdisplay.has_extension("XFIXES"):
if self.xdisplay.query_extension("XFIXES") is None:
logger_webrtc_input.error(
"XFIXES extension not supported, cannot fetch current cursor"
)
return None
# XFixes wants version negotiation before any other request, and
# this fetch runs whether or not the cursor monitor is up.
if not getattr(self, "_xfixes_negotiated", False):
self.xdisplay.xfixes_query_version()
self._xfixes_negotiated = True
screen = self.xdisplay.screen()
cursor_image = self.xdisplay.xfixes_get_cursor_image(screen.root)
return self._encode_cursor(cursor_image)
except Exception as e:
logger_webrtc_input.warning("exception from fetching current cursor image: %s", e)
return NoneparamselfReturns
typing.Optional[dict]func_encode_cursor(self, cursor) -> dictcursor_to_msg behind a one-entry cache keyed by the XFixes cursor serial (and the size cap the encode depends on): the monitor encodes each cursor once, off the loop, and the per-connect fetch — sync, on the loop — reuses that instead of encoding the same PNG again.
Source Code
def _encode_cursor(self, cursor: Any) -> dict:
"""cursor_to_msg behind a one-entry cache keyed by the XFixes cursor
serial (and the size cap the encode depends on): the monitor encodes
each cursor once, off the loop, and the per-connect fetch — sync, on
the loop — reuses that instead of encoding the same PNG again."""
key = (getattr(cursor, "cursor_serial", None), self.cursor_size_cap)
cached = self._cursor_msg_cache
if key[0] is not None and cached is not None and cached[0] == key:
return cached[1]
msg = self.cursor_to_msg(cursor)
if key[0] is not None:
self._cursor_msg_cache = (key, msg)
return msgparamselfparamcursorAnyReturns
dictfunc_cursor_image_to_pil(self, cursor) -> Image.ImageSource Code
def _cursor_image_to_pil(self, cursor: Any) -> Image.Image:
byte_data = b''.join(p.to_bytes(4, 'little') for p in cursor.cursor_image)
return Image.frombytes("RGBA", (cursor.width, cursor.height), byte_data, "raw", "BGRA")paramselfparamcursorAnyReturns
PIL.Image.PIL.Image.Imagefunccursor_to_msg(self, cursor) -> dictEncode an XFixes cursor image into the client cursor message.
Crops to the visible bounding box (clamping the hotspot with it), resizes down to the DPI-scaled cap, un-premultiplies alpha, and base64-encodes a PNG. Pure CPU work — callers on the event loop run it via a thread. XFixes pixels are premultiplied and are straightened only after any resize (resampling is linear per channel in premultiplied space), matching the pixelflux monitor's pipeline; the handle shares format_pixelflux_cursor's pixel-content space so the connect-time seed and the live path dedupe one shape to one client cache entry.
Source Code
def cursor_to_msg(self, cursor: Any) -> dict:
"""Encode an XFixes cursor image into the client cursor message.
Crops to the visible bounding box (clamping the hotspot with it),
resizes down to the DPI-scaled cap, un-premultiplies alpha, and
base64-encodes a PNG. Pure CPU work — callers on the event loop run it
via a thread. XFixes pixels are premultiplied and are straightened
only after any resize (resampling is linear per channel in
premultiplied space), matching the pixelflux monitor's pipeline; the
handle shares format_pixelflux_cursor's pixel-content space so the
connect-time seed and the live path dedupe one shape to one client
cache entry.
"""
if not cursor or cursor.width == 0 or cursor.height == 0:
return {
"curdata": "", "width": 0, "height": 0,
"hotx": 0, "hoty": 0, "handle": 0,
}
im = self._cursor_image_to_pil(cursor)
bbox = im.getbbox()
if bbox is None:
return {
"curdata": "", "width": 0, "height": 0,
"hotx": 0, "hoty": 0, "handle": 0,
}
cropped_im = im.crop(bbox)
left, upper, right, lower = bbox
# Browsers clamp a negative CSS cursor hotspot to 0 silently; clamping
# here, at the crop rebase, keeps both renderers agreeing.
new_hotx = max(0, cursor.xhot - left)
new_hoty = max(0, cursor.yhot - upper)
if cropped_im.width > self.cursor_size_cap or cropped_im.height > self.cursor_size_cap:
if self.cursor_debug:
logger_webrtc_input.info(f"Cursor ({cropped_im.width}x{cropped_im.height}) exceeds cap ({self.cursor_size_cap}x{self.cursor_size_cap}). Resizing.")
max_dim = max(cropped_im.width, cropped_im.height)
scale_factor = self.cursor_size_cap / max_dim
new_width = int(cropped_im.width * scale_factor)
new_height = int(cropped_im.height * scale_factor)
cropped_im = cropped_im.resize(
(new_width, new_height), resample=Image.Resampling.LANCZOS
)
new_hotx = min(round(new_hotx * scale_factor), max(0, new_width - 1))
new_hoty = min(round(new_hoty * scale_factor), max(0, new_height - 1))
# After the resize, never before (see the docstring).
cropped_im = unpremultiply_rgba(cropped_im)
with io.BytesIO() as f:
cropped_im.save(f, "PNG")
png_data = f.getvalue()
png_data_b64 = base64.b64encode(png_data)
return {
"curdata": png_data_b64.decode(),
"width": cropped_im.width,
"height": cropped_im.height,
"hotx": new_hotx,
"hoty": new_hoty,
"handle": cursor_content_handle(
cropped_im.tobytes(), cropped_im.width, cropped_im.height,
new_hotx, new_hoty),
}paramselfparamcursorAnyReturns
dictfuncstop_gamepad_servers(self) -> NoneSource Code
async def stop_gamepad_servers(self) -> None:
logger_webrtc_input.info("Stopping all gamepad instances.")
await self.__gamepad_disconnect()paramselfReturns
Nonefunc_keyboard_enqueue(self, item) -> NoneEnqueue input for the keyboard worker, evicting the oldest entry on overflow so a message flood can't grow the queue without bound. A held key orphaned by an evicted release is recovered by the stale sweep, which only covers keysyms tracked in pressed_keys — server-generated sequences are not tracked, so they go through _keyboard_enqueue_chord instead.
Source Code
def _keyboard_enqueue(self, item: tuple) -> None:
"""Enqueue input for the keyboard worker, evicting the oldest entry on
overflow so a message flood can't grow the queue without bound. A held key
orphaned by an evicted release is recovered by the stale sweep, which only
covers keysyms tracked in pressed_keys — server-generated sequences are not
tracked, so they go through _keyboard_enqueue_chord instead."""
try:
self.keyboard_queue.put_nowait(item)
except asyncio.QueueFull:
try:
evicted = self.keyboard_queue.get_nowait()
self.keyboard_queue.task_done()
# A queued server-side reset is awaited; its waiter must not
# sit out the timeout because a flood evicted it.
if evicted[0] == "kr" and evicted[1] is not None and not evicted[1].done():
evicted[1].set_result(None)
except asyncio.QueueEmpty:
pass
try:
self.keyboard_queue.put_nowait(item)
except asyncio.QueueFull:
logger_webrtc_input.warning("keyboard queue full; dropping input event.")paramselfparamitemtupleReturns
Nonefunc_keyboard_enqueue_chord(self, keys) -> NoneEnqueue a server-synthesized press/release sequence as ONE entry, so overflow eviction can only lose it whole.
keys is a sequence of (keysym, down) pairs the worker injects in order.
These keysyms never enter pressed_keys (no client 'ku' follows them), so a
release evicted on its own would leave the modifier held with nothing to
heal it — the stale sweep only covers tracked keys.
Source Code
def _keyboard_enqueue_chord(self, keys: Iterable[tuple]) -> None:
"""Enqueue a server-synthesized press/release sequence as ONE entry, so
overflow eviction can only lose it whole.
`keys` is a sequence of (keysym, down) pairs the worker injects in order.
These keysyms never enter pressed_keys (no client 'ku' follows them), so a
release evicted on its own would leave the modifier held with nothing to
heal it — the stale sweep only covers tracked keys.
"""
self._keyboard_enqueue(("chord", tuple(keys)))paramselfparamkeysIterable[tuple]Returns
Nonefunc_route_key_as_text(self, keysym) -> NoneRecord a keysym whose kd became buffered text, so its ku is swallowed instead of releasing a key that was never pressed. Bounded like pressed_keys, evicting the oldest entry.
Source Code
def _route_key_as_text(self, keysym: int) -> None:
"""Record a keysym whose kd became buffered text, so its ku is swallowed
instead of releasing a key that was never pressed. Bounded like
pressed_keys, evicting the oldest entry."""
if len(self._wl_text_routed) >= self.max_pressed_keys:
self._wl_text_routed.pop(next(iter(self._wl_text_routed)), None)
self._wl_text_routed[keysym] = TrueparamselfparamkeysymintReturns
Nonefunc_keyboard_worker(self) -> NoneWayland's single serialized key-injection loop.
Drains keyboard_queue so every rung (seat keymap, virtual-keyboard batch, clipboard paste) sees keys in client order. Character-bearing keysyms the seat cannot deliver — Unicode-plane keysyms off the seat, and legacy-plane keysyms the base layout lacks when a nested app compositor would re-translate a seat overlay bind — accumulate in a text buffer that flushes as one batch to the app compositor; the flush happens before any directly injected key so ordering still holds.
Chord-translated Cyrillic stays on the seat, where its QWERTY keysym resolves under any latin-based keymap; control and navigation keysyms spell no character, so the owner is never consulted for them. A modifier-held Unicode keysym normally goes through the seat so the app sees the chord, but under a nested app compositor, which never resolves seat-overlay keysyms, the text is typed plain rather than vanishing.
Source Code
async def _keyboard_worker(self) -> None:
"""Wayland's single serialized key-injection loop.
Drains keyboard_queue so every rung (seat keymap, virtual-keyboard
batch, clipboard paste) sees keys in client order. Character-bearing
keysyms the seat cannot deliver — Unicode-plane keysyms off the seat,
and legacy-plane keysyms the base layout lacks when a nested app
compositor would re-translate a seat overlay bind — accumulate in a
text buffer that flushes as one batch to the app compositor; the flush
happens before any directly injected key so ordering still holds.
Chord-translated Cyrillic stays on the seat, where its QWERTY keysym
resolves under any latin-based keymap; control and navigation keysyms
spell no character, so the owner is never consulted for them. A
modifier-held Unicode keysym normally goes through the seat so the app
sees the chord, but under a nested app compositor, which never
resolves seat-overlay keysyms, the text is typed plain rather than
vanishing.
"""
unicode_buffer = []
def native_inject():
"""Whether the keymap owner can deliver any keysym in order: only when
the capture compositor is also where the apps run, since a nested app
compositor never sees its overlay binds."""
return (bool(self.wayland_input
and hasattr(self.wayland_input, 'set_keymap_string'))
and not self._has_separate_app_compositor())
async def flush_buffer():
"""Type the buffered text (non-layout keysyms and IME strings) through
zwp_virtual_keyboard, or paste it via the clipboard on a compositor
without that protocol (KWin); never through the keymap overlay."""
if unicode_buffer:
combined_text = "".join(unicode_buffer)
unicode_buffer.clear()
try:
await self._wl_type_text(combined_text)
return
except Exception as e:
logger_webrtc_input.debug(
f"virtual-keyboard batch failed ({e}); pasting via clipboard")
if not await self._inject_text_via_clipboard(combined_text):
logger_webrtc_input.warning(
f"Batched text injection failed; {len(combined_text)} chars dropped.")
while True:
try:
if unicode_buffer:
try:
msg_type, data = await asyncio.wait_for(self.keyboard_queue.get(), timeout=0.05)
except asyncio.TimeoutError:
await flush_buffer()
continue
else:
msg_type, data = await self.keyboard_queue.get()
try:
keysym = data if msg_type in ("kd", "ku") else None
is_unicode_fallback = False
if keysym is not None:
is_unicode_fallback = (0xA0 <= keysym <= 0xFF) or keysym == 0x20AC or ((keysym & 0xFF000000) == 0x01000000)
if msg_type == "kd":
if (is_unicode_fallback and not native_inject()
and (not self.active_modifiers
or self._has_separate_app_compositor())):
unicode_codepoint = keysym & 0x00FFFFFF if (keysym & 0xFF000000) == 0x01000000 else keysym
try:
char_to_type = chr(unicode_codepoint)
unicode_buffer.append(char_to_type)
self._route_key_as_text(keysym)
continue
except ValueError:
pass
if keysym == 65288 and unicode_buffer:
unicode_buffer.pop()
continue
if (not is_unicode_fallback
and keysym is not None and not native_inject()
and keysym not in self.MODIFIER_KEYSYMS
and not ((self.active_modifiers & self.ACTION_MODIFIER_KEYSYMS)
and keysym in CYRILLIC_TO_QWERTY_KEYSYM)):
char_to_type = keysym_to_character(keysym)
if char_to_type is not None:
owner = await self._ensure_wayland_keymap_owner()
if owner is None or not owner.resolves(keysym):
unicode_buffer.append(char_to_type)
self._route_key_as_text(keysym)
continue
await flush_buffer()
if keysym in self.MODIFIER_KEYSYMS:
self.active_modifiers.add(keysym)
await self.send_x11_keypress(keysym, down=True)
elif msg_type == "ku":
if keysym is not None and self._wl_text_routed.pop(keysym, None):
# Its kd became text, whatever the topology is by now.
continue
if is_unicode_fallback and not native_inject():
# Buffered text was typed atomically; no held key.
continue
if keysym in self.MODIFIER_KEYSYMS:
self.active_modifiers.discard(keysym)
if keysym in self.atomically_typed_keys:
self.atomically_typed_keys.discard(keysym)
else:
await self.send_x11_keypress(keysym, down=False)
elif msg_type == "chord":
# Back-to-back, so no other queued key lands inside the chord.
await flush_buffer()
for chord_keysym, down in data:
if chord_keysym in self.MODIFIER_KEYSYMS:
if down:
self.active_modifiers.add(chord_keysym)
else:
self.active_modifiers.discard(chord_keysym)
await self.send_x11_keypress(chord_keysym, down=down)
elif msg_type == "kr":
# data is the future a server-side reset awaits; None from a client.
self._wl_text_routed.clear()
await flush_buffer()
try:
await self._reset_keyboard_wayland()
finally:
if data is not None and not data.done():
data.set_result(None)
elif msg_type == "co_end":
if native_inject():
# One keymap swap binds every missing keysym; per-char is the fallback.
typed = False
owner = await self._ensure_wayland_keymap_owner()
if owner is not None:
try:
typed = await asyncio.to_thread(
owner.type_text, data,
not (self.active_modifiers
& self.ACTION_MODIFIER_KEYSYMS))
except Exception as e:
logger_webrtc_input.warning(
f"Batched Wayland composition type failed; "
f"falling back per-char: {e}")
if not typed:
for ch in data:
cp = ord(ch)
ks = cp if 0x20 <= cp <= 0xFF else (0x01000000 | cp)
await self.send_x11_keypress(ks, down=True)
await self.send_x11_keypress(ks, down=False)
else:
unicode_buffer.append(data)
finally:
self.keyboard_queue.task_done()
except asyncio.CancelledError:
break
except Exception as e:
logger_webrtc_input.error(f"Error in keyboard worker: {e}", exc_info=True)paramselfReturns
Nonefunc_reset_multipart_clipboard(self) -> NoneReset all multi-part clipboard receive state to its idle defaults.
Used on completion/abort so no field (size, mime type, buffer, id, kind) is left stale to bleed into the next cbs/cws transfer.
Source Code
def _reset_multipart_clipboard(self) -> None:
"""Reset all multi-part clipboard receive state to its idle defaults.
Used on completion/abort so no field (size, mime type, buffer, id, kind)
is left stale to bleed into the next cbs/cws transfer.
"""
self.multipart_clipboard_buffer = None
self.multipart_clipboard_in_progress = False
self.multipart_clipboard_id = None
self.multipart_clipboard_kind = None
self.multipart_clipboard_total_size = 0
self.multipart_clipboard_mime_type = "text/plain"paramselfReturns
Nonefuncon_message(self, msg, display_id='primary', conn_id=None) -> NoneTransport entry point for one client message.
A malformed client message must not tear down the transport connection, so parse errors are logged and swallowed here.
Source Code
async def on_message(self, msg: str, display_id: str = 'primary',
conn_id: Any = None) -> None:
"""Transport entry point for one client message.
A malformed client message must not tear down the transport
connection, so parse errors are logged and swallowed here.
Args:
msg: Raw comma-delimited message string.
display_id: Transport-level id of the display whose channel
delivered the message (not a spoofable payload field).
conn_id: Transport connection identity, for per-connection state
(gamepad associations, clipboard debounce).
"""
try:
await self._dispatch_message(msg, display_id, conn_id)
except (IndexError, ValueError) as e:
logger_webrtc_input.warning(f"Malformed client message {msg[:64]!r}: {e}")
except Exception as e:
logger_webrtc_input.error(f"Error handling client message {msg[:64]!r}: {e}", exc_info=True)paramselfparammsgstrRaw comma-delimited message string.
paramdisplay_idstr= 'primary'Transport-level id of the display whose channel delivered the message (not a spoofable payload field).
paramconn_idAny= NoneTransport connection identity, for per-connection state (gamepad associations, clipboard debounce).
Returns
Nonefunc_dispatch_message(self, msg, display_id='primary', conn_id=None) -> NoneParse and act on one client message (the whole wire protocol lives here).
Source Code
async def _dispatch_message(self, msg: str, display_id: str = 'primary',
conn_id: Any = None) -> None:
"""Parse and act on one client message (the whole wire protocol lives here)."""
toks = msg.split(",")
msg_type = toks[0]
if msg_type == "pong":
if self.ping_start is None:
# A straggler from the other mode's client after a transport flip
# (WebRTC pings over its data channel; WS has no app-level ping).
logger_webrtc_input.debug("received pong before ping; ignoring")
return
self.on_ping_response(float("%.3f" % ((time.time() - self.ping_start) / 2 * 1000)))
elif msg_type == "kd":
keysym = int(toks[1])
# At the cap the oldest entry goes, not the new one: the key is
# injected below regardless, and untracked it would never auto-release.
if keysym in self.pressed_keys:
self.pressed_keys[keysym] = time.monotonic()
else:
if len(self.pressed_keys) >= self.max_pressed_keys:
oldest_keysym = min(self.pressed_keys, key=self.pressed_keys.get)
self.pressed_keys.pop(oldest_keysym, None)
self.pressed_keys[keysym] = time.monotonic()
self.reaped_atomic_keys.discard(keysym)
if self.is_wayland:
self._keyboard_enqueue(("kd", keysym))
else:
is_printable = (0x20 <= keysym <= 0xFF) or ((keysym & 0xFF000000) == 0x01000000)
if keysym in self.MODIFIER_KEYSYMS:
self.active_modifiers.add(keysym)
if is_printable and not self.active_modifiers:
unicode_codepoint = keysym & 0x00FFFFFF if (keysym & 0xFF000000) == 0x01000000 else keysym
try:
char_to_type = chr(unicode_codepoint)
if not char_to_type.isalpha() and char_to_type != ' ':
await self.on_message(f"co,end,{char_to_type}")
self.atomically_typed_keys.add(keysym)
else:
await self.send_x11_keypress(keysym, down=True)
except (ValueError, TypeError):
await self.send_x11_keypress(keysym, down=True)
else:
await self.send_x11_keypress(keysym, down=True)
# Arm auto-repeat: pop+insert moves the key to the end (only the
# newest repeats); modifiers never; atomic keys are armed too.
if (self.key_repeat_enabled and keysym not in self.MODIFIER_KEYSYMS):
self.key_repeat_state.pop(keysym, None)
self.key_repeat_state[keysym] = time.monotonic() + self.key_repeat_delay
else:
self.key_repeat_state.pop(keysym, None)
elif msg_type == "ku":
keysym = int(toks[1])
self.pressed_keys.pop(keysym, None)
self.key_repeat_state.pop(keysym, None)
if self.is_wayland:
self._keyboard_enqueue(("ku", keysym))
else:
if keysym in self.MODIFIER_KEYSYMS:
self.active_modifiers.discard(keysym)
if keysym in self.reaped_atomic_keys:
# Already reaped by the sweep; a keyup now would be spurious.
self.reaped_atomic_keys.discard(keysym)
elif keysym in self.atomically_typed_keys:
# Never physically held on X11; nothing to release.
self.atomically_typed_keys.discard(keysym)
else:
await self.send_x11_keypress(keysym, down=False)
elif msg_type == "kr":
if self.is_wayland:
self._keyboard_enqueue(("kr", None))
else:
await self.reset_keyboard()
elif msg_type == "kh":
# Refresh only, no injection. Atomic keys are refreshed like any
# other: the repeat loop pauses on a stale heartbeat, so skipping
# them would kill their auto-repeat before the first repeat is due.
now = time.monotonic()
# Bounded: a client could otherwise pack one frame with tens of
# thousands of tokens.
for tok in toks[1:1 + self.max_pressed_keys]:
try:
keysym = int(tok)
except ValueError:
continue
if keysym in self.pressed_keys:
self.pressed_keys[keysym] = now
elif msg_type in ["m", "m2"]:
relative = msg_type == "m2"
# Dropped rather than defaulted: a default would warp to the origin.
try: x, y, button_mask, scroll_magnitude = [int(i) for i in toks[1:]]
except (ValueError, IndexError): return
try: await self.send_x11_mouse(x, y, button_mask, scroll_magnitude, relative, display_id=display_id)
except Exception as e: logger_webrtc_input.warning(f"Failed to set mouse cursor: {e}")
elif msg_type == "p": await self.on_mouse_pointer_visible(bool(int(toks[1])))
elif msg_type == "vb":
try:
# kbps; per display, named by the delivering channel.
bitrate = int(toks[1])
if bitrate <= 0:
return
await self.on_video_encoder_bit_rate(bitrate, display_id)
except Exception as e:
logger_webrtc_input.error(f"Error video bitrate change: {e}")
elif msg_type == "ab":
try:
bitrate = int(toks[1])
if bitrate <= 0:
return
await self.on_audio_encoder_bit_rate(bitrate)
except Exception as e:
logger_webrtc_input.error(f"Error audio bitrate change: {e}")
elif msg_type == "js":
# Enforced server-side so a client cannot inject controller input
# whatever its own UI state.
if not settings.gamepad_enabled[0]:
return
cmd = toks[1]
gamepad_idx = int(toks[2])
if not (0 <= gamepad_idx < self.num_gamepads):
logger_webrtc_input.error(f"Client message for gamepad index {gamepad_idx} is out of range (0-{self.num_gamepads-1}).")
return
target_gamepad_instance = self.gamepad_instances.get(gamepad_idx)
if not target_gamepad_instance:
logger_webrtc_input.error(
f"CRITICAL: No persistent SelkiesGamepad instance found for index {gamepad_idx} in on_message. "
f"Gamepad system may not be initialized correctly."
)
return
if cmd == "c":
try: client_name_decoded = base64.b64decode(toks[3]).decode('latin-1', 'ignore')[:255]
except Exception as e: client_name_decoded = f"ClientGamepad{gamepad_idx}"; logger_webrtc_input.warning(f"Error decoding client gamepad name: {e}")
client_num_axes, client_num_btns = int(toks[4]), int(toks[5])
await self.__gamepad_connect(gamepad_idx, client_name_decoded, client_num_btns, client_num_axes, conn_id=conn_id)
elif cmd == "d":
await self.__gamepad_disconnect(gamepad_idx)
elif cmd == "b":
button_num = int(toks[3])
button_val = float(toks[4])
target_gamepad_instance.send_event(button_num, button_val, is_button_event=True)
elif cmd == "a":
axis_num = int(toks[3])
axis_val = float(toks[4])
target_gamepad_instance.send_event(axis_num, axis_val, is_button_event=False)
elif cmd == "h":
# Held-state heartbeat: refresh only (no injection), like 'kh'.
self.gamepad_heartbeats[gamepad_idx] = time.monotonic()
else: logger_webrtc_input.warning(f"Unhandled joystick command for slot {gamepad_idx}: js {cmd}")
elif msg_type == "cws":
if self.enable_clipboard in ["true", "in"]:
try:
transfer_id = toks[1]
declared_size = int(toks[2])
if declared_size < 0 or declared_size > MULTIPART_CLIPBOARD_MAX_SIZE:
logger_webrtc_input.error(f"Rejecting multi-part clipboard write: declared size {declared_size} out of bounds (max {MULTIPART_CLIPBOARD_MAX_SIZE}).")
return
if self.multipart_clipboard_in_progress and transfer_id != self.multipart_clipboard_id:
logger_webrtc_input.warning(f"Aborting previous in-progress clipboard transfer {self.multipart_clipboard_id} for new transfer {transfer_id}.")
self.multipart_clipboard_id = transfer_id
self.multipart_clipboard_kind = "text"
self.multipart_clipboard_total_size = declared_size
self.multipart_clipboard_mime_type = "text/plain"
self.multipart_clipboard_buffer = io.BytesIO()
self.multipart_clipboard_in_progress = True
logger_webrtc_input.info(f"Starting multi-part text clipboard receive, total size: {self.multipart_clipboard_total_size}")
except Exception as e:
logger_webrtc_input.error(f"Invalid cws message: {msg}, error: {e}")
else:
logger_webrtc_input.warning("Rejecting multi-part clipboard write: inbound clipboard disabled.")
elif msg_type == "cbs":
# Direction gate and binary gate: the server enforces its own policy.
if self.enable_clipboard in ["true", "in"] and self.enable_binary_clipboard in ["true", "in"]:
try:
transfer_id = toks[1]
declared_size = int(toks[3])
if declared_size < 0 or declared_size > MULTIPART_CLIPBOARD_MAX_SIZE:
logger_webrtc_input.error(f"Rejecting multi-part clipboard write: declared size {declared_size} out of bounds (max {MULTIPART_CLIPBOARD_MAX_SIZE}).")
return
if self.multipart_clipboard_in_progress and transfer_id != self.multipart_clipboard_id:
logger_webrtc_input.warning(f"Aborting previous in-progress clipboard transfer {self.multipart_clipboard_id} for new transfer {transfer_id}.")
self.multipart_clipboard_id = transfer_id
self.multipart_clipboard_kind = "binary"
self.multipart_clipboard_mime_type = toks[2]
self.multipart_clipboard_total_size = declared_size
self.multipart_clipboard_buffer = io.BytesIO()
self.multipart_clipboard_in_progress = True
logger_webrtc_input.info(f"Starting multi-part binary clipboard receive ({self.multipart_clipboard_mime_type}), total size: {self.multipart_clipboard_total_size}")
except Exception as e:
logger_webrtc_input.error(f"Invalid cbs message: {msg}, error: {e}")
else:
logger_webrtc_input.warning("Rejecting multi-part clipboard write: inbound clipboard disabled.")
elif msg_type == "cwd" or msg_type == "cbd":
expected_kind = "text" if msg_type == "cwd" else "binary"
# Token count first: a malformed chunk raising mid-transfer would
# leave the multipart state half-open, accumulating until overflow.
if len(toks) < 3:
logger_webrtc_input.warning(f"Malformed clipboard chunk ({msg_type}): missing fields; aborting transfer.")
self._reset_multipart_clipboard()
elif not (self.multipart_clipboard_in_progress and toks[1] == self.multipart_clipboard_id and self.multipart_clipboard_kind == expected_kind):
logger_webrtc_input.warning(f"Ignoring mismatched clipboard chunk ({msg_type}): id/kind does not match active transfer.")
else:
try:
chunk_data = base64.b64decode(toks[2])
if self.multipart_clipboard_buffer.tell() + len(chunk_data) > self.multipart_clipboard_total_size:
logger_webrtc_input.error("Multi-part clipboard exceeded its declared size; aborting transfer.")
self._reset_multipart_clipboard()
return
self.multipart_clipboard_buffer.write(chunk_data)
except Exception as e:
logger_webrtc_input.error(f"Failed to process clipboard data chunk: {e}")
self._reset_multipart_clipboard()
elif msg_type == "cwe" or msg_type == "cbe":
expected_kind = "text" if msg_type == "cwe" else "binary"
# Token count first, so a malformed end does not raise mid-state.
if len(toks) < 2:
logger_webrtc_input.warning(f"Malformed clipboard end ({msg_type}): missing id; aborting transfer.")
self._reset_multipart_clipboard()
elif not (self.multipart_clipboard_in_progress and toks[1] == self.multipart_clipboard_id and self.multipart_clipboard_kind == expected_kind):
logger_webrtc_input.warning(f"Ignoring mismatched clipboard end ({msg_type}): id/kind does not match active transfer.")
else:
received_size = self.multipart_clipboard_buffer.tell()
if received_size != self.multipart_clipboard_total_size:
logger_webrtc_input.error(f"Multi-part clipboard size mismatch. Expected {self.multipart_clipboard_total_size}, got {received_size}. Aborting.")
else:
logger_webrtc_input.info(f"Finished multi-part clipboard receive. Total size: {received_size}")
data = self.multipart_clipboard_buffer.getvalue()
mime_type = self.multipart_clipboard_mime_type
# Awaited in-line: a paste keystroke right behind the transfer
# must find the clipboard set. Bytes pass straight through; a
# multi-MB decode and re-encode on the loop would be redundant.
if await self.write_clipboard(data, mime_type=mime_type):
if mime_type == "text/plain":
logger_webrtc_input.info(f"Set multi-part clipboard content, length: {len(data)}")
else:
logger_webrtc_input.info(f"Set multi-part binary clipboard content ({mime_type}), size: {len(data)} bytes")
self._reset_multipart_clipboard()
elif msg_type == "cr":
if self.enable_clipboard in ["true", "out"]:
data, mime_type = await self.read_clipboard(use_binary=self.enable_binary_clipboard in ["true", "out"])
if data:
# Tagged (reply_to) so the client treats it cache-only without
# its connect-time 5 s heuristic, and sent to this client alone:
# unasked-for, another client would cache content it never pastes.
await self.send_clipboard_data(
data, mime_type, reply_to="cr", conn_id=conn_id)
else:
# Reply even when empty: the tag settles the client's connect-time
# fetch, so a real change seconds later is not taken for the snapshot.
logger_webrtc_input.debug("No clipboard content; sending empty tagged reply")
await self.send_clipboard_data(
"", "text/plain", reply_to="cr", conn_id=conn_id)
else: logger_webrtc_input.warning("Rejecting clipboard read: outbound clipboard disabled.")
elif msg_type == "REQUEST_CLIPBOARD":
if self.enable_clipboard in ["true", "out"]:
now = time.monotonic()
# display_id stands in when no per-connection id is supplied.
clip_key = conn_id if conn_id is not None else display_id
# Bounded across reconnecting connections.
if len(self._last_clipboard_request_ts) > 64:
self._last_clipboard_request_ts = {
k: ts for k, ts in self._last_clipboard_request_ts.items()
if now - ts < self._clipboard_request_debounce
}
if now - self._last_clipboard_request_ts.get(clip_key, 0.0) < self._clipboard_request_debounce:
logger_webrtc_input.debug("Debouncing REQUEST_CLIPBOARD (too frequent).")
else:
self._last_clipboard_request_ts[clip_key] = now
use_binary = self.enable_binary_clipboard in ["true", "out"]
async def _send_requested_clipboard():
"""Read and push the clipboard as a task, so a slow read
cannot stall the dispatch loop. A read that still matches
the baseline races the injected Ctrl+C (the app has not
published the new selection yet), so it waits briefly for
the owner change — without consuming the edge the monitor
loop broadcasts on — and re-reads."""
try:
data, mime_type = await self.read_clipboard(use_binary=use_binary)
data_bytes = (data.encode('utf-8')
if isinstance(data, str) else data)
if data_bytes is not None and data_bytes == self._clipboard_last_bytes:
monitor = self._x11_clipboard_monitor
if monitor is not None and monitor.alive():
await monitor.peek_change(0.15)
else:
await asyncio.sleep(0.15)
data, mime_type = await self.read_clipboard(use_binary=use_binary)
if data:
await self.send_clipboard_data(data, mime_type,
conn_id=conn_id)
else:
logger_webrtc_input.debug("No clipboard content to send on REQUEST_CLIPBOARD.")
except Exception as e:
logger_webrtc_input.warning(f"REQUEST_CLIPBOARD read failed: {e}")
self._spawn_task(_send_requested_clipboard())
else:
logger_webrtc_input.warning("Rejecting REQUEST_CLIPBOARD: outbound clipboard disabled.")
elif msg_type == "cb":
# Same double gate as cbs.
if self.enable_clipboard in ["true", "in"] and self.enable_binary_clipboard in ["true", "in"]:
try:
_, mime_type, b64_data = toks
data_bytes = base64.b64decode(b64_data)
# In-line so a paste keystroke right behind it pastes this content.
if await self.write_clipboard(data_bytes, mime_type=mime_type):
logger_webrtc_input.info(f"Set binary clipboard content ({mime_type}), size: {len(data_bytes)} bytes")
except Exception as e:
logger_webrtc_input.error(f"Binary clipboard write error: {e}")
else:
logger_webrtc_input.warning("Rejecting binary clipboard write: inbound binary clipboard disabled.")
elif msg_type == "cw":
if self.enable_clipboard in ["true", "in"]:
try:
data = base64.b64decode(toks[1]).decode("utf-8", 'ignore')
# In-line for paste-after-copy ordering (see the cb branch).
if await self.write_clipboard(data):
logger_webrtc_input.info(f"Set clipboard content, length: {len(data)}")
except Exception as e:
logger_webrtc_input.error(f"Clipboard decode error: {e}")
return
else:
logger_webrtc_input.warning("Rejecting clipboard write: inbound clipboard disabled.")
elif msg_type == "r":
res = toks[1]
if re.fullmatch(r"^\d+x\d+$", res):
# Passed through verbatim: even-dim normalization lives in
# parse_resize_dims so both transports realize it identically.
_r = self.on_resize(res, display_id)
if asyncio.iscoroutine(_r): await _r
else: logger_webrtc_input.warning(f"Rejecting resolution change, invalid: {res}")
elif msg_type == "s":
scale = toks[1]
if re.fullmatch(r"^\d+(\.\d+)?$", scale):
_s = self.on_scaling_ratio(float(scale))
if asyncio.iscoroutine(_s): await _s
else: logger_webrtc_input.warning(f"Rejecting scaling change, invalid: {scale}")
elif msg_type == "cmd":
if not settings.command_enabled[0]:
logger_webrtc_input.warning("Received 'cmd' message, but command execution is disabled by server settings.")
return
if len(toks) > 1:
command_to_run = ",".join(toks[1:])
logger_webrtc_input.info(f"Attempting to execute command: '{command_to_run}'")
async def _notify_cmd_error(text, conn_id=conn_id):
self.send_command_status(f"command_error,{text}", conn_id)
async def _notify_cmd_done(cmd, conn_id=conn_id):
self.send_command_status(f"command_done,{cmd}", conn_id)
await run_client_command(
command_to_run, logger_webrtc_input, notify=_notify_cmd_error,
env=self.app_launch_env(), done=_notify_cmd_done)
else:
logger_webrtc_input.warning("Received 'cmd' message without a command string.")
elif msg_type == "_arg_fps":
try:
fps = int(toks[1])
if fps <= 0:
return
await self.on_set_fps(fps, display_id)
except Exception as e:
logger_webrtc_input.error(f"Error fps change: {e}")
elif msg_type == "_arg_resize":
if len(toks) == 3:
enabled, res_str = toks[1].lower() == "true", toks[2]
enable_res = None
if re.fullmatch(r"^\d+x\d+$", res_str):
w,h = [int(i)+int(i)%2 for i in res_str.split("x")]; enable_res = f"{w}x{h}"
elif res_str: logger_webrtc_input.warning(f"Invalid resolution for enable_resize: {res_str}")
self.on_set_enable_resize(enabled, enable_res)
else: logger_webrtc_input.error("Invalid _arg_resize command format")
elif msg_type == "_f":
try: self.on_client_fps(int(toks[1]))
except (ValueError, IndexError): logger_webrtc_input.error(f"Failed to parse client FPS: {toks}")
elif msg_type == "_l":
try: self.on_client_latency(int(toks[1]))
except (ValueError, IndexError): logger_webrtc_input.error(f"Failed to parse client latency: {toks}")
elif msg_type in ["_stats_video", "_stats_audio"]:
try: await self.on_client_webrtc_stats(msg_type, ",".join(toks[1:]))
except (ValueError, IndexError): logger_webrtc_input.error("Failed to parse WebRTC Statistics")
elif msg_type == "co" and toks[1] == "end":
try:
text_to_type = msg[7:]
if self.is_wayland:
self._keyboard_enqueue(("co_end", text_to_type))
elif self._type_text_xtest(
text_to_type,
neutralize=not (self.active_modifiers
& self.ACTION_MODIFIER_KEYSYMS)):
# Typed in-process; xdotool below only when that fails.
pass
else:
cmd = ["xdotool", "type", "--", text_to_type]
process = await subprocess.create_subprocess_exec(
*cmd,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE
)
await self._communicate_or_kill(process, 0.5, "xdotool type co,end")
except Exception as e: logger_webrtc_input.warning(f"Error with co,end type: {e}")
elif msg_type == "_ebc":
try:
enable = toks[1].lower() == "true"
self._spawn_task(self.update_binary_clipboard_setting(enable))
except Exception as e:
logger_webrtc_input.error(f"Error updating binary clipboard setting: {e}")
elif msg_type == "_rc":
try:
mode = toks[1].strip().lower()
rc_mode = RateControlMode(mode)
self._spawn_task(self.on_update_rate_control_mode(rc_mode, display_id))
except Exception as e:
logger_webrtc_input.error(f"Error updating rate control mode: {e}")
elif msg_type == "_crf":
try:
crf_value = int(toks[1])
if not (0 <= crf_value <= 51):
logger_webrtc_input.warning(f"CRF value out of range (0-51): {crf_value}")
return
self._spawn_task(self.on_update_crf(crf_value, display_id))
except Exception as e:
logger_webrtc_input.error(f"Error updating CRF value: {e}")
elif toks[0].startswith("SETTINGS"):
settings_data = ','.join(toks[1:]) if len(toks) > 1 else ""
logger_webrtc_input.info(f"Received SETTINGS message: {settings_data}")
try:
settings_json = json.loads(settings_data)
# Applied to the delivering channel's display (not a spoofable
# payload field) and inline, so a resize behind it sees this policy.
applied = self.on_update_settings(settings_json, display_id)
if asyncio.iscoroutine(applied):
await applied
except Exception as e:
logger_webrtc_input.error(f"Failed to parse SETTINGS data: {e}")
elif toks[0] == "SET_NATIVE_CURSOR_RENDERING":
# WS-protocol alias of "p,N"; both map to the capture_cursor tunable.
try:
await self.on_mouse_pointer_visible(toks[1].strip().lower() in ("1", "true"))
except (IndexError, ValueError) as e:
logger_webrtc_input.warning(f"Malformed SET_NATIVE_CURSOR_RENDERING message: {msg[:60]}, error: {e}")
elif toks[0] == "REQUEST_KEYFRAME":
# Viewer-allowed IDR request, routed by delivering channel like RTCP PLI.
_kf = self.on_request_keyframe(display_id)
if asyncio.iscoroutine(_kf): await _kf
else:
logger_webrtc_input.info(f"Unknown data channel message: {msg[:100]}")paramselfparammsgstrparamdisplay_idstr= 'primary'paramconn_idAny= NoneReturns
Nonefuncinitialize_upload_dir(self) -> NoneResolve and create the client-upload directory, refusing unsafe roots.
Source Code
def initialize_upload_dir(self) -> None:
"""Resolve and create the client-upload directory, refusing unsafe roots."""
if self.upload_dir in ["/sys", "/proc", "/dev"]:
logger_webrtc_input.info("Can not initialize upload directory at /sys /proc /dev locations")
return
if not self.upload_dir:
logger_webrtc_input.info("Upload dir is empty")
return
if self.upload_dir == "~/Desktop":
self.upload_dir_path = os.path.expanduser(self.upload_dir)
else:
self.upload_dir_path = self.upload_dir
try:
os.makedirs(self.upload_dir_path, exist_ok=True)
logger_webrtc_input.info(f"Upload directory ensured: {self.upload_dir_path}")
except OSError as e:
logger_webrtc_input.error(f"Could not create upload directory {self.upload_dir_path}: {e}")
self.upload_dir_path = NoneparamselfReturns
None