ezri/VoidShell
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Sway service improvements

Browse Source
This commit is contained in:
Ezri Brimhall 2024-11-17 14:14:43 -07:00
parent c2504c9cf9
commit 68730f0e50
Signed by: ezri
GPG Key ID: 058A78E5680C6F24
2 changed files with 118 additions and 61 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

57
VoidShell/services/sway/config.py Normal file
View File

@ -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

122
VoidShell/services/sway/tree.py
View File

@ -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)