Pad: count gamescope's real typing time, with the settle margin added once

Review found the estimate used 12 ms a character where the agent takes 16 ms
(32 shifted), and added the margin on every request so it piled up.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
saphidandClaude Sonnet 5.5 committed 2026-10-01 21:01:53 +10:00
1 parent 3539215517
commit 0373555519
2 files changed
+32 -9

No files matched your search

+14 -1
View File
@@ -776,9 +776,22 @@ class PadDelivery(unittest.TestCase):
start = time.time()
self.s.remote_input({"events": [{"key": "é"}]})
self.settle()
self.assertGreater(time.time() - start, 0.5) # the 50 keys' time, plus a margin
self.assertGreater(time.time() - start, 0.6) # 50 keys at 16 ms, plus the margin
self.assertEqual(self.typed(), [{"key": "é"}])
def test_typing_time_counts_every_key_transition(self):
t = self.s.typing_seconds
self.assertAlmostEqual(t([{"text": "ab"}]), 4 * 0.008)
self.assertAlmostEqual(t([{"text": "A"}]), 4 * 0.008) # shift down, key down, key up, shift up
self.assertAlmostEqual(t([{"text": "x" * 500}]), 8.0) # a long paste
self.assertAlmostEqual(t([{"key": 29, "down": True}, {"dx": 3, "dy": 0}]), 0.008)
def test_the_margin_is_added_once_not_per_request(self):
for _ in range(10):
self.s.remote_input({"events": [{"key": "ab"}]})
# ten requests of 4 transitions each: the work queues up, with no margin added per request
self.assertLess(self.s._typed_until - time.time(), 10 * 4 * 0.008 + 0.05)
def test_trimming_while_sending_doesnt_drop_unsent_events(self):
slow = self.Agent()
gate = threading.Event()
+18 -8
View File
@@ -943,19 +943,29 @@ _kde_queue = []
_kde_lock = threading.Lock()
_kde_worker = [None]
_kde_until = 0.0 # keys typed through KDE Connect should have landed by then
_typed_until = 0.0 # likewise for gamescope's
_typed_until = 0.0 # when gamescope's agent will have typed everything it's been given
KDE_QUEUE_LIMIT = 200
KDE_WAIT = 90 # seconds to wait for KDE Connect before giving up on what's queued
TYPING_SETTLE = 0.6 # seconds past the estimate
TYPING_SETTLE = 0.6 # seconds past the estimate, once
def typing_seconds(touch):
"""How long gamescope's agent takes over these: it sleeps 8 ms after every key transition."""
transitions = 0
for e in touch:
if "text" in e:
transitions += sum(2 + 2 * frame_touch.ASCII[ch][1] for ch in e["text"] if ch in frame_touch.ASCII)
elif "key" in e:
transitions += 1
return transitions * 0.008
def _note_typed(touch):
"""Gamescope types a key every ~8 ms; remember when what was just sent should be done."""
global _typed_until
keys = sum(len(e["text"]) if "text" in e else 1 for e in touch if "text" in e or "key" in e)
if keys:
with _kde_lock:
_typed_until = max(_typed_until, time.time()) + keys * 0.012 + TYPING_SETTLE
seconds = typing_seconds(touch)
if seconds:
with _kde_lock: # the end of the work, which queues up; the margin is added once, where it's checked
_typed_until = max(_typed_until, time.time()) + seconds
def _queue_for_kde(events):
@@ -979,7 +989,7 @@ def _drain_kde():
if not batch:
_kde_worker[0] = None # retired under the lock, so a new request starts another
return
wait = _typed_until - time.time()
wait = _typed_until + TYPING_SETTLE - time.time()
_kde_until = time.time() + 3600 # held while in flight
if wait > 0:
time.sleep(wait) # gamescope's typing, from just before, goes first