Merge pull request #75 from yusufipk/claude/mouse-cursor-tracking-issue-490b6d

Put the indicator on the screen the session is actually on

dikte/app.py

@@ -164,12 +164,14 @@ class Dikte:
         self._front_watch = None
 
         self.overlay = Overlay(self.conf["overlay_corner"],
-                               screen_name=self.conf["overlay_screen"])
+                               screen_name=self.conf["overlay_screen"],
+                               follow_pointer=self.conf["overlay_follows_pointer"])
         # The agent's indicator sits on top of the dictation one when both are
         # up, and drops into the corner when it is alone there.
         self.ask_overlay = Overlay(self.conf["overlay_corner"], below=self.overlay,
                                    dismissable=True,
-                                   screen_name=self.conf["overlay_screen"])
+                                   screen_name=self.conf["overlay_screen"],
+                                   follow_pointer=self.conf["overlay_follows_pointer"])
         self.recorder = audio.Recorder()
         self.pipeline = Pipeline(self.conf)
         self.ask_pipeline = Pipeline(self.conf)
@@ -1297,10 +1299,10 @@ class Dikte:
             threading.Thread(target=warm, daemon=True).start()
 
     def _apply_settings(self):
-        self.overlay.corner = self.conf["overlay_corner"]
+        for indicator in (self.overlay, self.ask_overlay):
-        self.overlay.screen_name = self.conf["overlay_screen"]
+            indicator.corner = self.conf["overlay_corner"]
-        self.ask_overlay.corner = self.conf["overlay_corner"]
+            indicator.screen_name = self.conf["overlay_screen"]
-        self.ask_overlay.screen_name = self.conf["overlay_screen"]
+            indicator.follow_pointer = self.conf["overlay_follows_pointer"]
         self._apply_local()
         self._build_tray()
         self._refresh_tray()

dikte/config.py

