From f409d8ba307733612623b95bafa6a18c8d48671c Mon Sep 17 00:00:00 2001 From: rozodru Date: Sun, 20 Jul 2025 13:55:00 -0400 Subject: [PATCH] initial commit --- README.md | 94 ++++++++++++++ waybar-nedm-workspaces.py | 265 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 359 insertions(+) create mode 100644 README.md create mode 100755 waybar-nedm-workspaces.py diff --git a/README.md b/README.md new file mode 100644 index 0000000..8aa8496 --- /dev/null +++ b/README.md @@ -0,0 +1,94 @@ +# NEDM Waybar Integration + +A Python script that displays NEDM workspace information and focused window details in Waybar. + +## What it does + +- Shows current workspace number with a bullet indicator (●) +- Displays the name of the currently focused application (e.g., "foot", "firefox") +- Provides tooltip with detailed workspace and window information +- Automatically reconnects if NEDM restarts +- Updates in real-time as you switch workspaces or focus different windows + +## Requirements + +- NEDM (cagebreak) window manager with IPC enabled +- Python 3 +- Waybar + +## Installation + +1. Place the script in your scripts directory: + + ```bash + cp waybar-nedm-workspaces.py ~/scripts/ + chmod +x ~/scripts/waybar-nedm-workspaces.py + ``` + +2. Add the custom module to your Waybar configuration. + +## Waybar Configuration + +### Add to waybar config.json + +Add `"custom/nedm-workspaces"` to your modules list: + +```json +{ + "modules-left": ["custom/nedm-workspaces"], + "custom/nedm-workspaces": { + "exec": "python3 ~/scripts/waybar-nedm-workspaces.py", + "return-type": "json", + "restart-interval": 1, + "tooltip": true + } +} +``` + +### Add styling to style.css + +```css +#custom-nedm-workspaces { + background-color: #2e3440; + color: #d8dee9; + padding: 0 10px; + margin: 0 5px; + border-radius: 5px; +} + +#custom-nedm-workspaces.disconnected { + background-color: #bf616a; + color: #ffffff; +} + +#custom-nedm-workspaces.error { + background-color: #d08770; + color: #ffffff; +} +``` + +## Output Format + +The script displays information in the format: `WS: ●2 | foot` + +- `WS:` - Workspace indicator +- `●2` - Current workspace number with bullet +- `foot` - Name of the focused application + +## Security Note + +This script uses process names from PIDs instead of window titles for security. It does NOT require NEDM to be run with the `-b` flag, which would expose window titles but compromise security. + +## Troubleshooting + +If the script shows "NEDM: Disconnected": + +1. Ensure NEDM is running with IPC enabled +2. Check that the IPC socket exists in `/run/user/$(id -u)/` +3. Verify NEDM was started without disabling IPC + +If it shows generic "View X" instead of process names: + +- The process may have terminated +- The script will fall back to view IDs when process information is unavailable + diff --git a/waybar-nedm-workspaces.py b/waybar-nedm-workspaces.py new file mode 100755 index 0000000..be9c060 --- /dev/null +++ b/waybar-nedm-workspaces.py @@ -0,0 +1,265 @@ +#!/usr/bin/env python3 + +import os +import json +import sys +import time +import signal +import socket +import threading +import subprocess + +class NEDMWorkspaceTracker: + def __init__(self): + self.current_workspace = 1 # Default workspace (1-based) + self.current_window_title = "" # Current window title + self.socket_path = None + self.connected = False + + def find_socket(self): + """Find NEDM IPC socket""" + # Check environment variable first + socket_path = os.getenv('NEDM_SOCKET') or os.getenv('CAGEBREAK_SOCKET') + if socket_path and os.path.exists(socket_path): + return socket_path + + # Search in XDG_RUNTIME_DIR or /tmp for cagebreak-ipc socket + runtime_dir = os.getenv('XDG_RUNTIME_DIR', '/tmp') + uid = os.getuid() + + # Look for socket files matching pattern + try: + for file in os.listdir(runtime_dir): + if file.startswith(f'cagebreak-ipc.{uid}.') and file.endswith('.sock'): + socket_path = os.path.join(runtime_dir, file) + if os.path.exists(socket_path): + return socket_path + except OSError: + pass + + return None + + def connect_to_nedm(self): + """Connect to NEDM IPC socket""" + self.socket_path = self.find_socket() + if not self.socket_path: + self.connected = False + return False + + try: + self.sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + self.sock.connect(self.socket_path) + self.connected = True + # Query current state after connecting + self.query_current_state() + return True + except (socket.error, OSError): + self.connected = False + return False + + def query_current_state(self): + """Query NEDM for current workspace and window state""" + if not self.connected: + return + + try: + # Send dump command to get current state + self.sock.send(b'dump\n') + # Give it a moment to respond + self.sock.settimeout(1.0) + response = self.sock.recv(8192) # Increased buffer for dump response + self.sock.settimeout(None) + + if response: + response_str = response.decode('utf-8', errors='ignore') + # Parse the response to extract current workspace and window info + try: + # Response format should be: "cg-ipc"NULL + if response_str.startswith('cg-ipc'): + json_part = response_str[6:].split('\0')[0] + if json_part: + dump_data = json.loads(json_part) + self.parse_dump_data(dump_data) + except json.JSONDecodeError: + pass + except (socket.error, OSError, socket.timeout): + pass + + def parse_dump_data(self, dump_data): + """Parse dump data to extract current workspace and focused window""" + try: + # Get current output name and current workspace from dump data + curr_output = dump_data.get('curr_output', '') + views_curr_id = dump_data.get('views_curr_id') + tiles_curr_id = dump_data.get('tiles_curr_id') + + # Find the output and get workspace info + outputs = dump_data.get('outputs', {}) + if curr_output in outputs: + output_data = outputs[curr_output] + self.current_workspace = output_data.get('curr_workspace', 0) # Already 1-based in dump output + + # Find the focused view title by matching view_id and tile_id + workspaces = output_data.get('workspaces', []) + curr_ws_idx = output_data.get('curr_workspace', 0) - 1 # Convert to 0-based for array access + if workspaces and len(workspaces) > curr_ws_idx and curr_ws_idx >= 0: + workspace = workspaces[curr_ws_idx] + + # Find the view in the focused tile + tiles = workspace.get('tiles', []) + for tile in tiles: + if tile.get('id') == tiles_curr_id: + view_id = tile.get('view_id') # NEDM uses 'view_id' not 'view' + # Now find the view with this ID + views = workspace.get('views', []) + for view in views: + if view.get('id') == view_id: + # Get window title if available (requires NEDM -b flag for security) + title = view.get('title', '') + if title: + self.current_window_title = title + else: + # Try to get process name from PID as fallback + pid = view.get('pid') + if pid: + try: + import subprocess + proc_name = subprocess.check_output(['ps', '-p', str(pid), '-o', 'comm='], + stderr=subprocess.DEVNULL).decode().strip() + if proc_name: + self.current_window_title = proc_name + else: + self.current_window_title = f"View {view_id}" + except (subprocess.CalledProcessError, FileNotFoundError): + self.current_window_title = f"View {view_id}" + else: + self.current_window_title = f"View {view_id}" + break + break + + except (KeyError, TypeError): + pass + + def listen_for_events(self): + """Listen for workspace events from NEDM""" + if not self.connected: + return + + try: + while self.connected: + data = self.sock.recv(4096) + if not data: + break + + # Parse NEDM IPC protocol: "cg-ipc"NULL + # Handle multiple messages concatenated together + data_str = data.decode('utf-8', errors='ignore') + + # Split by null terminator to handle multiple messages + messages = data_str.split('\0') + + for message in messages: + if message.startswith('cg-ipc'): + json_part = message[6:] # Remove "cg-ipc" prefix + if json_part: # Only process non-empty JSON parts + try: + event = json.loads(json_part) + self.handle_event(event) + except json.JSONDecodeError: + pass + except (socket.error, OSError): + self.connected = False + + def handle_event(self, event): + """Handle workspace-related events""" + event_name = event.get('event_name') + + if event_name == 'switch_ws': + self.current_workspace = event.get('new_workspace', 1) # Already 1-based + # Clear window title when switching workspaces since we don't know the focused window + self.current_window_title = "" + elif event_name == 'focus_tile': + workspace = event.get('new_workspace') # Use new_workspace for focus_tile events + if workspace is not None: + self.current_workspace = workspace # focus_tile events already use 1-based indexing + # Get window title from focus_tile event + view_title = event.get('view_title', '') + self.current_window_title = view_title + elif event_name in ['view_map', 'view_unmap']: + # When windows are mapped/unmapped, query current state to update info + self.query_current_state() + elif event_name == 'close': + # When a window is closed, clear the title + self.current_window_title = "" + elif event_name == 'cycle_views': + # When cycling views, this might change the focused window + self.query_current_state() + elif event_name == 'dump': + # Handle dump responses + self.parse_dump_data(event) + + def get_workspace_info(self): + """Get current workspace information""" + if not self.connected: + return {"text": "NEDM: Disconnected", "class": "disconnected"} + + # Format the display text + if self.current_window_title: + # Truncate long titles to keep the bar readable + title = self.current_window_title[:30] + "..." if len(self.current_window_title) > 30 else self.current_window_title + display_text = f"WS: ●{self.current_workspace} | {title}" + tooltip_text = f"Workspace {self.current_workspace}: {self.current_window_title}" + else: + display_text = f"WS: ●{self.current_workspace}" + tooltip_text = f"Current workspace: {self.current_workspace}" + + return { + "text": display_text, + "class": "workspaces", + "tooltip": tooltip_text + } + +# Global tracker instance +tracker = NEDMWorkspaceTracker() + +def signal_handler(signum, frame): + """Handle termination signals gracefully""" + if tracker.connected: + try: + tracker.sock.close() + except: + pass + sys.exit(0) + +def main(): + """Main loop for waybar module""" + signal.signal(signal.SIGTERM, signal_handler) + signal.signal(signal.SIGINT, signal_handler) + + # Try to connect to NEDM + if tracker.connect_to_nedm(): + # Start listening thread for events + event_thread = threading.Thread(target=tracker.listen_for_events, daemon=True) + event_thread.start() + + while True: + try: + # Try to reconnect if disconnected + if not tracker.connected: + if tracker.connect_to_nedm(): + event_thread = threading.Thread(target=tracker.listen_for_events, daemon=True) + event_thread.start() + + workspace_info = tracker.get_workspace_info() + print(json.dumps(workspace_info), flush=True) + time.sleep(1) # Update every second + except KeyboardInterrupt: + break + except Exception: + # If something goes wrong, show error and continue + error_info = {"text": "NEDM: Error", "class": "error"} + print(json.dumps(error_info), flush=True) + time.sleep(5) # Wait longer on error + +if __name__ == "__main__": + main() \ No newline at end of file