diff --git a/tests/test_input.py b/tests/test_input.py index 22fbe9f..9d00a53 100644 --- a/tests/test_input.py +++ b/tests/test_input.py @@ -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() diff --git a/ui/server.py b/ui/server.py index 2bff7cc..a2c08f3 100755 --- a/ui/server.py +++ b/ui/server.py @@ -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