@@ -470,6 +470,9 @@ DEFAULTS = {
     "evdev_hotkey": False,
     "overlay_corner": "bottom-left",
     "overlay_screen": "",
+    # Off, so that an indicator stays where it appeared unless it is asked to
+    # keep up with the pointer. Nothing to say when a screen is named above.
+    "overlay_follows_pointer": False,
     "keep_audio": False,
     "history_limit": 200,
     # A look at the releases page once a day, and nothing more than a look:

dikte/i18n.py

@@ -189,8 +189,10 @@ TR = {
     "Restore the previous clipboard after pasting":
         "Yapıştırdıktan sonra eski pano içeriğini geri koy",
     "Indicator screen": "Gösterge ekranı",
-    "Follow the mouse pointer": "Fare imlecini takip et",
+    "Follow the active screen": "Etkin ekranı takip et",
     "{name} (not connected)": "{name} (bağlı değil)",
+    "Move it when the active screen changes":
+        "Etkin ekran değiştiğinde göstergeyi de taşı",
     "Indicator corner": "Gösterge köşesi",
     "bottom-left": "sol-alt",
     "bottom-right": "sağ-alt",

dikte/overlay.py

@@ -1,6 +1,7 @@
 """The small recording indicator that appears in a screen corner without taking focus."""
 
 import math
+import os
 import sys
 
 from PyQt6.QtCore import Qt, QTimer, QRectF, QPointF
@@ -15,6 +16,7 @@ MIN_WIDTH = 210
 MAX_WIDTH = 460
 MARGIN = 28
 GAP = 10        # between two indicators sharing a corner
+FOLLOW_EVERY = 8   # ticks between two looks for the pointer: about four a second
 
 BG = QColor(22, 24, 29, 238)
 BORDER = QColor(255, 255, 255, 28)
@@ -37,16 +39,68 @@ STATE_COLORS = {"recording": REC, "asking": ASK, "meeting": REC, "busy": BUSY,
 LIVE = ("recording", "asking", "meeting")
 
 
+# KWin's interface, kept once one has been built. See _compositor_screen.
+_kwin = None
+
+
+def _compositor_screen():
+    """The screen KWin says the session is on, or None where nothing says.
+
+    Wayland tells a client where the pointer is only while it is over one of
+    that client's own windows, and the indicator is never under the pointer, so
+    QCursor.pos() answers with a stale point or, when the pointer has never
+    been over a window of ours, with the origin. Either way the indicator lands
+    in the corner of whichever screen holds 0,0 instead of the one being worked
+    on, and on a two-monitor desk that is the wrong screen most of the time.
+    KWin does know, and it names outputs the way Qt names screens, by
+    connector, natively and through XWayland alike. No other Wayland desktop
+    answers this, so the rest are left with the pointer, which is right on X11
+    and wrong on Wayland exactly as before.
+
+    What it answers with is the active output, which is the one under the
+    pointer only where Plasma is set to let the active screen follow the mouse.
+    Under the default, click to focus, it is the focused window's screen, so
+    the indicator lands where the typing is going rather than where the mouse
+    was left. Which is why nothing here, and nothing in the settings window,
+    promises the pointer.
+    """
+    global _kwin
+    if _kwin is None or not _kwin.isValid():
+        # Which also leaves macOS and Windows out, where nothing sets it and
+        # the pointer can be asked where it is like anywhere else.
+        desktop = os.environ.get("XDG_CURRENT_DESKTOP", "").lower()
+        if "kde" not in desktop and "plasma" not in desktop:
+            return None
+        try:
+            from PyQt6.QtDBus import QDBusConnection, QDBusInterface
+            _kwin = QDBusInterface("org.kde.KWin", "/KWin", "org.kde.KWin",
+                                   QDBusConnection.sessionBus())
+        except Exception:
+            return None
+        if not _kwin.isValid():
+            return None
+        # A compositor busy enough not to answer in a fifth of a second is one
+        # the indicator should stop waiting for, not one it should freeze with.
+        _kwin.setTimeout(200)
+    answer = _kwin.call("activeOutputName").arguments()
+    name = answer[0] if answer else ""
+    return next((item for item in QApplication.screens() if item.name() == name),
+                None)
+
+
 class Overlay(QWidget):
     """One indicator. Give it `below` and it stacks on top of that one instead
     of covering it, which is what lets a dictation and a command to the agent be
     under way at the same time and still both be visible."""
 
     def __init__(self, corner="bottom-left", below=None, dismissable=False,
-                 screen_name=""):
+                 screen_name="", follow_pointer=False):
         super().__init__(None)
         self.corner = corner
         self.screen_name = screen_name
+        # Whether it goes on following the pointer once it is up, rather than
+        # settling on the screen it appeared on.
+        self.follow_pointer = follow_pointer
         self.below = below
         # A job that can run for ten minutes should not have to be watched for
         # ten minutes. Clicking such an indicator puts the progress away; the
@@ -65,6 +119,8 @@ class Overlay(QWidget):
         self.seconds = 0.0
         self._phase = 0.0
         self._concealed = True
+        self._shown_on = ""   # the screen it was last put on, by name
+        self._looks = 0       # ticks since the pointer was last looked for
 
         flags = (
             Qt.WindowType.FramelessWindowHint
@@ -252,16 +308,52 @@ class Overlay(QWidget):
                         min(MAX_WIDTH, metrics.horizontalAdvance(self.message) + extra))
         self.resize(width, HEIGHT)
 
-    def _reposition(self):
+    def _screen(self):
-        # The screen the settings name, or, when none is named or it is not
+        """The screen this indicator belongs on right now.
-        # plugged in right now, where the user actually is. Names are connector
+
-        # names on X11 and model names on macOS, where two identical monitors
+        The one the settings name, or, when none is named or it is not plugged
-        # can share one; the first then wins.
+        in right now, where the user actually is. Names are connector names on
-        screen = next(
+        X11 and model names on macOS, where two identical monitors can share
+        one; the first then wins.
+
+        One stacking on another belongs on that one's screen and nowhere else.
+        Asked for itself it would answer where the user is now, which is not
+        where the ribbon it stacks on was put a minute ago, and the pair would
+        end up a monitor apart with this one raised over nothing.
+        """
+        if self.below is not None and self.below.showing:
+            under = next((item for item in QApplication.screens()
+                          if item.name() == self.below._shown_on), None)
+            if under is not None:
+                return under
+        named = next(
             (item for item in QApplication.screens() if item.name() == self.screen_name),
             None,
         )
-        screen = screen or QApplication.screenAt(QCursor.pos()) or QApplication.primaryScreen()
+        return (named or _compositor_screen()
+                or QApplication.screenAt(QCursor.pos())
+                or QApplication.primaryScreen())
+
+    def _wandered_off(self):
+        """Whether the pointer has left the screen the indicator is on.
+
+        Only asked while it is following, and only every few ticks: the answer
+        costs a word with the compositor, and a hand moving a mouse across a
+        desk is slow next to a 33 ms ribbon. Every tick for one that stacks on
+        another, where the answer is free and waiting a third of a second for
+        it would leave the pair split over two monitors for that long.
+        """
+        if not self.follow_pointer or self.screen_name:
+            return False
+        if self.below is None or not self.below.showing:
+            self._looks = (self._looks + 1) % FOLLOW_EVERY
+            if self._looks:
+                return False
+        return self._screen().name() != self._shown_on
+
+    def _reposition(self):
+        screen = self._screen()
+        self._shown_on = screen.name()
         area = screen.availableGeometry()
         left = "left" in self.corner
         top = "top" in self.corner
@@ -277,8 +369,10 @@ class Overlay(QWidget):
     def _tick(self):
         self._phase += 0.12
         # The one underneath can come and go while this one is up; drop back to
-        # the corner when it does rather than leaving a gap where it was.
+        # the corner when it does rather than leaving a gap where it was. And
-        if self.below is not None and self.below.showing != self._stacked:
+        # the screen under the pointer can change while it is up too.
+        moved = self.below is not None and self.below.showing != self._stacked
+        if moved or self._wandered_off():
             self._reposition()
         if self.state in LIVE and not self.paused:
             # keep the ribbon moving even through a pause in speech

dikte/settings_ui.py

@@ -867,7 +867,11 @@ class SettingsWindow(QDialog):
         form = QFormLayout(page)
 
         self.indicator_screen = QComboBox()
-        self.indicator_screen.addItem(t("Follow the mouse pointer"), "")
+        # The active screen rather than the pointer, for the reason in
+        # overlay._compositor_screen: it is what a compositor will answer for,
+        # and on Plasma the two are one screen only where the active screen is
+        # set to follow the mouse.
+        self.indicator_screen.addItem(t("Follow the active screen"), "")
         for screen in QGuiApplication.screens():
             # The native resolution, so that a scaled 4K screen reads
             # 3840 × 2160 and not the 1920 × 1080 Qt sees through the scale.
@@ -881,12 +885,26 @@ class SettingsWindow(QDialog):
             )
         form.addRow(t("Indicator screen"), self.indicator_screen)
 
+        # Only the screen it appeared on is decided when it appears; this is
+        # what makes it keep up with a session that moves to another one
+        # mid-recording. The active screen and not the pointer, because that is
+        # what a compositor will answer for: on Plasma the two are the same
+        # screen only where the active screen is set to follow the mouse, and
+        # otherwise it is the focused window that decides. Nothing to offer
+        # when a screen is named above, since that name is the whole answer.
+        self.follow_pointer = QCheckBox(t("Move it when the active screen changes"))
+        self.indicator_screen.currentIndexChanged.connect(self._sync_follow_pointer)
+        form.addRow("", self.follow_pointer)
+
         self.corner = QComboBox()
         for value in CORNERS:
             self.corner.addItem(t(value), value)
         form.addRow(t("Indicator corner"), self.corner)
         return page
 
+    def _sync_follow_pointer(self):
+        self.follow_pointer.setEnabled(not self.indicator_screen.currentData())
+
     def _api_tab(self):
         page = QWidget()
         outer = QVBoxLayout(page)
@@ -1805,6 +1823,8 @@ class SettingsWindow(QDialog):
         if screen_name and self.indicator_screen.findData(screen_name) < 0:
             self.indicator_screen.addItem(t("{name} (not connected)", name=screen_name), screen_name)
         self._select_data(self.indicator_screen, screen_name)
+        self.follow_pointer.setChecked(conf["overlay_follows_pointer"])
+        self._sync_follow_pointer()
         self._select_data(self.corner, conf["overlay_corner"])
         self.max_seconds.setValue(conf["max_seconds"])
         self.skip_silent.setChecked(conf["skip_silent"])
@@ -1925,6 +1945,9 @@ class SettingsWindow(QDialog):
         conf["paste_shortcut"] = self.paste_shortcut.currentText().strip()
         conf["restore_clipboard"] = self.restore_clipboard.isChecked()
         conf["overlay_screen"] = self.indicator_screen.currentData() or ""
+        # Read even while it is greyed out, so that naming a screen and taking
+        # the name back again does not clear a preference nobody touched.
+        conf["overlay_follows_pointer"] = self.follow_pointer.isChecked()
         conf["overlay_corner"] = self.corner.currentData() or "bottom-left"
         conf["max_seconds"] = self.max_seconds.value()
         conf["skip_silent"] = self.skip_silent.isChecked()

tests/test_ui.py

@@ -52,6 +52,7 @@ CHANGED = {
     "restore_clipboard": True,
     "overlay_corner": "top-right",
     "overlay_screen": "DP-1",
+    "overlay_follows_pointer": True,
     "max_seconds": 120,
     "skip_silent": False,
     "silence_db": -42.0,
@@ -1069,6 +1070,121 @@ class Overlay(DikteTest):
         screen_at.assert_not_called()
         self.assertEqual(widget.pos(), QPoint(1948, 995))
 
+    def _screen(self, name, area):
+        screen = mock.Mock()
+        screen.name.return_value = name
+        screen.availableGeometry.return_value = area
+        return screen
+
+    def _kwin(self, *answer):
+        kwin = mock.Mock()
+        kwin.isValid.return_value = True
+        kwin.call.return_value.arguments.return_value = list(answer)
+        return kwin
+
+    def test_the_compositor_says_which_screen_the_pointer_is_on(self):
+        """Wayland tells a client where the pointer is only while it is over one
+        of that client's own windows, so QCursor.pos() comes back at the origin
+        and every indicator lands on whichever screen holds it. KWin knows."""
+        screens = [self._screen("DP-1", settings_ui.QRect(0, 0, 1920, 1080)),
+                   self._screen("DP-2", settings_ui.QRect(1920, 0, 1920, 1080))]
+        widget = self.overlay()
+        with mock.patch.object(overlay_module, "_kwin", self._kwin("DP-2")), \
+                mock.patch.object(QApplication, "screens", return_value=screens), \
+                mock.patch.object(QApplication, "screenAt") as screen_at:
+            widget._reposition()
+        screen_at.assert_not_called()
+        self.assertEqual(widget.pos(), QPoint(1948, 995))
+
+    def test_the_pointer_decides_when_the_compositor_will_not_say(self):
+        """Every desktop but Plasma, and Plasma while KWin is being replaced."""
+        screens = [self._screen("DP-1", settings_ui.QRect(0, 0, 1920, 1080))]
+        widget = self.overlay()
+        with mock.patch.object(overlay_module, "_kwin", self._kwin()), \
+                mock.patch.object(QApplication, "screens", return_value=screens), \
+                mock.patch.object(QApplication, "screenAt",
+                                  return_value=screens[0]) as screen_at:
+            widget._reposition()
+        screen_at.assert_called()
+        self.assertEqual(widget.pos(), QPoint(28, 995))
+
+    def _two_screens(self):
+        return [self._screen("DP-1", settings_ui.QRect(0, 0, 1920, 1080)),
+                self._screen("DP-2", settings_ui.QRect(1920, 0, 1920, 1080))]
+
+    def _ticks_on(self, widget, screens, kwin):
+        """Run the ribbon long enough for one look at where the pointer is."""
+        with mock.patch.object(overlay_module, "_kwin", kwin), \
+                mock.patch.object(QApplication, "screens", return_value=screens), \
+                mock.patch.object(QApplication, "screenAt", return_value=screens[0]):
+            for _ in range(overlay_module.FOLLOW_EVERY):
+                widget._tick()
+
+    def test_it_can_be_told_to_keep_up_with_the_pointer(self):
+        """The screen it started on is not always the screen you end up on."""
+        screens = self._two_screens()
+        kwin = self._kwin("DP-2")
+        widget = self.overlay(follow_pointer=True)
+        with mock.patch.object(overlay_module, "_kwin", kwin), \
+                mock.patch.object(QApplication, "screens", return_value=screens):
+            widget.show_recording()
+        self.assertEqual(widget.pos(), QPoint(1948, 995))
+        kwin.call.return_value.arguments.return_value = ["DP-1"]
+        self._ticks_on(widget, screens, kwin)
+        self.assertEqual(widget.pos(), QPoint(28, 995))
+
+    def test_it_stays_where_it_appeared_unless_it_was_told_otherwise(self):
+        """Left off, because an indicator that jumps desks mid-sentence is one
+        more thing moving while you are trying to talk."""
+        screens = self._two_screens()
+        kwin = self._kwin("DP-2")
+        widget = self.overlay()
+        with mock.patch.object(overlay_module, "_kwin", kwin), \
+                mock.patch.object(QApplication, "screens", return_value=screens):
+            widget.show_recording()
+        kwin.call.return_value.arguments.return_value = ["DP-1"]
+        self._ticks_on(widget, screens, kwin)
+        self.assertEqual(widget.pos(), QPoint(1948, 995))
+
+    def test_a_named_screen_is_never_left_for_the_pointer(self):
+        """Naming one is the whole answer; following it would undo the naming."""
+        screens = self._two_screens()
+        kwin = self._kwin("DP-2")
+        widget = self.overlay(screen_name="DP-1", follow_pointer=True)
+        with mock.patch.object(QApplication, "screens", return_value=screens):
+            widget.show_recording()
+        self._ticks_on(widget, screens, kwin)
+        kwin.call.assert_not_called()
+        self.assertEqual(widget.pos(), QPoint(28, 995))
+
+    def test_the_one_on_top_goes_where_the_one_underneath_is(self):
+        """Asking for itself would put the pair on two monitors, with this one
+        raised over a ribbon that is not underneath it."""
+        screens = self._two_screens()
+        kwin = self._kwin("DP-2")
+        first = self.overlay()
+        with mock.patch.object(overlay_module, "_kwin", kwin), \
+                mock.patch.object(QApplication, "screens", return_value=screens):
+            first.show_recording()
+            kwin.call.return_value.arguments.return_value = ["DP-1"]
+            second = self.overlay(below=first)
+            second.show_busy("Asking Claude…")
+        self.assertEqual(first.pos(), QPoint(1948, 995))
+        self.assertEqual(second.pos(), QPoint(1948, 929))
+
+    def test_the_compositor_is_asked_only_now_and_then(self):
+        """Every tick would be thirty conversations a second about a hand
+        moving a mouse."""
+        screens = self._two_screens()
+        kwin = self._kwin("DP-2")
+        widget = self.overlay(follow_pointer=True)
+        with mock.patch.object(overlay_module, "_kwin", kwin), \
+                mock.patch.object(QApplication, "screens", return_value=screens):
+            widget.show_recording()
+        kwin.call.reset_mock()
+        self._ticks_on(widget, screens, kwin)
+        self.assertEqual(kwin.call.call_count, 1)
+
     def test_a_warning_and_an_error_both_show(self):
         widget = self.overlay()
         widget.show_warning("cleanup failed")