From 68730f0e501e196701820cb8b8906e1b15a68ba3 Mon Sep 17 00:00:00 2001 From: Ezri Brimhall Date: Sun, 17 Nov 2024 14:14:43 -0700 Subject: [PATCH] Sway service improvements --- VoidShell/services/sway/config.py | 57 ++++++++++++++ VoidShell/services/sway/tree.py | 122 +++++++++++++++--------------- 2 files changed, 118 insertions(+), 61 deletions(-) create mode 100644 VoidShell/services/sway/config.py diff --git a/VoidShell/services/sway/config.py b/VoidShell/services/sway/config.py new file mode 100644 index 0000000..6f5b9b9 --- /dev/null +++ b/VoidShell/services/sway/config.py @@ -0,0 +1,57 @@ +from typing import TypedDict, NotRequired, Tuple + + +class MemoryProfile(TypedDict): + """Defines a memory profile for an application implemented through SystemD resource control""" + + high: str + max: str + + +class Workspace(TypedDict): + """Defines a workspace""" + + index: int + name: str + exec: str + program_name: str + args: NotRequired[list[str]] + environ: NotRequired[dict[str, str]] + systemd: NotRequired[bool] + void_ouptut: NotRequired[bool] + memory_profile: NotRequired[MemoryProfile] + + +class Output(TypedDict): + """Defines an output to match for contexts""" + + make: NotRequired[str] + model: NotRequired[str] + serial: NotRequired[str] + names: NotRequired[list[str]] + group: NotRequired[str] + position: Tuple[int, int] + mode: str + bars: list[str] + + +class Group(TypedDict): + """Defines a group of workspaces, that exists on a single output""" + + workspaces: list[int] + reverse: NotRequired[bool] + + +class Context(TypedDict): + """Defines a context for a workspace""" + + outputs: list[Output] + primary: str + groups: dict[str, Group] + priority: NotRequired[int] + + +class Config(TypedDict): + workspaces: list[Workspace] + contexts: dict[str, Context] + default_context: str diff --git a/VoidShell/services/sway/tree.py b/VoidShell/services/sway/tree.py index 061b31b..a50a3c6 100644 --- a/VoidShell/services/sway/tree.py +++ b/VoidShell/services/sway/tree.py @@ -1,30 +1,31 @@ from typing import Any import json from i3ipc.replies import OutputReply, WorkspaceReply -from i3ipc.aio import Connection import asyncio -import subprocess from .utils import OutputMatch from .sway import SwayIPC from gi.repository import GObject +from . import config class Workspace(GObject.Object): - """A class representing a Sway workspace in the format we use. - - Attributes: - index: The index of the workspace, as a string. This is the Sway workspace name, and is never displayed to the user. - name: The name of the workspace, as a string. This is the name that is displayed to the user, as defined in `workspaces.json`. - """ + """A class representing a Sway workspace in the format we use.""" _index: int = None _name: str = None + _definition: config.Workspace = None _active: bool = False _visible: bool = False _focused: bool = False _alerted: bool = False + # This is not a GObject property because it won't ever change, and we don't + # need GObject to be aware of it. + @property + def definition(self) -> config.Workspace: + return self._definition + @GObject.Property def index(self) -> int: return self._index @@ -49,12 +50,12 @@ class Workspace(GObject.Object): def alerted(self) -> bool: return self._alerted - def __init__(self, dictionary: dict[str, str | int | dict[str, str]]): + def __init__(self, dictionary: config.Workspace): super().__init__() - self.index = dictionary.get("index") - self.name = dictionary.get("name") + self._index = dictionary.get("index") + self._name = dictionary.get("name") - self.definition = dictionary + self._definition = dictionary def update_state(self, state_update: Any): self._active = True @@ -72,7 +73,7 @@ class Workspace(GObject.Object): self.visible = False self.focused = False self.alerted = False - + self.notify("active") self.notify("visible") self.notify("focused") @@ -104,13 +105,16 @@ class Workspace(GObject.Object): class WorkspaceGroup(GObject.Object): - """A class representing a group of workspaces. Generally maps to a monitor.""" + """ + A class representing a group of workspaces. + Generally maps to a monitor. + """ name: str = None workspaces: list[Workspace] = None reverse: bool = False - def __init__(self, output_data: dict[str, str]): + def __init__(self, output_data: config.Output): super().__init__() self.name = output_data["group"] self.output_names = output_data.get("names", []) @@ -120,12 +124,17 @@ class WorkspaceGroup(GObject.Object): self.position = output_data.get("position", [0, 0]) self.mode = output_data.get("mode", None) self.transform = output_data.get("transform", None) - self.eww_windows = output_data.get("eww_windows", []) + self.eww_windows = output_data.get("bars", output_data.get("eww_windows", [])) self.workspaces = [] def add_workspace(self, workspace: Workspace): """Add a workspace to the group.""" self.workspaces.append(workspace) + workspace.connect("notify::active", self.on_workspace_active) + + def on_workspace_active(self, workspace: Workspace, _): + """Handle the active state of a workspace changing.""" + self.notify("active") def __iter__(self): return iter(self.workspaces) @@ -133,7 +142,7 @@ class WorkspaceGroup(GObject.Object): def __repr__(self): return f"WorkspaceGroup({self.name}, {repr(self.workspaces)})" - @property + @GObject.Property def active_workspace(self) -> Workspace: """Returns the active workspace in the group.""" return next( @@ -141,7 +150,7 @@ class WorkspaceGroup(GObject.Object): None, ) - @property + @GObject.Property def active(self) -> bool: """Returns whether the group is active.""" return any(workspace.focused for workspace in self.workspaces) @@ -159,30 +168,19 @@ class WorkspaceGroup(GObject.Object): "No output name or make/model/serial provided, cannot focus group" ) - async def configure(self, outputs: list[OutputReply]): + async def configure(self, output: OutputReply): """Configure the group output in Sway.""" - transform = "" - mode = "" - selector = "" - if self.transform: - transform = f"transform {self.transform}" - if self.mode: - mode = f"mode {self.mode}" - if self.make and self.model and self.serial: - selector = f'"{self.make} {self.model} {self.serial}"' - elif len(self.output_names) > 0: - for name in self.output_names: - if name in [output.name for output in outputs]: - selector = name - break + sway = SwayIPC.get_instance() + self.output_name = output.name # Configure the output - await i3.command( - f"output {selector} position {self.position[0]} {self.position[1]} {mode} {transform} enable" + await sway.command( + f"output {self.output_name} position {self.position[0]} {self.position[1]} {self.mode} enable" ) - async def validate(self, i3: Connection, workspaces: list[WorkspaceReply]): + async def validate(self, workspaces: list[WorkspaceReply]): """Validate that each workspace in the group is assigned to and present on the correct output.""" - ouput_name = self.get_output_name(i3) + sway = SwayIPC.get_instance() + ouput_name = self.get_output_name(sway) for workspace in self.workspaces: # Get output name for workspace @@ -205,11 +203,12 @@ class WorkspaceGroup(GObject.Object): f"Workspace {workspace.index} is assigned to {workspace_output.output}, not {ouput_name}, reassigning", flush=True, ) - await workspace.relocate(i3, ouput_name) + await workspace.relocate(sway, ouput_name) - async def get_output_name(self, i3: Connection) -> str: + async def get_output_name(self) -> str: """Get the name of the output in Sway.""" - outputs = await i3.get_outputs() + sway = SwayIPC.get_instance() + outputs = await sway.get_outputs() # If we have make, model, and serial, search by those first if self.make and self.model and self.serial: for output in outputs: @@ -311,25 +310,24 @@ class WorkspaceContext: result += group_result return result - async def deactivate(self, i3: Connection): + async def deactivate(self): """Deactivate the context in Sway.""" # First, close all EWW windows + sway = SwayIPC.get_instance() proc = await asyncio.create_subprocess_exec("eww", "close-all") await proc.wait() # Then, disable all displays - await i3.command("output * disable") + await sway.command("output * disable") - async def activate(self, i3: Connection): + async def activate(self): """Activate the context in Sway.""" - defined_displays = [ - f"{group.make} {group.model} {group.serial}" for group in self.groups - ] - outputs = await i3.get_outputs() + sway = SwayIPC.get_instance() + outputs = await sway.get_outputs() # Configure all displays defined in the context for group in self.groups: - await group.configure(i3, outputs) + await group.configure(sway, outputs) # Then, open all EWW windows defined in the context on the appropriate windows proc = await asyncio.create_subprocess_exec("eww", "reload") @@ -341,14 +339,14 @@ class WorkspaceContext: "open", window, "--screen", - await group.get_output_name(i3), + await group.get_output_name(sway), ) await proc.wait() # Finally, validate workspaces and move them if necessary for group in self.groups: - workspaces = await i3.get_workspaces() - await group.validate(i3, workspaces) + workspaces = await sway.get_workspaces() + await group.validate(sway, workspaces) @property def active_group(self) -> WorkspaceGroup: @@ -481,10 +479,11 @@ class WorkspaceTree: if ws not in touched: ws.deactivate() - async def update_context(self, i3: Connection, match_context_on_name: bool = False): + async def update_context(self, match_context_on_name: bool = False): """Activates a new context in Sway based on the current display configuration.""" # First, get the current display configuration - outputs = await i3.get_outputs() + sway = SwayIPC.get_instance() + outputs = await sway.get_outputs() print(outputs, flush=True) # Next, calculate match scores for each context @@ -503,28 +502,29 @@ class WorkspaceTree: ) return if self.current_context is None: - await scores[0][0].activate(i3) + await scores[0][0].activate(sway) self.current_context = scores[0][0] return elif scores[0][0] == self.current_context: print("Context is already active.", flush=True) elif match_context_on_name and scores[0][0].name == self.current_context.name: print("Context is already active.", flush=True) - await scores[0][0].activate(i3) + await scores[0][0].activate(sway) self.current_context = scores[0][0] else: - await self.current_context.deactivate(i3) - await scores[0][0].activate(i3) + await self.current_context.deactivate(sway) + await scores[0][0].activate(sway) self.current_context = scores[0][0] - async def activate_context(self, i3: Connection, name: str): + async def activate_context(self, name: str): """Activates a context by name. This will fail if the current display configuration is incompatible.""" + sway = SwayIPC.get_instance() context = next( (context for context in self.contexts if context.name == name), None, ) - outputs = await i3.get_outputs() + outputs = await sway.get_outputs() score = context.compatability_rank(outputs) if score == 0: @@ -533,7 +533,7 @@ class WorkspaceTree: ) print(f"Activating context {context.name} with score {score}.", flush=True) if self.current_context is not None and self.current_context != context: - await self.current_context.deactivate(i3) + await self.current_context.deactivate() self.current_context = context - await context.activate(i3) + await context.activate(sway)