"""Decode the browser input wire format into :class:`InputRouter` calls.
Two paths deliver input from a browser: the exported static page
(``simvx.web``) and the streaming dev client (``simvx.graphics.streaming``).
Both page scripts serialise the same JSON events, so both decode here.
Event shapes, all keys optional with the defaults shown:
================ ====================================================
``key`` ``code`` (engine key code), ``pressed``, ``repeat``
``char`` ``codepoint``
``pointer`` ``id``, ``ptype`` (``mouse``/``touch``/``pen``),
``primary``, ``phase`` (0=down, 1=up, 2=move,
3=cancel, 4=enter, 5=leave), ``x``, ``y``, ``button``
(DOM ordering, -1 for none), ``buttons`` (the DOM
bitmask), ``pressure``, ``tiltX``/``tiltY`` and
``twist`` in degrees, ``tangentialPressure``,
``width``/``height`` of the contact in engine units,
or ``dx``/``dy`` with ``relative`` under pointer lock
``scroll`` ``dx``, ``dy`` (positive ``dy`` scrolls up)
``focus`` ``focused``; only ``false`` carries information
``controllers`` ``pads``: the whole pad list this frame (see below)
``capabilities`` one key per :mod:`~simvx.core.input.capabilities`
fact the page queried, each a bool
================ ====================================================
A ``pointer`` with phase 3 is the system taking the pointer away: the browser
fires ``pointercancel`` for an edge gesture, a notification, or a gesture the
page chrome decided to own. A ``focus`` event with ``focused: false`` is the
page losing input, sent from ``blur`` and from ``visibilitychange``; regaining
focus needs no event, because nothing is held on the way back in.
**A pad is polled, not delivered.** ``navigator.getGamepads()`` is a snapshot
the page reads in its frame callback, so one ``controllers`` event carries every
pad the page can see and a pad missing from it has gone: the decode is one
poll transaction, exactly as a desktop backend's is, and unplugging a
controller releases what it held. A page that has never seen a pad sends no
event at all, which is the difference between "no pads" and "cannot see pads":
an empty snapshot from a page that never polls would read as every pad
leaving at once.
Each entry in ``pads`` is ``index`` (the browser's slot, which is the pad id),
``id`` (what the browser calls the device), ``mapping`` (``"standard"`` when
the browser has laid the device out as a controller), ``pressed`` (one 0 or 1 per
button), ``buttons`` (one value per button, which is how an analog trigger
reports) and ``axes``.
**The wire is older than the pages sending it.** A page exported before the
pointer event existed speaks the three the DOM had then, and those bundles are
static files nobody re-exports, so they are decoded here too:
=========== =========================================================
``mouse`` ``button`` (DOM ordering), ``pressed``
``mousemove````x``/``y``, or ``dx``/``dy`` with ``relative`` under pointer lock
``touch`` ``id``, ``action`` (0=down, 1=up, 2=move, 3=cancel), ``x``, ``y``,
``pressure``
=========== =========================================================
Any event may carry ``ctrl``/``shift``/``alt``/``meta`` modifier flags, and a
``time``: the DOM event's own ``timeStamp`` verbatim, in milliseconds on the
page's ``performance.timeOrigin`` clock. A page that sends none leaves the
event to be stamped at receipt, which is what a page exported before the field
existed does.
"""
import logging
import math
import re
from collections.abc import Iterable
from time import monotonic
from typing import Any, TypedDict
from . import capabilities, controller
from .enums import PenAxis, PointerPhase, PointerTool, PointerType
from .layouts import DOM_MOUSE_LAYOUT, W3C_PEN_LAYOUT
from .router import InputRouter
log = logging.getLogger(__name__)
#: ``PointerEvent.pointerType`` verbatim. The DOM's three names are the
#: engine's three, so the mapping is an identity check rather than a
#: translation, and anything else the browser invents is treated as a mouse.
_POINTER_TYPES = {"mouse": PointerType.MOUSE, "touch": PointerType.TOUCH, "pen": PointerType.PEN}
#: Phases on the wire, which are also the older ``touch`` event's actions: the
#: two encodings were already the same four values in the same order, and a pen
#: coming into and going out of range added two more.
_WIRE_PHASES = {
0: PointerPhase.DOWN,
1: PointerPhase.UP,
2: PointerPhase.MOVE,
3: PointerPhase.CANCEL,
4: PointerPhase.ENTER,
5: PointerPhase.LEAVE,
}
#: The DOM's button number for a pen's eraser end, and its bit in ``buttons``.
_DOM_ERASER_BUTTON = 5
_DOM_ERASER_BIT = 32
class _Modifiers(TypedDict):
"""Exactly the four modifier arguments every router call takes.
Named rather than a ``dict[str, bool]`` because these are splatted into
router signatures that carry other keywords: a plain string-keyed dict is a
claim that any of them may be filled with a bool, which is how a modifier
flag could land in ``label`` or ``device_id`` with nothing to catch it.
"""
ctrl: bool
shift: bool
alt: bool
meta: bool
class _PageClock:
"""Maps a page's ``DOMHighResTimeStamp`` onto the engine's monotonic clock.
A DOM event's ``timeStamp`` is milliseconds since that page's
``performance.timeOrigin``: not the epoch, and not ``time.monotonic()``,
which is what the engine times events on. The two are reconciled with a
single offset, captured from the first stamped event as ``monotonic()`` at
the moment of receipt minus that event's own stamp, and added to every
stamp after it.
Deltas between events are then exactly the deltas the page measured, and
the absolute value sits on the engine's clock. The whole error is one
constant: the first event's transport latency, a JavaScript call for the
exported page and a network hop plus a queue wait for the streaming
client. Re-fitting the offset per batch would trade that fixed constant for
a wobble on every cross-batch delta, of the difference between two batches'
latencies, which is the measurement the timestamps exist to provide.
The conversion belongs here rather than in the page scripts for the reason
the button remap above does: the page transcribes the DOM and the decoder
interprets the wire. It also cannot live anywhere else, because the offset
can only be measured where ``monotonic()`` can be read and a page has no
access to it. Unlike the button remap it wins nothing for pages already
exported, which send no ``time`` at all.
"""
__slots__ = ("_offset",)
def __init__(self) -> None:
self._offset: float | None = None
def to_monotonic(self, page_ms: float) -> float:
seconds = page_ms / 1000.0
if self._offset is None:
self._offset = monotonic() - seconds
return seconds + self._offset
def reset(self) -> None:
self._offset = None
# One page feeds one process, so the offset is process-wide, the same way
# ``Input.capabilities`` is: it describes the clock of the browser on the other
# end of the wire.
_page_clock = _PageClock()
def _mods(evt: dict[str, Any]) -> _Modifiers:
"""Extract modifier flags from an event, defaulting to not-held."""
return {
"ctrl": bool(evt.get("ctrl", False)),
"shift": bool(evt.get("shift", False)),
"alt": bool(evt.get("alt", False)),
"meta": bool(evt.get("meta", False)),
}
def _time(evt: dict[str, Any]) -> float | None:
"""The event's time on the monotonic clock, or None for a page that sent none."""
stamp = evt.get("time")
return None if stamp is None else _page_clock.to_monotonic(float(stamp))
def _drop_touch_synthesized_mouse(events: list[dict[str, Any]]) -> list[dict[str, Any]]:
"""Remove the mouse events a page emits alongside its touch events.
Page scripts have historically sent a synthetic left-button ``mouse`` event
and a ``mousemove`` next to every touch, so that pages predating touch
support still received clicks. A finger is a pointer of its own and the
primary one already stands in for the mouse, so those duplicates would
dispatch every tap twice. Touch and
real mouse input do not meaningfully interleave inside one frame's batch,
so dropping them whenever the batch carries touch is safe, and it keeps
already-exported pages working correctly.
Nothing timed is lost with them: the finger is stamped with its own time,
so the dropped duplicate would only have carried the same instant a second
time.
"""
if not any(evt.get("type") == "touch" for evt in events):
return events
return [
evt
for evt in events
if not (evt.get("type") == "mousemove" or (evt.get("type") == "mouse" and evt.get("button", 0) == 0))
]
def _touch_pressure(evt: dict[str, Any]) -> float | None:
"""A finger's pressure, or None where the screen did not measure one.
The DOM reports 0.5 for a finger on hardware that cannot measure pressure,
some browsers report 0, and pages exported before this decode sent 1.0 in
either case; none of the three is a measurement, and a screen that does
measure will report something else on the next sample.
"""
raw = evt.get("pressure")
if raw is None:
return None
pressure = float(raw)
return None if pressure in (0.0, 0.5, 1.0) else pressure
def _contact_size(evt: dict[str, Any]) -> tuple[float, float] | None:
"""The contact's width and height, or None where the browser has none.
The DOM reports 1 by 1 for a device that cannot say, which is not a
contact anyone has.
"""
width, height = float(evt.get("width", 0.0)), float(evt.get("height", 0.0))
return (width, height) if width > 1.0 or height > 1.0 else None
def _route_pen(
router: InputRouter, evt: dict[str, Any], phase: PointerPhase, x: float, y: float, when: float | None
) -> None:
"""One pen sample: hovering, touching, a barrel button, or entering and leaving range.
The eraser end arrives as the DOM's button 5 and bit 32 of ``buttons``,
which the engine models as the tool rather than as a button, because on
every other platform it is a tool. Tilt and twist arrive in degrees and are
stored in radians. The DOM reports zero for an axis the pen does not have,
which cannot be told from a pen held exactly upright, so a zero is passed
on as "not measured" and the first real reading declares the axis.
"""
known = capabilities.current().pen_axes
dom_button = int(evt.get("button", -1))
eraser = dom_button == _DOM_ERASER_BUTTON or bool(int(evt.get("buttons", 0)) & _DOM_ERASER_BIT)
raw_pressure = float(evt.get("pressure", 0.0))
# 0.5 is the DOM's "pressed, cannot measure", unless the pen has already
# shown it measures, in which case it is a real half press.
measured = raw_pressure > 0.0 and (raw_pressure != 0.5 or PenAxis.PRESSURE in known)
tilt_x, tilt_y = float(evt.get("tiltX", 0.0)), float(evt.get("tiltY", 0.0))
twist = float(evt.get("twist", 0.0))
tangential = float(evt.get("tangentialPressure", 0.0))
# The pen's own DOM index goes through as it is, named by the W3C layout
# the page declared; the eraser's press is the contact of the eraser tool.
button = 0
if phase is PointerPhase.DOWN or phase is PointerPhase.UP:
if dom_button < 0:
return
if dom_button != _DOM_ERASER_BUTTON:
button = dom_button
router.pen(
int(evt.get("id", 0)),
phase,
x,
y,
button=button,
pressure=raw_pressure if measured else None,
tilt=(math.radians(tilt_x), math.radians(tilt_y)) if tilt_x or tilt_y or PenAxis.TILT in known else None,
twist=math.radians(twist) if twist or PenAxis.TWIST in known else None,
tangential_pressure=tangential if tangential or PenAxis.TANGENTIAL in known else None,
tool=PointerTool.ERASER if eraser else PointerTool.TIP,
time=when,
**_mods(evt),
)
def _route_pointer(router: InputRouter, evt: dict[str, Any], when: float | None) -> None:
"""Route one W3C pointer event: a mouse, a finger or a pen, all one shape.
The DOM unified the three and so does the engine: each is a pointer of its
own and all three reach :meth:`InputRouter.publish`. A finger goes through
:meth:`InputRouter.touch` and a pen through :meth:`InputRouter.pen`, which
give each its own pointer id and decide which one stands in for the mouse;
the mouse drives the cursor directly.
A pen sends every sample the browser coalesced, because the intermediate
samples of a stroke are the pressure curve and nothing else records it.
The engine still delivers one motion event per frame for each pointer, and
each sample only to a handler declared with ``samples=True``, so sending
them costs a stroke nothing it does not need.
"""
phase = _WIRE_PHASES.get(int(evt.get("phase", 2)), PointerPhase.MOVE)
ptype = _POINTER_TYPES.get(str(evt.get("ptype", "mouse")), PointerType.MOUSE)
x, y = float(evt.get("x", 0.0)), float(evt.get("y", 0.0))
if evt.get("relative"):
# Under pointer lock there is no position, only motion, and whatever
# moved it is the cursor.
if phase is PointerPhase.MOVE:
router.mouse_motion_relative(float(evt.get("dx", 0.0)), float(evt.get("dy", 0.0)), time=when)
return
elif ptype is PointerType.TOUCH:
if phase is not PointerPhase.ENTER and phase is not PointerPhase.LEAVE:
router.touch(
int(evt.get("id", 0)),
phase,
x,
y,
_touch_pressure(evt),
contact_size=_contact_size(evt),
time=when,
)
return
elif ptype is PointerType.PEN:
_route_pen(router, evt, phase, x, y, when)
return
if phase is PointerPhase.MOVE:
router.mouse_motion(x, y, time=when)
return
if phase is PointerPhase.CANCEL:
# The browser took the pointer away mid-gesture. Whoever holds it is
# told the gesture ended; no release edge fires, because the player
# never let go.
router.cancel_pointers()
return
if phase is not PointerPhase.DOWN and phase is not PointerPhase.UP:
return
button = int(evt.get("button", 0))
if button < 0:
return
# The press carries where the pointer is, so the position lands first.
# Under pointer lock there is no position to land.
if not evt.get("relative"):
router.mouse_motion(x, y, time=when)
router.mouse_button(button, phase is PointerPhase.DOWN, time=when, **_mods(evt))
#: The W3C standard controller's button order, onto the engine's names. Indices
#: 6 and 7 are missing on purpose: the browser reports the two triggers as
#: buttons, and the engine has them as axes, so they are read from the button
#: *values* below rather than from this table.
_STANDARD_BUTTONS = {
0: "a", 1: "b", 2: "x", 3: "y", 4: "lb", 5: "rb", 8: "back", 9: "start",
10: "l3", 11: "r3", 12: "dpad_up", 13: "dpad_down", 14: "dpad_left", 15: "dpad_right", 16: "guide",
} # fmt: skip
#: The two analog triggers, which the standard layout carries as buttons whose
#: ``value`` runs 0 to 1.
_STANDARD_TRIGGERS = {6: "lt", 7: "rt"}
#: The standard controller's four stick axes, in order. ``+y`` is down the screen
#: in both the W3C layout and the engine's, so no axis is flipped here.
_STANDARD_AXES = ("left_x", "left_y", "right_x", "right_y")
#: The vendor and product a browser prints inside a pad's ``id``. Chrome and
#: Safari write ``(Vendor: 045e Product: 028e)``; Firefox leads with
#: ``045e-028e-``. Neither is required by the specification, which says only
#: that the string identifies the device.
_ID_VENDOR_PRODUCT = (
re.compile(r"vendor:\s*([0-9a-f]{4}).*?product:\s*([0-9a-f]{4})", re.IGNORECASE | re.DOTALL),
re.compile(r"^([0-9a-f]{4})-([0-9a-f]{4})-", re.IGNORECASE),
)
[docs]
def browser_pad_guid(pad_id: str) -> str:
"""The stable device id inside a browser pad's ``id`` string, or "".
The Gamepad API gives a slot index and a descriptive string, and the index
is not identity: a controller unplugged and plugged back in can land in a
different slot, and a game that keyed a rebinding on the slot would hand
player two's controls to player one. The string is what survives, and the
part of it that is an identity rather than a description is the vendor and
product the browser prints into it, which this returns as
``"045e:028e"``.
A browser that prints neither gets ``""``, the same answer a backend with
no concept of a device id gives, and the same meaning: this pad has no
identity to save a rebinding against, so a per-controller setting for it
lasts the session. Inventing one from the slot would be worse than having
none, because it would look stable and would not be.
The spelling is deliberately not SDL's thirty-two hex digits. The browser
cannot supply the bus, version and driver those encode, and a value shaped
like an SDL GUID would claim a desktop rebinding file applies here.
"""
for pattern in _ID_VENDOR_PRODUCT:
found = pattern.search(pad_id)
if found:
return f"{found[1].lower()}:{found[2].lower()}"
return ""
def _route_controllers(router: InputRouter, evt: dict[str, Any]) -> None:
"""One frame's whole pad list, as a poll transaction.
The browser has no pad events worth routing (``gamepadconnected`` fires
once and says nothing about buttons), so the page reads
``navigator.getGamepads()`` in its frame callback and the engine treats
the result exactly as it treats a desktop backend's poll: what is not in
the list has gone.
A pad the browser could not lay out (``mapping`` is not ``"standard"``) is
reported as an unmapped device rather than dropped. Naming its buttons
after a controller's would be a guess, so every standard name reads at rest
and the driver's own indices go through as the raw state a rebind screen
and a raw binding read.
"""
state = router._input
with state._controller_poll() as report:
for pad in evt.get("pads") or []:
pressed = [bool(value) for value in pad.get("pressed") or []]
values = [float(value) for value in pad.get("buttons") or []]
axes = [float(value) for value in pad.get("axes") or []]
standard = str(pad.get("mapping", "")) == "standard"
name = str(pad.get("id", ""))
held = dict.fromkeys(controller.BUTTON_NAMES, False)
raw = dict.fromkeys(controller.AXIS_NAMES, 0.0)
if standard:
for index, button in _STANDARD_BUTTONS.items():
held[button] = index < len(pressed) and pressed[index]
for index, trigger in _STANDARD_TRIGGERS.items():
raw[trigger] = values[index] if index < len(values) else 0.0
for index, axis in enumerate(_STANDARD_AXES):
raw[axis] = axes[index] if index < len(axes) else 0.0
report(
int(pad.get("index", 0)),
controller.normalise_buttons(held),
controller.normalise_axes(raw, stick_scale=1.0, trigger_range=(0.0, 1.0)),
name=name,
guid=browser_pad_guid(name),
recognised=standard,
raw_buttons=dict(enumerate(pressed)),
raw_axes=dict(enumerate(axes)),
)
# A pad is polled rather than delivered, so the changes the poll found
# become events here, the way the desktop adapter does it after its own
# transaction. Without it a pad drives polled state and no handler.
router.flush_joy_events()
[docs]
def route_browser_events(router: InputRouter, events: Iterable[dict[str, Any]]) -> None:
"""Route a batch of browser input events through *router*.
Args:
router: The :class:`InputRouter` bound to the running scene tree.
events: Decoded JSON event objects, in the order the page sent them.
"""
# The browser numbers the mouse in DOM order, middle before right, and a
# pen in the W3C numbering, where the barrel is 2. The indices go through
# as the page reports them and these layouts name them.
router.declare_layout(DOM_MOUSE_LAYOUT)
router.declare_layout(W3C_PEN_LAYOUT)
for evt in _drop_touch_synthesized_mouse(list(events)):
etype = evt.get("type")
when = _time(evt)
if etype == "key":
code = int(evt.get("code", 0))
pressed = bool(evt.get("pressed", False))
# A held key repeats: the browser re-fires keydown, and pages that
# predate the `repeat` flag do not say so. Treating a press of an
# already-held key as a repeat keeps game actions firing once per
# physical press on those pages too.
echo = bool(evt.get("repeat", False)) or (pressed and router.is_key_down(code))
router.key(code, pressed, echo=echo, time=when, **_mods(evt))
elif etype == "char":
router.char(chr(int(evt.get("codepoint", 0))), time=when)
elif etype == "pointer":
_route_pointer(router, evt, when)
elif etype == "gamepads":
_route_controllers(router, evt)
elif etype == "capabilities":
# What the page could ask the browser directly, rather than infer
# from an event that has happened: whether the primary pointer is
# coarse, and whether the page can lock it. Declaring is one-way,
# so a page repeating itself costs nothing.
capabilities.declare(**{name: bool(value) for name, value in evt.items() if name != "type"})
elif etype == "insets":
# Edges of the glass something is covering: the soft keyboard the
# browser slid up, and the notch and home indicator the platform
# reports through the safe-area environment variables. A page that
# can read them at all says so, which is what makes the capability
# true; whether any of them is non-zero right now is a different
# question and not one a capability answers.
tree = router.tree
if tree is not None:
tree.screen_insets = (
float(evt.get("left", 0.0)),
float(evt.get("top", 0.0)),
float(evt.get("right", 0.0)),
float(evt.get("bottom", 0.0)),
)
capabilities.declare(screen_insets=True)
elif etype == "mouse":
button = int(evt.get("button", 0))
if button >= 0:
router.mouse_button(button, bool(evt.get("pressed", False)), time=when, **_mods(evt))
elif etype == "mousemove":
if evt.get("relative"):
router.mouse_motion_relative(float(evt.get("dx", 0.0)), float(evt.get("dy", 0.0)), time=when)
else:
router.mouse_motion(float(evt.get("x", 0.0)), float(evt.get("y", 0.0)), time=when)
elif etype == "scroll":
router.scroll(float(evt.get("dx", 0.0)), float(evt.get("dy", 0.0)), time=when)
elif etype == "focus":
# The loss releases what the page was holding, since the real key
# and button releases go to whatever took the focus; either way the
# router records it, because only a focused page moves focus by hover.
if evt.get("focused", False):
router.focus_gained()
else:
router.focus_lost()
# The three below are the older vocabulary, decoded for bundles
# exported before the page spoke pointer events.
elif etype == "touch":
phase = _WIRE_PHASES.get(int(evt.get("action", 0)))
if phase is not None and phase is not PointerPhase.ENTER and phase is not PointerPhase.LEAVE:
router.touch(
int(evt.get("id", 0)),
phase,
float(evt.get("x", 0.0)),
float(evt.get("y", 0.0)),
_touch_pressure(evt),
time=when,
)
__all__ = ["browser_pad_guid", "route_browser_events"]