merged worktrees, task gkt window and gnome extension

dmapp/dmos/task_window.py

@@ -0,0 +1,167 @@
+#!/usr/bin/env python3
+"""
+Deskmeter Task Window - Regular Mode
+Shows current task in an always-on-top window visible on all workspaces
+"""
+
+import gi
+gi.require_version('Gtk', '4.0')
+from gi.repository import Gtk, GLib
+import urllib.request
+import json
+import sys
+import os
+import subprocess
+
+# Try ports in order: env var, command line arg, or try common ports
+DEFAULT_PORTS = [10001, 10000]  # worktree first, then default
+UPDATE_INTERVAL = 500  # milliseconds - fast updates to catch workspace changes
+WORKSPACE_CHECK_INTERVAL = 200  # milliseconds - check for workspace changes
+
+# Global API URL (will be set after port detection)
+DESKMETER_API_URL = None
+
+
+class TaskWindow(Gtk.ApplicationWindow):
+    def __init__(self, app, api_url):
+        super().__init__(application=app, title="Deskmeter Task")
+
+        self.api_url = api_url
+
+        # Set window properties
+        self.set_default_size(400, 60)
+
+        # Remove window decorations (no title bar, close button, etc.)
+        self.set_decorated(False)
+
+        # Create label for task display
+        self.label = Gtk.Label(label="Loading...")
+        self.label.set_markup('<span font_desc="14">Loading...</span>')
+        self.label.set_margin_top(20)
+        self.label.set_margin_bottom(20)
+        self.label.set_margin_start(20)
+        self.label.set_margin_end(20)
+
+        # Set label to allow selection (useful for copying)
+        self.label.set_selectable(True)
+
+        # Create box container
+        box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL)
+        box.append(self.label)
+
+        self.set_child(box)
+
+        # Track current workspace for change detection
+        self.current_workspace = self._get_current_workspace()
+        self.last_task = None
+
+        # Start periodic updates
+        GLib.timeout_add(UPDATE_INTERVAL, self.update_task)
+
+        # Monitor workspace changes more frequently
+        GLib.timeout_add(WORKSPACE_CHECK_INTERVAL, self.check_workspace_change)
+
+        # Initial update
+        self.update_task()
+
+    def _get_current_workspace(self):
+        """Get current workspace number using wmctrl"""
+        try:
+            result = subprocess.run(['wmctrl', '-d'], capture_output=True, text=True, timeout=1)
+            for line in result.stdout.splitlines():
+                if '*' in line:  # Current workspace marked with *
+                    return line.split()[0]
+        except:
+            pass
+        return None
+
+    def check_workspace_change(self):
+        """Check if workspace changed and trigger immediate update"""
+        new_workspace = self._get_current_workspace()
+        if new_workspace != self.current_workspace and new_workspace is not None:
+            self.current_workspace = new_workspace
+            # Workspace changed, update immediately
+            GLib.timeout_add(2200, self.update_task)  # Wait 2.2s for dmcore to update
+        return True
+
+    def update_task(self):
+        """Fetch current task from API and update label"""
+        try:
+            with urllib.request.urlopen(self.api_url, timeout=2) as response:
+                data = json.loads(response.read().decode('utf-8'))
+                task_path = data.get('task_path', 'no task')
+
+                # Only update label if task changed (reduces flicker)
+                if task_path != self.last_task:
+                    self.last_task = task_path
+                    # Update label with markup for better visibility
+                    self.label.set_markup(f'<span font_desc="14" weight="bold">{task_path}</span>')
+
+        except urllib.error.URLError as e:
+            self.label.set_markup('<span font_desc="14" foreground="red">offline - dmweb not running</span>')
+            self.last_task = None
+
+        except json.JSONDecodeError as e:
+            self.label.set_markup('<span font_desc="14" foreground="orange">error - invalid API response</span>')
+            self.last_task = None
+
+        except Exception as e:
+            self.label.set_markup(f'<span font_desc="14" foreground="red">error: {str(e)}</span>')
+            self.last_task = None
+
+        # Return True to keep the timer running
+        return True
+
+
+class TaskApp(Gtk.Application):
+    def __init__(self, api_url):
+        super().__init__(application_id='local.deskmeter.task-window')
+        self.api_url = api_url
+
+    def do_activate(self):
+        window = TaskWindow(self, self.api_url)
+        window.present()
+
+
+def detect_api_port():
+    """Try to find which port the dmweb API is running on"""
+    # Check environment variable first
+    env_port = os.environ.get('DESKMETER_PORT')
+    if env_port:
+        url = f'http://localhost:{env_port}/api/current_task'
+        print(f"Using port from DESKMETER_PORT env var: {env_port}")
+        return url
+
+    # Try common ports
+    for port in DEFAULT_PORTS:
+        url = f'http://localhost:{port}/api/current_task'
+        try:
+            with urllib.request.urlopen(url, timeout=1) as response:
+                # If we get here, the port is responding
+                print(f"Found dmweb API on port {port}")
+                return url
+        except:
+            continue
+
+    # Fallback to default
+    print(f"Could not detect dmweb API, using default port {DEFAULT_PORTS[-1]}")
+    return f'http://localhost:{DEFAULT_PORTS[-1]}/api/current_task'
+
+
+def main():
+    # Check for port argument
+    if len(sys.argv) > 1 and sys.argv[1].isdigit():
+        port = sys.argv[1]
+        api_url = f'http://localhost:{port}/api/current_task'
+        print(f"Using port from command line: {port}")
+    else:
+        api_url = detect_api_port()
+
+    print(f"API URL: {api_url}")
+
+    app = TaskApp(api_url)
+    return app.run([])
+
+
+if __name__ == '__main__':
+    main()