153 lines
6.0 KiB
Python
153 lines
6.0 KiB
Python
from sdbus import (
|
|
DbusInterfaceCommonAsync,
|
|
dbus_method_async,
|
|
dbus_signal_async,
|
|
)
|
|
from .workspace_tree import WorkspaceTree
|
|
from i3ipc.aio import Connection
|
|
import json
|
|
|
|
|
|
class ContextMngrInterface(
|
|
DbusInterfaceCommonAsync,
|
|
interface_name="dev.ezri.sway.ContextManager",
|
|
):
|
|
|
|
workspace_tree: WorkspaceTree = None
|
|
connection: Connection = None
|
|
|
|
def __init__(
|
|
self, workspace_tree: WorkspaceTree, connection: Connection, *args, **kwargs
|
|
):
|
|
super().__init__(*args, **kwargs)
|
|
self.workspace_tree = workspace_tree
|
|
self.connection = connection
|
|
|
|
@dbus_method_async(input_signature="", result_signature="as")
|
|
async def get_available_contexts(self) -> list[str]:
|
|
"""
|
|
Request a list of contexts compatible with the current monitor configuration, sorted by score.
|
|
"""
|
|
scores = await self.workspace_tree.score_contexts(self.connection)
|
|
contexts = [score[0].name for score in scores if score[1] > 0]
|
|
return contexts
|
|
|
|
@dbus_method_async(input_signature="s", result_signature="s")
|
|
async def request_context(self, context: str) -> str:
|
|
"""Request a context switch. This will fail if the current monitor configuration is not compatible with the requested context."""
|
|
print("requesting context", context)
|
|
try:
|
|
await self.workspace_tree.activate_context(self.connection, context)
|
|
return "OK"
|
|
except Exception as e:
|
|
return str(e)
|
|
|
|
@dbus_method_async(input_signature="", result_signature="s")
|
|
async def get_current_context(self) -> str:
|
|
"""Get the current context."""
|
|
return self.workspace_tree.current_context.name
|
|
|
|
@dbus_method_async(input_signature="y", result_signature="b")
|
|
async def focus_workspace(self, user_index: int) -> bool:
|
|
"""Focus a workspace by its 1-indexed ID in its group."""
|
|
print("focusing workspace", user_index)
|
|
workspace, group = self.workspace_tree.get_workspace(user_index)
|
|
if workspace is None or group is None:
|
|
return False
|
|
await workspace.relocate(
|
|
self.connection, await group.get_output_name(self.connection)
|
|
)
|
|
return True
|
|
|
|
@dbus_method_async(input_signature="y", result_signature="b")
|
|
async def move_container(self, user_index: int) -> bool:
|
|
"""Move the focused container to a workspace by its 1-indexed ID in its group."""
|
|
try:
|
|
print("Moving focused container to workspace", user_index, flush=True)
|
|
workspace, _ = self.workspace_tree.get_workspace(user_index)
|
|
await workspace.move_container(self.connection)
|
|
return True
|
|
except Exception as e:
|
|
print(e, flush=True)
|
|
return False
|
|
|
|
@dbus_method_async(input_signature="", result_signature="b")
|
|
async def launch_default(self) -> bool:
|
|
"""Launch the default application for the current workspace."""
|
|
print("launching default")
|
|
await self.workspace_tree.current_context.active_workspace.launch_default(
|
|
self.connection
|
|
)
|
|
return True
|
|
|
|
@dbus_method_async(input_signature="y", result_signature="s")
|
|
async def get_workspace(self, user_index: int) -> str:
|
|
"""Get a workspace by its 1-indexed ID in its group."""
|
|
workspace = self.workspace_tree.get_workspace(user_index)
|
|
if workspace is None:
|
|
return ""
|
|
return json.dumps(workspace.__json__())
|
|
|
|
@dbus_method_async(input_signature="", result_signature="")
|
|
async def focus_next_workspace(self) -> None:
|
|
"""Switch to the next workspace on the focused output."""
|
|
try:
|
|
await self.workspace_tree.current_context.active_group.get_relative_workspace(1).focus(self.connection)
|
|
except Exception as e:
|
|
print(e, flush=True)
|
|
|
|
@dbus_method_async(input_signature="", result_signature="")
|
|
async def focus_previous_workspace(self) -> None:
|
|
"""Switch to the previous workspace on the focused output."""
|
|
try:
|
|
await self.workspace_tree.current_context.active_group.get_relative_workspace(-1).focus(self.connection)
|
|
except Exception as e:
|
|
print(e, flush=True)
|
|
|
|
async def __get_workspace_data(self) -> dict:
|
|
"""Get the workspace tree."""
|
|
try:
|
|
active = self.workspace_tree.current_context.active_workspace
|
|
return {
|
|
"ws": self.workspace_tree.__json__(),
|
|
"current": active is not None and active.__json__() or None,
|
|
"context": self.workspace_tree.current_context.name,
|
|
"visible": {
|
|
name: ws is not None
|
|
and ws.__json__()
|
|
or {
|
|
"name": f"<undefined:{name}>",
|
|
"index": -1,
|
|
"active": True,
|
|
"visible": True,
|
|
"focused": False,
|
|
"alerted": False,
|
|
}
|
|
for name, ws in self.workspace_tree.current_context.visible_workspaces.items()
|
|
},
|
|
}
|
|
except Exception as e:
|
|
print(e)
|
|
return {}
|
|
|
|
@dbus_method_async(input_signature="", result_signature="s")
|
|
async def get_workspace_data(self) -> str:
|
|
"""Get the workspace tree."""
|
|
data = await self.__get_workspace_data()
|
|
return json.dumps(data)
|
|
|
|
@dbus_method_async(input_signature="", result_signature="s")
|
|
async def get_workspace_definition(self) -> str:
|
|
"""Get the definition of the currently focused workspace."""
|
|
workspace = self.workspace_tree.current_context.active_workspace
|
|
return json.dumps(workspace.definition)
|
|
|
|
@dbus_signal_async("s")
|
|
async def tree_changed(self):
|
|
"""Signal emitted when the workspace tree changes."""
|
|
raise NotImplementedError
|
|
|
|
async def on_tree_changed(self, workspace_tree: str):
|
|
print("tree changed, emitting signal")
|
|
self.tree_changed.emit(json.dumps(await self.__get_workspace_data()))
|