dotfiles/.config/eww/scripts/sway.py

#!/usr/bin/env python3

import asyncio
import json
import os
import sys
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:

    index: str = None
    name: str = None
    exec_cmd: str = None
    group: str = None

    active: bool = False
    visible: bool = False
    focused: bool = False
    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.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)

    def update_state(self, state_update: Any):
        self.active = True
        self.visible = state_update.visible
        self.focused = state_update.focused
        self.alerted = state_update.urgent

    def deactivate(self):
        self.active = False
        self.visible = False
        self.focused = False
        self.alerted = False

    @classmethod
    def parse_file(cls: type, filename: str) -> Tuple[WorkspaceTree, WorkspaceList]:

        result = {}

        initial: dict = None
        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'))
            if index in workspaces:
                return workspaces[index]
            else:
                workspaces[index] = cls(workspace, group)
                return workspaces[index]

        result = {
            context: {
                group: [
                    find_workspace(workspace, group) for workspace in workspaces
                ] for group, workspaces in groups.items()
            } for context, groups in initial.items()
        }

        return result, workspaces

    @staticmethod
    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()
        }

    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
        }

workspace_tree, workspace_list = Workspace.parse_file(f"{os.environ['HOME']}/.config/sway/workspaces.json")

data = {
    "ws": {},
    "mode": "default",
    "current": {},
    "context": "work",
    "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()}

    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)
            if ws_data.focused:
                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()
            if ws_state.visible:
                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
    ]

    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
            }

    data['ws'] = Workspace.full_dictify(workspace_tree)
    write_data()

async def mode_event(self: Connection, event):
    global data
    data['mode'] = event.change
    write_data()

async def main():
    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())