"""Native-backed gesture system with composition and arbitration.
Attach gestures to any view-like element via the ``gestures=`` prop:
```python
import pythonnative as pn
from pythonnative import gestures
@pn.component
def Draggable():
tx = pn.use_animated_value(0.0)
ty = pn.use_animated_value(0.0)
def on_end(event):
pn.Animated.spring(tx, to=0.0).start()
pn.Animated.spring(ty, to=0.0).start()
return pn.Animated.View(
pn.Text("Drag me"),
style={"transform": [{"translate_x": tx}, {"translate_y": ty}], "padding": 24},
gestures=[
gestures.Pan(
on_change=pn.Animated.event(translation_x=tx, translation_y=ty),
on_end=on_end,
)
],
)
```
Each gesture descriptor is a frozen dataclass holding numeric
configuration plus user callbacks. The reconciler serializes the
configuration into plain dicts for the native handler (so prop diffing
never compares closures) and routes the callbacks through the
tag-based event channel. Recognition itself is native:
- **iOS** attaches real ``UIGestureRecognizer`` instances
(``PythonNativeKit``).
- **Android** runs an equivalent recognizer set in Kotlin (the
``pythonnative`` Gradle module) on top of ``MotionEvent`` streams.
- **The browser preview** feeds DOM pointer events into the pure-Python
[`GestureArbiter`][pythonnative.gestures.GestureArbiter] below, which
doubles as the executable specification for the native ports.
Composition
-----------
Gestures in a plain ``gestures=[...]`` list all recognize
*simultaneously* (a press ripple plus a pan plus a pinch can all run at
once). To control how gestures interact, wrap them in composition
nodes, which nest arbitrarily:
- [`Simultaneous`][pythonnative.gestures.Simultaneous]: members may all
activate together (the flat-list default, useful inside other nodes).
- [`Race`][pythonnative.gestures.Race]: the first member to activate
wins; the rest fail for the remainder of the interaction.
- [`Exclusive`][pythonnative.gestures.Exclusive]: priority order. A
member may only activate after every member listed *before* it has
failed. ``Exclusive(double_tap, single_tap)`` is the classic
double-tap-wins arrangement: the single tap fires only after the
double-tap window expires.
```python
gestures=[
gestures.Race(
gestures.Pan(on_change=drag),
gestures.LongPress(on_long_press=show_menu),
),
]
```
Every callback receives a [`GestureEvent`][pythonnative.gestures.GestureEvent]
with position, translation, velocity, scale, and rotation populated as
appropriate for the gesture kind.
Every descriptor accepts ``enabled=False`` to keep it in the list (so
sibling indices and composition stay stable) while it neither
recognizes nor takes part in arbitration. [`Pan`][pythonnative.gestures.Pan]
adds React Native Gesture Handler's activation criteria
(``active_offset_x``, ``fail_offset_y``, ``max_pointers``,
``min_velocity``, ...); the rules are spelled out on the class and
implemented identically by the arbiter below and by the native
recognizers.
"""
from __future__ import annotations
import math
from dataclasses import dataclass, field
from enum import Enum
from typing import Any, Callable, Dict, List, Literal, Optional, Sequence, Set, Tuple, Union
__all__ = [
"DEFAULT_PAN_MIN_DISTANCE",
"GestureState",
"GestureEvent",
"GestureSpec",
"SwipeDirection",
"Tap",
"LongPress",
"Pan",
"Swipe",
"Fling",
"Pinch",
"Rotation",
"Simultaneous",
"Race",
"Exclusive",
"GestureGroup",
"GestureArbiter",
"serialize_gestures",
]
class GestureState(str, Enum):
"""Lifecycle states reported on [`GestureEvent.state`][pythonnative.gestures.GestureEvent].
A ``str`` enum, so members compare equal to their wire value
(``GestureState.ENDED == "ended"``) and serialize as plain strings
across the native bridge, while callers get exhaustive
``match`` support and autocomplete.
Attributes:
BEGAN: The gesture activated (first callback).
CHANGED: A continuous gesture updated (pan, pinch, rotation).
ENDED: The gesture completed successfully.
CANCELLED: The gesture was interrupted (lost arbitration, view
unmounted, pointer left the window).
"""
BEGAN = "began"
CHANGED = "changed"
ENDED = "ended"
CANCELLED = "cancelled"
def __str__(self) -> str:
return self.value
GestureCallback = Callable[["GestureEvent"], Any]
SwipeDirection = Literal["left", "right", "up", "down"]
"""A resolved swipe or fling direction; ``None`` on a descriptor means any direction."""
_SWIPE_DIRECTIONS = frozenset({"left", "right", "up", "down"})
Offset = Union[float, Tuple[float, float]]
"""A pan activation or failure bound: one number or a ``(negative_bound, positive_bound)`` pair."""
DEFAULT_PAN_MIN_DISTANCE = 10.0
"""Points a pan must travel before activating when no other activation criterion is given."""
@dataclass(frozen=True)
class GestureEvent:
"""Snapshot delivered to gesture callbacks.
Attributes:
kind: Gesture kind (``"tap"``, ``"long_press"``, ``"pan"``,
``"swipe"``, ``"fling"``, ``"pinch"``, ``"rotation"``).
state: One of [`GestureState`][pythonnative.gestures.GestureState].
x: Pointer x-position in the view's coordinate space (points).
y: Pointer y-position in the view's coordinate space (points).
absolute_x: Pointer x-position in the window's coordinate
space (points).
absolute_y: Pointer y-position in the window's coordinate
space (points).
translation_x: Horizontal displacement since the gesture
activated (pan only).
translation_y: Vertical displacement since the gesture
activated (pan only).
velocity_x: Horizontal pointer velocity in points/second
(pan, swipe, and fling).
velocity_y: Vertical pointer velocity in points/second
(pan, swipe, and fling).
scale: Pinch scale factor relative to activation (pinch only).
rotation: Rotation in radians relative to activation
(rotation only).
pointer_count: Number of pointers currently down.
direction: Resolved swipe/fling direction, or ``None`` for
other gesture kinds.
"""
kind: str
state: GestureState
x: float = 0.0
y: float = 0.0
absolute_x: float = 0.0
absolute_y: float = 0.0
translation_x: float = 0.0
translation_y: float = 0.0
velocity_x: float = 0.0
velocity_y: float = 0.0
scale: float = 1.0
rotation: float = 0.0
pointer_count: int = 1
direction: Optional[SwipeDirection] = None
def __post_init__(self) -> None:
# Payloads from the native bridge carry plain strings; coerce
# so ``event.state`` is always a ``GestureState`` member.
if not isinstance(self.state, GestureState):
object.__setattr__(self, "state", GestureState(self.state))
_EVENT_FIELDS = frozenset(
{
"kind",
"state",
"x",
"y",
"absolute_x",
"absolute_y",
"translation_x",
"translation_y",
"velocity_x",
"velocity_y",
"scale",
"rotation",
"pointer_count",
"direction",
}
)
def event_from_payload(payload: Dict[str, Any]) -> GestureEvent:
"""Build a [`GestureEvent`][pythonnative.gestures.GestureEvent] from a payload dict.
Unknown keys are dropped so platform handlers can attach extra
diagnostics without breaking the public dataclass.
"""
return GestureEvent(**{k: v for k, v in payload.items() if k in _EVENT_FIELDS})
# ======================================================================
# Public gesture descriptors
# ======================================================================
@dataclass(frozen=True)
class _BaseGesture:
"""Shared callback slots and the ``enabled`` switch for every gesture.
Attributes:
on_begin: Called when a continuous gesture activates.
on_change: Called on every update of a continuous gesture.
on_end: Called when the gesture ends or is cancelled.
enabled: When ``False`` the gesture stays in the list (indices
and composition are unchanged) but never recognizes and is
skipped by ``Race`` and ``Exclusive`` arbitration, so a
later ``Exclusive`` member doesn't wait for it.
"""
on_begin: Optional[GestureCallback] = None
on_change: Optional[GestureCallback] = None
on_end: Optional[GestureCallback] = None
enabled: bool = True
kind: str = ""
def _config(self) -> Dict[str, Any]:
return {}
def _to_spec(self) -> Dict[str, Any]:
spec: Dict[str, Any] = {"kind": self.kind, "enabled": bool(self.enabled)}
spec.update(self._config())
return spec
def _dispatch(self, event: GestureEvent) -> None:
if event.state == GestureState.BEGAN:
callback = self.on_begin
elif event.state == GestureState.CHANGED:
callback = self.on_change
else:
callback = self.on_end
if callback is not None:
from .runtime import invoke
invoke(callback, event)
@dataclass(frozen=True)
class Tap(_BaseGesture):
"""Recognize ``n_taps`` quick taps.
Attributes:
on_tap: Called once the tap (or multi-tap) completes.
n_taps: Number of consecutive taps required (``2`` for
double-tap).
max_distance: Maximum pointer travel (points) for a touch to
still count as a tap.
"""
on_tap: Optional[GestureCallback] = None
n_taps: int = 1
max_distance: float = 12.0
kind: str = "tap"
def _config(self) -> Dict[str, Any]:
return {"n_taps": int(self.n_taps), "max_distance": float(self.max_distance)}
def _dispatch(self, event: GestureEvent) -> None:
if event.state == GestureState.ENDED and self.on_tap is not None:
from .runtime import invoke
invoke(self.on_tap, event)
else:
super()._dispatch(event)
@dataclass(frozen=True)
class LongPress(_BaseGesture):
"""Recognize a sustained press.
``on_long_press`` fires as soon as the press has been held for
``min_duration_ms`` (matching ``UILongPressGestureRecognizer``);
``on_end`` fires when the finger lifts.
Attributes:
on_long_press: Called at activation time.
min_duration_ms: Hold duration required to activate.
max_distance: Maximum pointer travel before the press fails.
"""
on_long_press: Optional[GestureCallback] = None
min_duration_ms: float = 500.0
max_distance: float = 12.0
kind: str = "long_press"
def _config(self) -> Dict[str, Any]:
return {
"min_duration_ms": float(self.min_duration_ms),
"max_distance": float(self.max_distance),
}
def _dispatch(self, event: GestureEvent) -> None:
if event.state == GestureState.BEGAN and self.on_long_press is not None:
from .runtime import invoke
invoke(self.on_long_press, event)
else:
super()._dispatch(event)
def _normalize_offset(name: str, value: Any) -> Optional[List[Optional[float]]]:
"""Turn an ``Offset`` argument into the wire form ``[negative_bound, positive_bound]``.
A single number ``n`` is a one-sided threshold: ``n >= 0`` bounds the
positive direction only (``[None, n]``) and ``n < 0`` the negative
direction only (``[n, None]``). A pair is ``(negative_bound,
positive_bound)`` and must satisfy ``negative_bound n``
when ``n >= 0`` and ``dx < n`` when ``n < 0``; a
``(negative_bound, positive_bound)`` pair means ``dx <
negative_bound or dx > positive_bound``, and the pair must satisfy
``negative_bound None:
# Race check: blocked by any non-simultaneous gesture that has
# already activated (or completed) this interaction.
for j in self._indices:
if j == index:
continue
if self._states[j] in (_ACTIVE, _DONE) and not self._is_simultaneous(index, j):
self._set_failed(index, discard_buffer=True)
return
targets = [t for t in self._wait_for.get(index, ()) if t in self._states]
if any(self._states[t] in (_ACTIVE, _DONE) for t in targets):
self._set_failed(index, discard_buffer=True)
return
if any(self._states[t] in (_POSSIBLE, _WAITING) for t in targets):
self._states[index] = _WAITING
self._buffers.setdefault(index, []).append(payload)
return
self._activate(index, [payload])
def _activate(self, index: int, payloads: List[Dict[str, Any]]) -> None:
discrete_done = bool(payloads) and payloads[-1].get("state") in (
GestureState.ENDED,
GestureState.CANCELLED,
)
self._states[index] = _DONE if discrete_done else _ACTIVE
# Winning a race force-fails every unresolved non-simultaneous
# competitor before any output, so their recognizers stand down
# (no stray callbacks later in the interaction).
for j in self._indices:
if j == index:
continue
if self._states[j] in (_POSSIBLE, _WAITING) and not self._is_simultaneous(index, j):
recognizer = self._recognizer_for(j)
if recognizer is not None:
recognizer.force_fail(self._last_t)
self._set_failed(j, discard_buffer=True)
for payload in payloads:
self._deliver(index, payload)
if self._states[index] == _DONE:
self._on_resolved(index, succeeded=True)
def _set_failed(self, index: int, discard_buffer: bool) -> None:
if self._states.get(index) == _ST_FAILED:
return
self._states[index] = _ST_FAILED
if discard_buffer:
self._buffers.pop(index, None)
self._on_resolved(index, succeeded=False)
def _on_resolved(self, index: int, succeeded: bool) -> None:
"""Re-evaluate gestures waiting on ``index`` after it resolves."""
for waiter in self._indices:
if self._states.get(waiter) != _WAITING:
continue
if index not in self._wait_for.get(waiter, ()):
continue
if succeeded:
recognizer = self._recognizer_for(waiter)
if recognizer is not None:
recognizer.force_fail(self._last_t)
self._set_failed(waiter, discard_buffer=True)
continue
targets = [t for t in self._wait_for.get(waiter, ()) if t in self._states]
if any(self._states[t] in (_POSSIBLE, _WAITING) for t in targets):
continue # still waiting on someone else
if any(self._states[t] in (_ACTIVE, _DONE) for t in targets):
recognizer = self._recognizer_for(waiter)
if recognizer is not None:
recognizer.force_fail(self._last_t)
self._set_failed(waiter, discard_buffer=True)
continue
payloads = self._buffers.pop(waiter, [])
self._states[waiter] = _POSSIBLE
self._activate(waiter, payloads)
# Re-exported via ``pythonnative.gestures`` for handler-side construction.
def make_arbiter(specs: Sequence[Dict[str, Any]], emit: EmitFn) -> GestureArbiter:
"""Build a [`GestureArbiter`][pythonnative.gestures.GestureArbiter] from serialized specs."""
return GestureArbiter(specs, emit)