Fixed sway eww script

This commit is contained in:
2024-03-07 14:21:27 -07:00
parent 71413ed216
commit c795ee94ff

View File

@@ -9,11 +9,11 @@ from i3ipc.aio import Connection
from i3ipc import Event from i3ipc import Event
from typing import List, Dict, Any, Union, Tuple from typing import List, Dict, Any, Union, Tuple
WorkspaceTree = Dict[str, Dict[str, List['Workspace']]] WorkspaceTree = Dict[str, Dict[str, List["Workspace"]]]
WorkspaceList = Dict[str, 'Workspace'] WorkspaceList = Dict[str, "Workspace"]
class Workspace:
class Workspace:
index: str = None index: str = None
name: str = None name: str = None
exec_cmd: str = None exec_cmd: str = None
@@ -25,15 +25,15 @@ class Workspace:
alerted: bool = False alerted: bool = False
def __init__(self, dictionary: Dict[str, Union[str, int]], group: str): def __init__(self, dictionary: Dict[str, Union[str, int]], group: str):
self.index = str(dictionary.get('index')) self.index = str(dictionary.get("index"))
self.name = dictionary.get('name') self.name = dictionary.get("name")
self.exec_cmd = dictionary.get('exec') self.exec_cmd = dictionary.get("exec")
self.group = group self.group = group
self.active = dictionary.get('active', False) self.active = dictionary.get("active", False)
self.visible = dictionary.get('visible', False) self.visible = dictionary.get("visible", False)
self.focused = dictionary.get('focused', False) self.focused = dictionary.get("focused", False)
self.alerted = dictionary.get('alerted', False) self.alerted = dictionary.get("alerted", False)
def update_state(self, state_update: Any): def update_state(self, state_update: Any):
self.active = True self.active = True
@@ -49,17 +49,18 @@ class Workspace:
@classmethod @classmethod
def parse_file(cls: type, filename: str) -> Tuple[WorkspaceTree, WorkspaceList]: def parse_file(cls: type, filename: str) -> Tuple[WorkspaceTree, WorkspaceList]:
result = {} result = {}
initial: dict = None initial: dict = None
with open(filename, 'r') as file: with open(filename, "r") as file:
initial = json.load(file) initial = json.load(file)
workspaces: WorkspaceList = {} workspaces: WorkspaceList = {}
def find_workspace(workspace: Dict[str, Union[str, int]], group: str) -> 'Workspace': def find_workspace(
index = str(workspace.get('index')) workspace: Dict[str, Union[str, int]], group: str
) -> "Workspace":
index = str(workspace.get("index"))
if index in workspaces: if index in workspaces:
return workspaces[index] return workspaces[index]
else: else:
@@ -68,10 +69,10 @@ class Workspace:
result = { result = {
context: { context: {
group: [ group: [find_workspace(workspace, group) for workspace in workspaces]
find_workspace(workspace, group) for workspace in workspaces for group, workspaces in groups.items()
] for group, workspaces in groups.items() }
} for context, groups in initial.items() for context, groups in initial.items()
} }
return result, workspaces return result, workspaces
@@ -80,106 +81,114 @@ class Workspace:
def full_dictify(tree: WorkspaceTree): def full_dictify(tree: WorkspaceTree):
return { return {
context: { context: {
group: [ group: [workspace.dictify() for workspace in workspaces]
workspace.dictify() for workspace in workspaces for group, workspaces in groups.items()
] for group, workspaces in groups.items() }
} for context, groups in tree.items() for context, groups in tree.items()
} }
def dictify(self): def dictify(self):
return { return {
'index': self.index, "index": self.index,
'name': self.name, "name": self.name,
'exec': self.exec_cmd, "exec": self.exec_cmd,
'active': self.active, "active": self.active,
'visible': self.visible, "visible": self.visible,
'focused': self.focused, "focused": self.focused,
'alerted': self.alerted "alerted": self.alerted,
} }
workspace_tree, workspace_list = Workspace.parse_file(f"{os.environ['HOME']}/.config/sway/workspaces.json")
workspace_tree, workspace_list = Workspace.parse_file(
f"{os.environ['HOME']}/.config/sway/workspaces.json"
)
data = { data = {
"ws": {}, "ws": {},
"mode": "default", "mode": "default",
"current": {}, "current": {},
"context": "work" if gethostname() == "tycho" else "personal", "context": "work" if gethostname() == "tycho.vpn.ezri.dev" else "personal",
"visible": {}, "visible": {},
} }
def write_data(): def write_data():
global data global data
print(json.dumps(data), flush=True) print(json.dumps(data), flush=True)
async def workspace_event(self: Connection, _): async def workspace_event(self: Connection, _):
global workspace_tree, workspace_list, data global workspace_tree, workspace_list, data
ws_sway_dict = { ws.name: ws for ws in await self.get_workspaces()} ws_sway_dict = {ws.name: ws for ws in await self.get_workspaces()}
touched = [] touched = []
for index, ws_data in ws_sway_dict.items(): for index, ws_data in ws_sway_dict.items():
# Update data for all active workspaces # Update data for all active workspaces
ws_state = workspace_list.get(index) ws_state = workspace_list.get(index)
if ws_state is None: if ws_state is None:
# If the workspace isn't configured, note it and handle gracefully # If the workspace isn't configured, note it and handle gracefully
print(f"Warning: found unconfigured workspace {index}", file=sys.stderr, flush=True) print(
f"Warning: found unconfigured workspace {index}",
file=sys.stderr,
flush=True,
)
if ws_data.focused: if ws_data.focused:
data['current'] = { data["current"] = {
'index': index, "index": index,
'name': 'undefined', "name": "undefined",
'exec': 'alacritty', "exec": "alacritty",
'active': True, "active": True,
'visible': True, "visible": True,
'focused': True, "focused": True,
'alerted': False "alerted": False,
} }
else: else:
# Otherwise, 'touch' it and update the status data # Otherwise, 'touch' it and update the status data
touched.append(index) touched.append(index)
ws_state.update_state(ws_data) ws_state.update_state(ws_data)
if ws_state.focused: if ws_state.focused:
data['current'] = ws_state.dictify() data["current"] = ws_state.dictify()
if ws_state.visible: if ws_state.visible:
data['visible'][ws_state.group] = ws_state.dictify() data["visible"][ws_state.group] = ws_state.dictify()
# 'Deactivate' any untouched workspaces # 'Deactivate' any untouched workspaces
inactives = [ inactives = [v for k, v in workspace_list.items() if v.active and k not in touched]
v for k, v in workspace_list.items() if v.active and k not in touched
]
for workspace in inactives: for workspace in inactives:
workspace.deactivate() workspace.deactivate()
# Create an 'unknown' workspace object for each group that doesn't yet have a listed visible workspace # Create an 'unknown' workspace object for each group that doesn't yet have a listed visible workspace
for group in workspace_tree[data['context'] or 'personal'].keys(): for group in workspace_tree[data["context"] or "personal"].keys():
if group not in data['visible']: if group not in data["visible"]:
data['visible'][group] = { data["visible"][group] = {
'index': 'U', "index": "U",
'name': 'undefined', "name": "undefined",
'exec': 'alacritty', "exec": "alacritty",
'active': True, "active": True,
'visible': True, "visible": True,
'focused': False, "focused": False,
'alerted': False "alerted": False,
} }
data['ws'] = Workspace.full_dictify(workspace_tree) data["ws"] = Workspace.full_dictify(workspace_tree)
write_data() write_data()
async def mode_event(self: Connection, event): async def mode_event(self: Connection, event):
global data global data
data['mode'] = event.change data["mode"] = event.change
write_data() write_data()
async def main(): async def main():
sway = await Connection(auto_reconnect = True).connect() sway = await Connection(auto_reconnect=True).connect()
sway.on(Event.WORKSPACE, workspace_event) sway.on(Event.WORKSPACE, workspace_event)
sway.on(Event.MODE, mode_event) sway.on(Event.MODE, mode_event)
await workspace_event(sway, None) await workspace_event(sway, None)
await sway.main() await sway.main()
asyncio.run(main()) asyncio.run(main())