diff --git a/.config/eww/scripts/sway.py b/.config/eww/scripts/sway.py index 1fccee7..fa6ce1f 100755 --- a/.config/eww/scripts/sway.py +++ b/.config/eww/scripts/sway.py @@ -9,11 +9,11 @@ from i3ipc.aio import Connection from i3ipc import Event from typing import List, Dict, Any, Union, Tuple -WorkspaceTree = Dict[str, Dict[str, List['Workspace']]] -WorkspaceList = Dict[str, 'Workspace'] - -class Workspace: +WorkspaceTree = Dict[str, Dict[str, List["Workspace"]]] +WorkspaceList = Dict[str, "Workspace"] + +class Workspace: index: str = None name: str = None exec_cmd: str = None @@ -25,15 +25,15 @@ class Workspace: alerted: bool = False def __init__(self, dictionary: Dict[str, Union[str, int]], group: str): - self.index = str(dictionary.get('index')) - self.name = dictionary.get('name') - self.exec_cmd = dictionary.get('exec') + self.index = str(dictionary.get("index")) + self.name = dictionary.get("name") + self.exec_cmd = dictionary.get("exec") self.group = group - self.active = dictionary.get('active', False) - self.visible = dictionary.get('visible', False) - self.focused = dictionary.get('focused', False) - self.alerted = dictionary.get('alerted', False) + self.active = dictionary.get("active", False) + self.visible = dictionary.get("visible", False) + self.focused = dictionary.get("focused", False) + self.alerted = dictionary.get("alerted", False) def update_state(self, state_update: Any): self.active = True @@ -49,17 +49,18 @@ class Workspace: @classmethod def parse_file(cls: type, filename: str) -> Tuple[WorkspaceTree, WorkspaceList]: - result = {} initial: dict = None - with open(filename, 'r') as file: + with open(filename, "r") as file: initial = json.load(file) workspaces: WorkspaceList = {} - def find_workspace(workspace: Dict[str, Union[str, int]], group: str) -> 'Workspace': - index = str(workspace.get('index')) + def find_workspace( + workspace: Dict[str, Union[str, int]], group: str + ) -> "Workspace": + index = str(workspace.get("index")) if index in workspaces: return workspaces[index] else: @@ -68,10 +69,10 @@ class Workspace: result = { context: { - group: [ - find_workspace(workspace, group) for workspace in workspaces - ] for group, workspaces in groups.items() - } for context, groups in initial.items() + group: [find_workspace(workspace, group) for workspace in workspaces] + for group, workspaces in groups.items() + } + for context, groups in initial.items() } return result, workspaces @@ -80,106 +81,114 @@ class Workspace: def full_dictify(tree: WorkspaceTree): return { context: { - group: [ - workspace.dictify() for workspace in workspaces - ] for group, workspaces in groups.items() - } for context, groups in tree.items() + group: [workspace.dictify() for workspace in workspaces] + for group, workspaces in groups.items() + } + for context, groups in tree.items() } def dictify(self): - return { - 'index': self.index, - 'name': self.name, - 'exec': self.exec_cmd, - 'active': self.active, - 'visible': self.visible, - 'focused': self.focused, - 'alerted': self.alerted + "index": self.index, + "name": self.name, + "exec": self.exec_cmd, + "active": self.active, + "visible": self.visible, + "focused": self.focused, + "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 = { "ws": {}, "mode": "default", "current": {}, - "context": "work" if gethostname() == "tycho" else "personal", + "context": "work" if gethostname() == "tycho.vpn.ezri.dev" else "personal", "visible": {}, } + def write_data(): global data print(json.dumps(data), flush=True) + async def workspace_event(self: Connection, _): 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 = [] for index, ws_data in ws_sway_dict.items(): - # Update data for all active workspaces ws_state = workspace_list.get(index) if ws_state is None: # 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: - data['current'] = { - 'index': index, - 'name': 'undefined', - 'exec': 'alacritty', - 'active': True, - 'visible': True, - 'focused': True, - 'alerted': False + data["current"] = { + "index": index, + "name": "undefined", + "exec": "alacritty", + "active": True, + "visible": True, + "focused": True, + "alerted": False, } else: # Otherwise, 'touch' it and update the status data touched.append(index) ws_state.update_state(ws_data) if ws_state.focused: - data['current'] = ws_state.dictify() + data["current"] = ws_state.dictify() if ws_state.visible: - data['visible'][ws_state.group] = ws_state.dictify() + data["visible"][ws_state.group] = ws_state.dictify() # 'Deactivate' any untouched workspaces - inactives = [ - v for k, v in workspace_list.items() if v.active and k not in touched - ] + inactives = [v for k, v in workspace_list.items() if v.active and k not in touched] for workspace in inactives: workspace.deactivate() # 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(): - if group not in data['visible']: - data['visible'][group] = { - 'index': 'U', - 'name': 'undefined', - 'exec': 'alacritty', - 'active': True, - 'visible': True, - 'focused': False, - 'alerted': False + for group in workspace_tree[data["context"] or "personal"].keys(): + if group not in data["visible"]: + data["visible"][group] = { + "index": "U", + "name": "undefined", + "exec": "alacritty", + "active": True, + "visible": True, + "focused": False, + "alerted": False, } - data['ws'] = Workspace.full_dictify(workspace_tree) + data["ws"] = Workspace.full_dictify(workspace_tree) write_data() + async def mode_event(self: Connection, event): global data - data['mode'] = event.change + data["mode"] = event.change write_data() + 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.MODE, mode_event) await workspace_event(sway, None) await sway.main() + asyncio.run(main())