kitty_new_tab.py

kitty_new_tab.py

/usr/bin/python3 -c '
#!/usr/bin/python3
import glob
import os
import shlex
import shutil
import stat
import subprocess
import sys
import time


DEBUG = os.environ.get("KITTYNEWTAB_DEBUG") == "1"
SOCKET_PREFIX = "/tmp/mykitty"
SOCKET_TEMPLATE = f"{SOCKET_PREFIX}-{{kitty_pid}}"
KITTEN = None


def main():
    dirs = dirs_from_input(sys.stdin.read())

    if os.environ.get("KITTYNEWTAB_DRY_RUN") == "1":
        print(*(dirs or ["(default)"]), sep="\n")
        return 0

    try:
        open_in_kitty(dirs)
    except (OSError, RuntimeError, subprocess.CalledProcessError) as exc:
        print(f"knt: {exc}", file=sys.stderr)
        return 1

    return 0


def dirs_from_input(text):
    dirs = []
    seen = set()

    for line in text.splitlines():
        path = clean_path(line)
        if not path:
            continue

        if os.path.exists(path) and not os.path.isdir(path):
            path = os.path.dirname(path)

        if os.path.isdir(path) and path not in seen:
            dirs.append(path)
            seen.add(path)

    logd("dirs =", dirs or "(default)")
    return dirs


def clean_path(path):
    path = path.strip()
    if not path:
        return ""

    if len(path) >= 2 and path[0] == path[-1] and path[0] in ("\x27", "\x22"):
        path = path[1:-1]

    if path == "~":
        return os.environ["HOME"]
    if path.startswith("~/"):
        return os.path.join(os.environ["HOME"], path[2:])
    return path


def open_in_kitty(dirs):
    socket = newest_socket()

    if socket:
        logd("kitty: unix socket found:", socket)
        for cwd in dirs or [None]:
            launch_tab(socket, cwd)
        bounce_window(socket)
        return

    logd("kitty: no unix socket found")
    open_kitty(dirs[0] if dirs else None)

    if len(dirs) <= 1:
        return

    socket = wait_for_socket()
    if not socket:
        print("kitty: socket did not appear in time", file=sys.stderr)
        return

    logd("kitty: unix socket found:", socket)
    for cwd in dirs[1:]:
        launch_tab(socket, cwd)
    bounce_window(socket)


def open_kitty(cwd):
    cmd = [
        "/usr/bin/open",
        "-a",
        "kitty.app",
        "--args",
        "--single-instance",
        f"--listen-on=unix:{SOCKET_TEMPLATE}",
    ]
    if cwd:
        cmd.append(f"--directory={cwd}")
    run(cmd)


def launch_tab(socket, cwd):
    cmd = ["launch", "--type=tab"]
    if cwd:
        cmd.append(f"--cwd={cwd}")
    kitten(socket, *cmd)


def bounce_window(socket):
    kitten(socket, "resize-os-window", "--action=hide")
    kitten(socket, "resize-os-window", "--action=show")


def kitten(socket, *args):
    run([kitten_bin(), "@", f"--to=unix:{socket}", *args])


def kitten_bin():
    global KITTEN

    if KITTEN:
        return KITTEN

    for path in (
        shutil.which("kitten"),
        "/usr/local/bin/kitten",
        "/opt/homebrew/bin/kitten",
        "/Applications/kitty.app/Contents/MacOS/kitten",
    ):
        if path and os.access(path, os.X_OK):
            KITTEN = path
            return KITTEN

    raise RuntimeError("could not find kitten executable")


def newest_socket():
    newest_path = None
    newest_mtime = -1

    for path in glob.glob(f"{SOCKET_PREFIX}-*"):
        try:
            path_stat = os.stat(path)
        except OSError:
            continue

        if stat.S_ISSOCK(path_stat.st_mode) and path_stat.st_mtime_ns > newest_mtime:
            newest_path = path
            newest_mtime = path_stat.st_mtime_ns

    return newest_path


def wait_for_socket(timeout=5.0, interval=0.1):
    deadline = time.monotonic() + timeout
    while time.monotonic() < deadline:
        socket = newest_socket()
        if socket:
            return socket
        time.sleep(interval)
    return None


def run(cmd):
    logd("+", shlex.join(cmd))
    subprocess.run(cmd, check=True)


def logd(*parts):
    if DEBUG:
        print(*parts, file=sys.stderr)

main()

'