Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion src/floww/core/config.py
Original file line number Diff line number Diff line change
Expand Up @@ -46,7 +46,8 @@ def _load_and_merge_config(self) -> Dict[str, Any]:

logger.debug(f"User configuration: {user_config}")

merged_config = self.default_conf
import copy
merged_config = copy.deepcopy(self.default_conf)

if isinstance(user_config.get("timing"), dict):
user_timing = user_config["timing"]
Expand Down
43 changes: 42 additions & 1 deletion src/floww/core/workspace.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,12 +3,16 @@
# But as a backup I am using wmctrl which works on wayland

import logging
import os
import json
import subprocess

try:
from ewmhlib import EwmhRoot

EWMHLIB_AVAILABLE = True
except ImportError:
except Exception:
# ewmhlib might fail to import if X11/Xlib is not available or DISPLAY is not set
EWMHLIB_AVAILABLE = False


Expand All @@ -23,6 +27,14 @@ class WorkspaceManager:

def __init__(self):
self.use_ewmh = EWMHLIB_AVAILABLE
self.is_hyprland = os.environ.get("HYPRLAND_INSTANCE_SIGNATURE") is not None

if self.is_hyprland:
logger.info("Hyprland detected. Using hyprctl for workspace management.")
self.use_ewmh = False
self.wmctrl_cmd = None
self.hyprctl_cmd = "hyprctl"
return

if not self.use_ewmh:
logger.warning(
Expand Down Expand Up @@ -52,6 +64,13 @@ def switch(self, desktop_num: int) -> bool:
Raises:
WorkspaceError: If switching fails using the preferred method.
"""
if self.is_hyprland:
if not self._switch_with_hyprctl(desktop_num):
raise WorkspaceError(
f"Failed to switch to workspace {desktop_num} using hyprctl."
)
return True

if self.use_ewmh:
try:
num_desktops = self.ewmh.getNumberOfDesktops()
Expand All @@ -77,11 +96,33 @@ def switch(self, desktop_num: int) -> bool:

def get_total_workspaces(self):
"""Get the total number of workspaces available"""
if self.is_hyprland:
return self._get_total_workspaces_with_hyprctl()
if self.use_ewmh:
return self.ewmh.getNumberOfDesktops()
else:
return self._get_total_workspaces_with_wmctrl()

def _switch_with_hyprctl(self, desktop_num: int) -> bool:
"""Internal helper to switch using hyprctl command. Returns success bool."""
try:
cmd = [self.hyprctl_cmd, "dispatch", "workspace", str(desktop_num)]
return run_command(cmd)
except Exception:
return False

def _get_total_workspaces_with_hyprctl(self) -> int:
"""Get total workspaces using hyprctl command."""
try:
cmd = [self.hyprctl_cmd, "workspaces", "-j"]
result = subprocess.run(cmd, capture_output=True, text=True, check=True)
workspaces = json.loads(result.stdout)
if not workspaces:
return 1
return max(ws["id"] for ws in workspaces)
except Exception:
return 0

def _switch_with_wmctrl(self, desktop_num: int) -> bool:
"""Internal helper to switch using wmctrl command. Returns success bool."""
try:
Expand Down
80 changes: 80 additions & 0 deletions tests/test_workspace_manager.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,80 @@
import unittest
from unittest.mock import patch, MagicMock
import os
import json
from floww.core.workspace import WorkspaceManager
from floww.core.errors import WorkspaceError


class TestWorkspaceManager(unittest.TestCase):
def setUp(self):
# Clear environment before each test
if "HYPRLAND_INSTANCE_SIGNATURE" in os.environ:
del os.environ["HYPRLAND_INSTANCE_SIGNATURE"]

@patch("floww.core.workspace.EWMHLIB_AVAILABLE", False)
@patch("floww.core.workspace.run_command")
def test_init_no_hyprland_no_ewmh(self, mock_run):
wm = WorkspaceManager()
self.assertFalse(wm.is_hyprland)
self.assertFalse(wm.use_ewmh)
self.assertEqual(wm.wmctrl_cmd, "wmctrl")

@patch("floww.core.workspace.EWMHLIB_AVAILABLE", False)
@patch.dict(os.environ, {"HYPRLAND_INSTANCE_SIGNATURE": "test_sig"})
def test_init_hyprland(self):
wm = WorkspaceManager()
self.assertTrue(wm.is_hyprland)
self.assertFalse(wm.use_ewmh)
self.assertEqual(wm.hyprctl_cmd, "hyprctl")
self.assertIsNone(wm.wmctrl_cmd)

@patch("floww.core.workspace.EWMHLIB_AVAILABLE", False)
@patch.dict(os.environ, {"HYPRLAND_INSTANCE_SIGNATURE": "test_sig"})
@patch("floww.core.workspace.run_command")
def test_switch_hyprland_success(self, mock_run):
mock_run.return_value = True
wm = WorkspaceManager()
result = wm.switch(2)
self.assertTrue(result)
mock_run.assert_called_with(["hyprctl", "dispatch", "workspace", "2"])

@patch("floww.core.workspace.EWMHLIB_AVAILABLE", False)
@patch.dict(os.environ, {"HYPRLAND_INSTANCE_SIGNATURE": "test_sig"})
@patch("floww.core.workspace.run_command")
def test_switch_hyprland_failure(self, mock_run):
mock_run.return_value = False
wm = WorkspaceManager()
with self.assertRaises(WorkspaceError):
wm.switch(2)

@patch("floww.core.workspace.EWMHLIB_AVAILABLE", False)
@patch.dict(os.environ, {"HYPRLAND_INSTANCE_SIGNATURE": "test_sig"})
@patch("subprocess.run")
def test_get_total_workspaces_hyprland(self, mock_sub_run):
mock_response = MagicMock()
mock_response.stdout = json.dumps(
[{"id": 1, "name": "1"}, {"id": 3, "name": "3"}, {"id": 2, "name": "2"}]
)
mock_sub_run.return_value = mock_response

wm = WorkspaceManager()
total = wm.get_total_workspaces()
self.assertEqual(total, 3)
mock_sub_run.assert_called()

@patch("floww.core.workspace.EWMHLIB_AVAILABLE", False)
@patch.dict(os.environ, {"HYPRLAND_INSTANCE_SIGNATURE": "test_sig"})
@patch("subprocess.run")
def test_get_total_workspaces_hyprland_empty(self, mock_sub_run):
mock_response = MagicMock()
mock_response.stdout = json.dumps([])
mock_sub_run.return_value = mock_response

wm = WorkspaceManager()
total = wm.get_total_workspaces()
self.assertEqual(total, 1)


if __name__ == "__main__":
unittest.main()
Loading