tabbed-casilda/tabbed_casilda.py

#!/usr/bin/env python3
# /// script
# requires-python = ">=3.11,<3.14"
# dependencies = [
#     "PyGObject",
# ]
# ///
import argparse
import atexit
import os
import shlex
import shutil
import sys

import gi

gi.require_version("Gtk", "4.0")
gi.require_version("Casilda", "1.0")
from gi.repository import Gio, GLib, Gtk, Casilda

import tabbed_protocol as protocol


class TabbedCasilda(Gtk.Application):
    def __init__(self, commands):
        super().__init__(application_id="local.tabbed-casilda")
        self.win = None
        self.commands = commands
        self.notebook = None
        self.tab_ids = {}
        self.next_tab_id = 0
        self.socket_paths = set()
        self.runtime_dir = os.environ.get("XDG_RUNTIME_DIR", "/tmp")
        self.ctrl_path = os.path.join(self.runtime_dir, "tabbed-casilda.sock")

    def _new_socket_path(self, tab_id):
        path = os.path.join(
            self.runtime_dir, f"tabbed-casilda-{os.getpid()}-{tab_id}.sock"
        )
        try:
            os.unlink(path)
        except FileNotFoundError:
            pass
        self.socket_paths.add(path)
        return path

    def _create_tab(self, command, creator_id=None, spawn=True):
        tab_id = self.next_tab_id
        self.next_tab_id += 1
        compositor = Casilda.Compositor(socket=self._new_socket_path(tab_id))
        compositor.props.maximized = True
        label = Gtk.Label(label=command)
        position = -1
        if creator_id is not None:
            creator = self.tab_ids.get(creator_id)
            if creator is not None:
                page = self.notebook.page_num(creator)
                if page != -1:
                    position = page + 1
        self.notebook.insert_page(compositor, label, position)
        self.tab_ids[tab_id] = compositor
        self.notebook.set_current_page(self.notebook.page_num(compositor))
        if self.win:
            compositor.grab_focus()
        if not spawn:
            return tab_id
        argv = shlex.split(command)
        if not argv:
            return
        argv[0] = shutil.which(argv[0]) or argv[0]
        repo_dir = os.path.dirname(os.path.abspath(__file__))
        env = dict(os.environ)
        env["PATH"] = f"{repo_dir}:{env.get('PATH', '')}"
        env["TABBED_CASILDA_CTRL"] = self.ctrl_path
        env["TABBED_CASILDA_TAB_ID"] = str(tab_id)
        env["G_APPLICATION_NON_UNIQUE"] = "1"
        envp = [f"{k}={v}" for k, v in env.items()]
        compositor.spawn_async(None, argv, envp, GLib.SpawnFlags.DEFAULT)

    def start_control_channel(self):
        try:
            os.unlink(self.ctrl_path)
        except FileNotFoundError:
            pass
        service = Gio.SocketService()
        address = Gio.UnixSocketAddress.new(self.ctrl_path)
        service.add_address(
            address, Gio.SocketType.STREAM, Gio.SocketProtocol.DEFAULT, None
        )
        service.connect("incoming", self.on_incoming)
        service.start()
        self.ctrl_service = service
        atexit.register(self.cleanup_socket)

    def cleanup_socket(self):
        for path in list(self.socket_paths):
            try:
                os.unlink(path)
            except FileNotFoundError:
                pass
        try:
            os.unlink(self.ctrl_path)
        except FileNotFoundError:
            pass

    def on_incoming(self, service, connection, source_object):
        stream = Gio.DataInputStream.new(connection.get_input_stream())
        creator_line, _ = stream.read_line()
        command_line, _ = stream.read_line()
        if command_line:
            creator_id, command = protocol.parse(
                creator_line.decode().strip(), command_line.decode().strip()
            )
            tab_id = self._create_tab(command, creator_id, spawn=False)
            compositor = self.tab_ids[tab_id]
            out = Gio.DataOutputStream.new(connection.get_output_stream())
            out.put_string(protocol.serialize_reply(tab_id, compositor.props.socket))
            out.flush()
        return False

    def do_activate(self):
        if self.win:
            self.win.present()
            return
        win = Gtk.ApplicationWindow(
            application=self,
            title="tabbed casilda",
            default_width=1280,
            default_height=720,
        )
        nb = Gtk.Notebook()
        win.set_child(nb)
        self.notebook = nb
        self.start_control_channel()
        for command in self.commands:
            self._create_tab(command)
        win.present()
        self.win = win
        first = self.tab_ids.get(0)
        if first is not None:
            first.grab_focus()


if __name__ == "__main__":
    parser = argparse.ArgumentParser(
        description="GTK4 notebook, one tab per embedded Wayland terminal",
    )
    parser.add_argument(
        "commands",
        nargs="+",
        metavar="COMMAND",
        help='quoted command, e.g. "foot -T htop"',
    )
    args = parser.parse_args(sys.argv[1:])
    app = TabbedCasilda(args.commands)
    sys.exit(app.run([sys.argv[0]]))