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

#!/usr/bin/env python3

import asyncio
import json
import os
import sys
from socket import gethostname
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, str]:
        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["contexts"].items()
        }

        default_context = initial.get("default_context", "personal")

        return result, workspaces, default_context

    @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, default_context = Workspace.parse_file(
    f"{os.environ['HOME']}/.config/sway/workspaces.json"
)

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