#!/usr/bin/env python # ── tab_bar.py · kitty tab titles: directory + git state ── # # kitty loads this file from the config dir and exposes `draw_title` as the # `{custom}` field of tab_title_template (see kitty/tab_bar.py: # load_custom_draw_title). kitty keeps drawing the powerline chrome itself; only # the text inside a tab comes from here. # # A title reads: 2 chezmoi main ⇡1 +2 ~1 ?3 # ^ ^ ^ ^ ^ ^ # | dir branch staged / unstaged / untracked files # tab index # # git is *never* run on the render path. `draw_title` only ever reads a cache # and notes which working dirs are on screen; a 1s repeating kitty timer (main # thread, same as every other kitty callback) starts `git status` processes, # reaps the finished ones, and marks the tab bar dirty when a value changed. # Output goes to a temp file, not a pipe, so a repo with thousands of dirty # files can never fill a 64K pipe buffer and wedge the child. from __future__ import annotations import os import subprocess import tempfile import time from typing import Any from kitty.boss import get_boss from kitty.fast_data_types import add_timer # Nerd Font glyphs (VictorMono Nerd Font Mono covers all of these). FOLDER = '\uf07b' # nf-fa-folder BRANCH = '\ue0a0' # nf-pl-branch AHEAD = '\u21e1' BEHIND = '\u21e3' # A cached git summary is reused for this long; a working dir that stopped being # displayed is dropped after PRUNE_AFTER so closed tabs stop being polled. TTL = 4.0 PRUNE_AFTER = 30.0 POLL_INTERVAL = 1.0 MAX_CONCURRENT = 4 KILL_AFTER = 15.0 MAX_DIR_LEN = 24 # wd -> git summary spec: (branch, ahead, behind, staged, unstaged, untracked, # conflicts) or None when wd is not a work tree. _git: dict[str, tuple[str, int, int, int, int, int, int] | None] = {} _fresh_until: dict[str, float] = {} _wanted: dict[str, float] = {} _running: dict[str, tuple[subprocess.Popen[bytes], Any, float]] = {} _timer = 0 def draw_title(data: dict[str, Any]) -> str: global _timer if not _timer: _timer = add_timer(_tick, POLL_INTERVAL, True) wd = data['tab'].active_wd fmt = data['fmt'] parts = [str(data['index'])] if wd: _wanted[wd] = _now() parts.append(f'{FOLDER} {_dir_label(wd)}') summary = _git.get(wd) if summary is not None: parts.append(_render_git(summary, fmt)) else: parts.append(data['title']) return ' '.join(parts) def _dir_label(wd: str) -> str: home = os.path.expanduser('~') if wd == home: return '~' name = os.path.basename(wd.rstrip('/')) or wd if len(name) > MAX_DIR_LEN: name = name[: MAX_DIR_LEN - 1] + '…' return name def _render_git( summary: tuple[str, int, int, int, int, int, int], fmt: Any ) -> str: branch, ahead, behind, staged, unstaged, untracked, conflicts = summary # color2/3/4/1 are palette slots, so the active Omarchy theme picks the # actual greens and reds; fg.tab restores the tab's own foreground so the # powerline chrome keeps drawing in its own colors. tab = fmt.fg.tab out = [f'{BRANCH} {branch}' if branch else BRANCH] for count, mark, color in ( (ahead, AHEAD, fmt.fg.color6), (behind, BEHIND, fmt.fg.color6), (staged, '+', fmt.fg.color2), (unstaged, '~', fmt.fg.color3), (untracked, '?', fmt.fg.color4), (conflicts, '!', fmt.fg.color1), ): if count: out.append(f'{color}{mark}{count}{tab}') return ' '.join(out) def _now() -> float: return time.monotonic() def _tick(timer_id: int | None) -> None: now = _now() changed = _reap(now) for wd, last_wanted in tuple(_wanted.items()): if now - last_wanted > PRUNE_AFTER: del _wanted[wd] _git.pop(wd, None) _fresh_until.pop(wd, None) continue if len(_running) >= MAX_CONCURRENT: break if wd in _running or _fresh_until.get(wd, 0.0) > now: continue _start(wd, now) if changed: for tm in get_boss().os_window_map.values(): tm.mark_tab_bar_dirty() def _start(wd: str, now: float) -> None: try: out = tempfile.TemporaryFile() proc = subprocess.Popen( ( 'git', '--no-optional-locks', 'status', '--porcelain=v2', '--branch', '--untracked-files=normal', ), cwd=wd, stdout=out, stderr=subprocess.DEVNULL, stdin=subprocess.DEVNULL, # start_new_session, not preexec_fn: kitty is multi-threaded, and # only the former is implemented without running Python between # fork and exec. start_new_session=True, ) except OSError: # No git, or wd vanished: remember "not a repo" so we stop retrying # until the TTL lapses. _git[wd] = None _fresh_until[wd] = now + TTL return _running[wd] = (proc, out, now) def _reap(now: float) -> bool: changed = False for wd, (proc, out, started) in tuple(_running.items()): rc = proc.poll() if rc is None: if now - started > KILL_AFTER: proc.kill() continue del _running[wd] summary = None if rc == 0: try: out.seek(0) summary = _parse(out.read().decode('utf-8', 'replace')) except OSError: summary = None out.close() if _git.get(wd, ...) != summary: changed = True _git[wd] = summary _fresh_until[wd] = now + TTL return changed def _parse(text: str) -> tuple[str, int, int, int, int, int, int]: """git status --porcelain=v2 --branch -> title counts. Header lines carry the branch and the ahead/behind pair; `1`/`2` records carry a two-char XY status (X = index, Y = work tree), `?` is untracked and `u` is unmerged. Renames (`2`) put the old path after a tab on the same line, so line-splitting stays correct. """ head = oid = '' ahead = behind = staged = unstaged = untracked = conflicts = 0 for line in text.splitlines(): if line.startswith('# branch.oid '): oid = line[13:20] elif line.startswith('# branch.head '): head = line[14:] elif line.startswith('# branch.ab '): for token in line[12:].split(): if token.startswith('+'): ahead = int(token[1:] or 0) elif token.startswith('-'): behind = int(token[1:] or 0) elif line[:2] in ('1 ', '2 '): xy = line[2:4] if xy[0] != '.': staged += 1 if xy[1] != '.': unstaged += 1 elif line.startswith('? '): untracked += 1 elif line.startswith('u '): conflicts += 1 branch = oid if head in ('', '(detached)') else head return branch, ahead, behind, staged, unstaged, untracked, conflicts