#!/usr/bin/env python3 """termwm -- a tiny mouse-driven window manager that lives inside one terminal. It spawns real shells in pseudo-terminals and shows each one in a floating window you can move and resize with the mouse. It runs screen/tmux-style: a detached daemon owns the shells, and a client attaches the terminal to it. If the terminal dies the shells keep running. wm.py start the daemon if needed, then attach wm.py --attach attach to an already-running daemon wm.py --kill tell the daemon to quit (closing all shells) Ctrl-G d detach (leave everything running in the background) Mouse: * drag a window's title bar -> move it * drag the bottom-right corner -> resize it * click [x] on the title bar -> close that window * click a window -> raise / focus it * click [+ New] / [Quit] on the top menu bar Keyboard (forwarded to the focused shell), with a Ctrl-G prefix for commands: Ctrl-G n new terminal Ctrl-G w close focused window Ctrl-G m maximize/restore Ctrl-G t tile into a grid Ctrl-G f cycle full-screen (float / 1-col / 2-col) Ctrl-Space circulate Ctrl-Left/Right move between columns (full 2-col, needs width > 160) Ctrl-Enter (or Ctrl-G Enter) switch to previously focused window Ctrl-G Tab cycle focus Ctrl-G q quit Ctrl-G g send literal Ctrl-G [Search] enters a timestamped history viewer of the focused window: up/down move a line, PgUp/PgDn a page, the date/time of the top line shows on the bar, ESC exits. No third-party dependencies -- a small built-in VT100/ECMA-48 emulator (see the VT class) parses each child's output, so plain `python3 wm.py` is enough. """ import os import sys import pty import time import codecs import select import signal import socket import struct import fcntl import termios import tty import subprocess # --- terminal control sequences -------------------------------------------- ALT_ON, ALT_OFF = "\033[?1049h", "\033[?1049l" HIDE_CUR, SHOW_CUR = "\033[?25l", "\033[?25h" MOUSE_ON = "\033[?1002h\033[?1006h" # 1002: drag reporting, 1006: SGR coords MOUSE_OFF = "\033[?1006l\033[?1002l" PREFIX = b"\x07" # Ctrl-G # Ctrl-Enter, as reported by CSI-u (kitty/fixterms) or xterm modifyOtherKeys. # (Plain terminals can't tell Ctrl-Enter from Enter; Ctrl-G Enter is the fallback.) KEY_SWITCH = (b"\x1b[13;5u", b"\x1b[27;5;13~") # Ctrl-Left / Ctrl-Right (move between full-screen columns) -> target column 0/1 FS_COLKEYS = {b"\x1b[1;5D": 0, b"\x1b[1;5C": 1} TWO_COL_MIN = 160 # need this many columns to offer full x2 # color name -> ANSI 0..7 NAMED = { "black": 0, "red": 1, "green": 2, "brown": 3, "yellow": 3, "blue": 4, "magenta": 5, "cyan": 6, "white": 7, } COLOR_NAMES = ["black", "red", "green", "brown", "blue", "magenta", "cyan", "white"] MIN_W, MIN_H = 14, 5 SCROLLBACK_MAX = 10000 # per-window timestamped history lines HELP_LINES = [ "Mouse", " drag title bar ....... move window", " drag border / corner . resize window", " [x] on title bar ..... close window", " click a window ....... focus / raise", " drag in the body ..... select text (copies on release)", " double-click a word .. select that word", " double-click title ... maximize / restore", "", "Keyboard (Ctrl-G is the command prefix)", " Ctrl-G n new window Ctrl-G w close window", " Ctrl-G m max / restore Ctrl-G t tile into a grid", " Ctrl-G f cycle full (float / 1-col / 2-col) Ctrl-G d detach", " Ctrl-G q quit (kill) Ctrl-G g send a literal Ctrl-G", " Ctrl-Space circulate Ctrl-Enter switch to previous window", " Ctrl-Left / Ctrl-Right move between columns (full 2-col)", "", "Menu New Grid Full Search Detach Help Quit", "Session wm.py [--attach] [--kill] (shells survive detach)", "", "Press any key to close this help.", ] # cheesy 3D background banner ------------------------------------------------ LOGO_WORD = "WM-TERM" LOGO_FONT = { "W": ["# #", "# #", "# # #", "# # # #", " # # "], "M": ["# #", "## ##", "# # # #", "# # #", "# #"], "-": [" ", " ", "#####", " ", " "], "T": ["#######", " # ", " # ", " # ", " # "], "E": ["######", "# ", "#### ", "# ", "######"], "R": ["##### ", "# #", "##### ", "# # ", "# # "], } # bright per-letter face colors (cheesy rainbow) LOGO_COLORS = ["0;1;31", "0;1;33", "0;1;32", "0;1;36", "0;1;34", "0;1;35", "0;1;31"] def out(s): sys.stdout.write(s) sys.stdout.flush() def xterm_hex(i): """Map an xterm-256 palette index to an 'rrggbb' hex string.""" if i < 16: base = [(0, 0, 0), (205, 0, 0), (0, 205, 0), (205, 205, 0), (0, 0, 238), (205, 0, 205), (0, 205, 205), (229, 229, 229), (127, 127, 127), (255, 0, 0), (0, 255, 0), (255, 255, 0), (92, 92, 255), (255, 0, 255), (0, 255, 255), (255, 255, 255)] r, g, b = base[i] elif i < 232: i -= 16 conv = lambda v: 0 if v == 0 else 55 + 40 * v r, g, b = conv(i // 36), conv((i // 6) % 6), conv(i % 6) else: v = 8 + (i - 232) * 10 r, g, b = v, v, v return "%02x%02x%02x" % (r, g, b) def cell_sgr(c): """Translate a screen cell into an absolute SGR parameter string.""" p = ["0"] if c.reverse: p.append("7") if c.bold: p.append("1") if c.italics: p.append("3") if c.underscore: p.append("4") for color, base in ((c.fg, 30), (c.bg, 40)): if not color or color == "default": continue code = NAMED.get(color) if code is not None: p.append(str(base + code)) elif len(color) == 6: try: r, g, b = (int(color[i:i + 2], 16) for i in (0, 2, 4)) p += [str(base + 8), "2", str(r), str(g), str(b)] except ValueError: pass return ";".join(p) # --- a small built-in terminal emulator ------------------------------------- class Cell: __slots__ = ("data", "fg", "bg", "bold", "italics", "underscore", "reverse") def __init__(self): self.data = " " self.fg = self.bg = "default" self.bold = self.italics = self.underscore = self.reverse = False class Cursor: __slots__ = ("x", "y", "hidden") def __init__(self): self.x = self.y = 0 self.hidden = False class VT: """A compact VT100/ECMA-48 screen emulator: feed it bytes, read .buffer. Supports the subset that ordinary shells and curses apps (vim, htop, less, top) rely on: cursor motion, erase, SGR colors/attrs (16/256/truecolor), scroll regions, insert/delete line & char, the alternate screen, autowrap, DECTCEM cursor visibility, and OSC 0/2 window titles. """ def __init__(self, columns, lines): self.columns, self.lines = columns, lines self.cursor = Cursor() self.title = "" self.scrollback = [] # [(epoch_time, text), ...] self._dec = codecs.getincrementaldecoder("utf-8")("replace") self._state = "ground" self._params = "" self._priv = "" self._osc = "" self._osc_esc = False self._wrap = False self._saved = (0, 0) self._top, self._bottom = 0, lines - 1 self._reset_attrs() self._main = self._blank_grid() self._alt = self._blank_grid() self._in_alt = False self.buffer = self._main # -- helpers ------------------------------------------------------------- def _reset_attrs(self): self.fg = self.bg = "default" self.bold = self.italics = self.underscore = self.reverse = False def _blank_grid(self): return [[Cell() for _ in range(self.columns)] for _ in range(self.lines)] def _blank_line(self): return [Cell() for _ in range(self.columns)] def resize(self, lines, columns): def rebuild(old): new = [[Cell() for _ in range(columns)] for _ in range(lines)] for y in range(min(lines, self.lines)): for x in range(min(columns, self.columns)): new[y][x] = old[y][x] return new self._main = rebuild(self._main) self._alt = rebuild(self._alt) self.lines, self.columns = lines, columns self.buffer = self._alt if self._in_alt else self._main self._top, self._bottom = 0, lines - 1 self.cursor.x = min(self.cursor.x, columns - 1) self.cursor.y = min(self.cursor.y, lines - 1) self._wrap = False # -- the byte/char state machine ----------------------------------------- def feed(self, data): for ch in self._dec.decode(data): st = self._state if st == "ground": self._ground(ch) elif st == "esc": self._esc(ch) elif st == "csi": self._csi(ch) elif st == "osc": self._osc_consume(ch) else: # 'charset' -> swallow one self._state = "ground" def _ground(self, ch): o = ord(ch) if o == 0x1b: self._state = "esc" elif o == 0x0d: self.cursor.x = 0 self._wrap = False elif o in (0x0a, 0x0b, 0x0c): self._linefeed() elif o == 0x08: if self.cursor.x > 0: self.cursor.x -= 1 self._wrap = False elif o == 0x09: self.cursor.x = min((self.cursor.x // 8 + 1) * 8, self.columns - 1) elif o >= 0x20: self._putc(ch) # other C0 controls (incl. BEL) ignored def _putc(self, ch): if self._wrap: self.cursor.x = 0 self._linefeed() self._wrap = False x, y = self.cursor.x, self.cursor.y cell = self.buffer[y][x] cell.data = ch cell.fg, cell.bg, cell.bold = self.fg, self.bg, self.bold cell.italics, cell.underscore, cell.reverse = self.italics, self.underscore, self.reverse if x >= self.columns - 1: self._wrap = True # defer the wrap (DEC autowrap) else: self.cursor.x = x + 1 def _linefeed(self): if self.cursor.y == self._bottom: self._scroll_up(1) elif self.cursor.y < self.lines - 1: self.cursor.y += 1 def _scroll_up(self, n): for _ in range(n): if not self._in_alt: # record the line scrolling away line = self.buffer[self._top] self.scrollback.append( (time.time(), "".join(c.data for c in line).rstrip())) if len(self.scrollback) > SCROLLBACK_MAX: del self.scrollback[:len(self.scrollback) - SCROLLBACK_MAX] self.buffer.pop(self._top) self.buffer.insert(self._bottom, self._blank_line()) def _scroll_down(self, n): for _ in range(n): self.buffer.pop(self._bottom) self.buffer.insert(self._top, self._blank_line()) def _esc(self, ch): self._state = "ground" if ch == "[": self._params, self._priv, self._state = "", "", "csi" elif ch == "]": self._osc, self._osc_esc, self._state = "", False, "osc" elif ch in "()": self._state = "charset" elif ch == "M": # reverse index if self.cursor.y == self._top: self._scroll_down(1) elif self.cursor.y > 0: self.cursor.y -= 1 elif ch == "D": # index self._linefeed() elif ch == "E": # next line self.cursor.x = 0 self._linefeed() elif ch == "7": self._saved = (self.cursor.x, self.cursor.y) elif ch == "8": self.cursor.x, self.cursor.y = self._saved elif ch == "c": self._full_reset() # '=', '>', etc. ignored def _csi(self, ch): o = ord(ch) if ch == "?" and not self._params: self._priv = "?" elif 0x30 <= o <= 0x3f: # parameter bytes self._params += ch elif 0x20 <= o <= 0x2f: # intermediate bytes pass else: # final byte self._dispatch_csi(ch) self._state = "ground" def _ints(self): res = [] for p in self._params.split(";"): try: res.append(int(p) if p else 0) except ValueError: res.append(0) return res def _dispatch_csi(self, ch): p = self._ints() n0 = p[0] if p else 0 cur = self.cursor if ch in "AB": cur.y = max(0, min(self.lines - 1, cur.y + (-(n0 or 1) if ch == "A" else (n0 or 1)))) self._wrap = False elif ch in "CD": cur.x = max(0, min(self.columns - 1, cur.x + ((n0 or 1) if ch == "C" else -(n0 or 1)))) self._wrap = False elif ch == "E": cur.y = min(self.lines - 1, cur.y + (n0 or 1)); cur.x = 0 elif ch == "F": cur.y = max(0, cur.y - (n0 or 1)); cur.x = 0 elif ch in "G`": cur.x = max(0, min(self.columns - 1, (n0 or 1) - 1)); self._wrap = False elif ch == "d": cur.y = max(0, min(self.lines - 1, (n0 or 1) - 1)) elif ch in "Hf": row = (p[0] if len(p) >= 1 and p[0] else 1) col = (p[1] if len(p) >= 2 and p[1] else 1) cur.y = max(0, min(self.lines - 1, row - 1)) cur.x = max(0, min(self.columns - 1, col - 1)) self._wrap = False elif ch == "J": self._erase_display(n0) elif ch == "K": self._erase_line(n0) elif ch == "m": self._sgr(p) elif ch == "r": top = (p[0] if len(p) >= 1 and p[0] else 1) - 1 bot = (p[1] if len(p) >= 2 and p[1] else self.lines) - 1 self._top, self._bottom = (top, bot) if 0 <= top < bot < self.lines else (0, self.lines - 1) cur.x, cur.y = 0, self._top elif ch == "L": self._insert_lines(n0 or 1) elif ch == "M": self._delete_lines(n0 or 1) elif ch == "P": self._delete_chars(n0 or 1) elif ch == "@": self._insert_chars(n0 or 1) elif ch == "X": self._erase_chars(n0 or 1) elif ch == "S": self._scroll_up(n0 or 1) elif ch == "T": self._scroll_down(n0 or 1) elif ch in "hl" and self._priv == "?": self._dec_private(p, ch == "h") # -- erase / edit -------------------------------------------------------- def _erase_in_line(self, y, x0, x1): row = self.buffer[y] for x in range(max(0, x0), min(self.columns - 1, x1) + 1): row[x] = Cell() def _erase_display(self, mode): if mode == 0: self._erase_in_line(self.cursor.y, self.cursor.x, self.columns - 1) for y in range(self.cursor.y + 1, self.lines): self.buffer[y] = self._blank_line() elif mode == 1: for y in range(0, self.cursor.y): self.buffer[y] = self._blank_line() self._erase_in_line(self.cursor.y, 0, self.cursor.x) else: for y in range(self.lines): self.buffer[y] = self._blank_line() def _erase_line(self, mode): if mode == 0: self._erase_in_line(self.cursor.y, self.cursor.x, self.columns - 1) elif mode == 1: self._erase_in_line(self.cursor.y, 0, self.cursor.x) else: self.buffer[self.cursor.y] = self._blank_line() def _erase_chars(self, n): self._erase_in_line(self.cursor.y, self.cursor.x, self.cursor.x + n - 1) def _insert_lines(self, n): if self._top <= self.cursor.y <= self._bottom: for _ in range(n): self.buffer.pop(self._bottom) self.buffer.insert(self.cursor.y, self._blank_line()) def _delete_lines(self, n): if self._top <= self.cursor.y <= self._bottom: for _ in range(n): self.buffer.pop(self.cursor.y) self.buffer.insert(self._bottom, self._blank_line()) def _delete_chars(self, n): row = self.buffer[self.cursor.y] for _ in range(n): if self.cursor.x < len(row): row.pop(self.cursor.x) row.append(Cell()) def _insert_chars(self, n): row = self.buffer[self.cursor.y] for _ in range(n): row.insert(self.cursor.x, Cell()) if len(row) > self.columns: row.pop() # -- SGR, private modes, OSC, reset -------------------------------------- def _sgr(self, p): if not p: p = [0] i = 0 while i < len(p): v = p[i] if v == 0: self._reset_attrs() elif v == 1: self.bold = True elif v == 22: self.bold = False elif v == 3: self.italics = True elif v == 23: self.italics = False elif v == 4: self.underscore = True elif v == 24: self.underscore = False elif v == 7: self.reverse = True elif v == 27: self.reverse = False elif 30 <= v <= 37: self.fg = COLOR_NAMES[v - 30] elif v == 39: self.fg = "default" elif 40 <= v <= 47: self.bg = COLOR_NAMES[v - 40] elif v == 49: self.bg = "default" elif 90 <= v <= 97: self.fg = xterm_hex(8 + v - 90) elif 100 <= v <= 107: self.bg = xterm_hex(8 + v - 100) elif v in (38, 48): tgt = "fg" if v == 38 else "bg" if i + 1 < len(p) and p[i + 1] == 5 and i + 2 < len(p): setattr(self, tgt, xterm_hex(p[i + 2])) i += 2 elif i + 1 < len(p) and p[i + 1] == 2 and i + 4 < len(p): setattr(self, tgt, "%02x%02x%02x" % (p[i + 2], p[i + 3], p[i + 4])) i += 4 i += 1 def _dec_private(self, params, on): for v in params: if v == 25: self.cursor.hidden = not on elif v in (1049, 1047, 47): self._set_alt(on, save=(v == 1049)) def _set_alt(self, on, save): if on and not self._in_alt: if save: self._saved = (self.cursor.x, self.cursor.y) self._alt = self._blank_grid() self.buffer, self._in_alt = self._alt, True self.cursor.x = self.cursor.y = 0 elif not on and self._in_alt: self.buffer, self._in_alt = self._main, False if save: self.cursor.x, self.cursor.y = self._saved def _osc_consume(self, ch): if ch == "\x07": self._finish_osc() elif ch == "\x1b": self._osc_esc = True elif self._osc_esc: self._osc_esc = False if ch == "\\": self._finish_osc() else: self._osc += ch def _finish_osc(self): self._state, self._osc_esc = "ground", False s, self._osc = self._osc, "" ps, _, pt = s.partition(";") if ps in ("0", "2"): self.title = pt def _full_reset(self): self._reset_attrs() self.cursor.x = self.cursor.y = 0 self.cursor.hidden = False self._top, self._bottom = 0, self.lines - 1 self._in_alt = False self._main = self._blank_grid() self._alt = self._blank_grid() self.buffer = self._main # --- a single child terminal ------------------------------------------------ class Window: _next_id = 1 def __init__(self, x, y, w, h): self.x, self.y, self.w, self.h = x, y, w, h self.id = Window._next_id Window._next_id += 1 self.default_title = "sh" self.maximized = False self.saved = None # geometry stashed while maximized iw, ih = self.inner() self.screen = VT(iw, ih) self.pid, self.fd = pty.fork() if self.pid == 0: # ---- child ---- os.environ["TERM"] = "xterm-256color" shell = os.environ.get("SHELL", "/bin/sh") os.execvp(shell, [shell]) os._exit(127) # ---- parent ---- flags = fcntl.fcntl(self.fd, fcntl.F_GETFL) fcntl.fcntl(self.fd, fcntl.F_SETFL, flags | os.O_NONBLOCK) self.set_winsize() def inner(self): return max(1, self.w - 2), max(1, self.h - 2) def set_winsize(self): iw, ih = self.inner() fcntl.ioctl(self.fd, termios.TIOCSWINSZ, struct.pack("HHHH", ih, iw, 0, 0)) def resize(self, w, h): self.w, self.h = w, h iw, ih = self.inner() self.screen.resize(ih, iw) self.set_winsize() def resize_inner(self, iw, ih): """Resize the emulated screen/pty directly (used by full-screen view).""" iw, ih = max(1, iw), max(1, ih) self.screen.resize(ih, iw) fcntl.ioctl(self.fd, termios.TIOCSWINSZ, struct.pack("HHHH", ih, iw, 0, 0)) def feed(self, data): self.screen.feed(data) @property def title(self): return f"{self.id}: {self.screen.title or self.default_title}" def close(self): try: os.close(self.fd) except OSError: pass try: os.kill(self.pid, signal.SIGHUP) except ProcessLookupError: pass try: os.waitpid(self.pid, 0) except ChildProcessError: pass # --- the compositor: an in-memory cell grid flushed with per-row diffing ---- class Compositor: def __init__(self, cols, rows, write=None): self.write = write or out self.resize(cols, rows) def resize(self, cols, rows): self.cols, self.rows = cols, rows self.prev = [None] * rows def begin(self): self.chars = [[" "] * self.cols for _ in range(self.rows)] self.attrs = [["0"] * self.cols for _ in range(self.rows)] def put(self, x, y, ch, attr="0"): if 1 <= x <= self.cols and 1 <= y <= self.rows: self.chars[y - 1][x - 1] = ch or " " self.attrs[y - 1][x - 1] = attr def text(self, x, y, s, attr="0"): for i, ch in enumerate(s): self.put(x + i, y, ch, attr) def flush(self): buf = [] for r in range(self.rows): cur = None parts = [f"\033[{r + 1};1H"] ca, cc = self.attrs[r], self.chars[r] for c in range(self.cols): a = ca[c] if a != cur: parts.append(f"\033[{a}m") cur = a parts.append(cc[c]) parts.append("\033[0m") line = "".join(parts) if line != self.prev[r]: buf.append(line) self.prev[r] = line if buf: self.write("".join(buf)) # --- the window manager ----------------------------------------------------- class WM: def __init__(self, cols=80, rows=24, write=None): self.write = write or out self.cols, self.rows = cols, rows self.comp = Compositor(cols, rows, self.write) self.windows = [] self.detach_requested = False self.quit_requested = False self.show_help = False self.drag = None # dict(mode, win, dx, dy) self.selection = None # dict(win, ax, ay, bx, by, moved) in interior coords self.flash = None # (message, expire_time) shown on the menu bar self._last_click = None # (time, col, row) for double-click detection self.fullscreen = False # stacked single/double-column view self.fullcols = 1 # 1 or 2 columns (2 only if width > TWO_COL_MIN) self.fs_focus_col = 0 # which column holds the focused window self.fs_other = None # window in the non-focused column (2-col) self.fs_content = None # (ox, oy, w, h) of the focused content area self.fs_cells = [] # [(win, x, y, w, h)] content panes for the mouse self.timesearch = None # dict(win, lines, pos) when browsing history self._ts_esc_time = None # idle timer to disambiguate a lone ESC self.prefix = False self.inbuf = b"" self.menu = [] # list of (c0, c1, action) self.spawn() # -- window helpers ------------------------------------------------------ @property def focused(self): return self.windows[-1] if self.windows else None def spawn(self): w = min(max(MIN_W, self.cols * 2 // 3), self.cols - 2) h = min(max(MIN_H, self.rows * 2 // 3), self.rows - 3) n = len(self.windows) x = min(2 + n * 3, max(2, self.cols - w)) y = min(3 + n * 2, max(2, self.rows - h)) try: self.windows.append(Window(x, y, w, h)) except OSError: pass def raise_win(self, win): if self.windows and self.windows[-1] is win: return self.windows.remove(win) self.windows.append(win) def close_win(self, win): if win in self.windows: self.windows.remove(win) if win is self.fs_other: self.fs_other = None win.close() def tile(self): """Re-flow windows into a grid of >=80x24 cells, z-order kept. The grid only has as many cells as 80x24 windows actually fit on the screen; any extra windows are superposed onto existing cells (cycling through them) rather than being shrunk below 80x24. """ n = len(self.windows) if not n: return avail_h = self.rows - 1 # rows 2..self.rows (row 1 = menu) ncols = max(1, self.cols // 80) nrows = max(1, avail_h // 24) capacity = ncols * nrows cell_w = self.cols // ncols # >=80 unless the screen is narrower cell_h = avail_h // nrows # >=24 unless the screen is shorter cells = [] # geometry of each grid cell for r in range(nrows): for c in range(ncols): x = 1 + c * cell_w y = 2 + r * cell_h w = (self.cols - c * cell_w) if c == ncols - 1 else cell_w h = (avail_h - r * cell_h) if r == nrows - 1 else cell_h cells.append((x, y, max(MIN_W, w), max(MIN_H, h))) for i, win in enumerate(self.windows): # row-major, wrapping onto cells x, y, w, h = cells[i % capacity] win.x, win.y = x, y win.resize(w, h) win.maximized = False over = n - capacity msg = f"grid {nrows}x{ncols}" + (f", {over} superposed" if over > 0 else "") self.flash = (msg, time.monotonic() + 1.5) def toggle_max(self, win): """Maximize the window to the full work area, or restore its geometry.""" if win.maximized: x, y, w, h = win.saved win.x, win.y = x, y win.resize(w, h) win.maximized = False self.flash = ("restored", time.monotonic() + 1.0) else: win.saved = (win.x, win.y, win.w, win.h) win.x, win.y = 1, 2 win.resize(self.cols, self.rows - 1) win.maximized = True self.flash = ("maximized", time.monotonic() + 1.0) def toggle_full(self): """Cycle floating -> full -> full x2 (if wide enough) -> floating.""" if not self.fullscreen: self.fullscreen, self.fullcols, self.fs_focus_col = True, 1, 0 self.fs_other = None msg = "full-screen" elif self.fullcols == 1 and self.cols > TWO_COL_MIN: self.fullcols, self.fs_focus_col = 2, 0 others = [w for w in reversed(self.windows) if w is not self.focused] self.fs_other = others[0] if others else None msg = "full-screen x2" else: self.fullscreen, self.fs_other = False, None for win in self.windows: # restore each window's framed size win.resize(win.w, win.h) msg = "floating" self.selection = self.drag = None self.write("\033[2J") self.comp.prev = [None] * self.rows # force a full repaint self.flash = (msg, time.monotonic() + 1.2) def _ordered(self): return sorted(self.windows, key=lambda w: w.id) def circulate(self): """Load the next window (by id) into the focused column.""" if not self.windows: return if self.fullscreen: cands = self._ordered() if self.fullcols == 2 and self.fs_other is not None: cands = [w for w in cands if w is not self.fs_other] if not cands: return i = cands.index(self.focused) if self.focused in cands else -1 self.raise_win(cands[(i + 1) % len(cands)]) else: self.windows.insert(0, self.windows.pop()) def move_column(self, target): """Move focus to the given full-screen column (0=left, 1=right).""" if self.fullcols != 2 or target == self.fs_focus_col or self.fs_other is None: return old = self.focused self.raise_win(self.fs_other) self.fs_other = old self.fs_focus_col = target def _fs_col_at(self, col): if self.fullcols != 2: return None leftw = (self.cols - 1) // 2 if col <= leftw: return 0 if col >= leftw + 2: return 1 return None # on the divider def fs_click_bar(self, win, tc): """Show win in full-screen column tc, making that column focused.""" if self.fullcols != 2 or tc is None: self.raise_win(win) return cur = [None, None] cur[self.fs_focus_col] = self.focused cur[1 - self.fs_focus_col] = self.fs_other if win is cur[1 - tc]: # already in the other col self.fs_focus_col = 1 - tc # -> just focus that column self.fs_other = cur[tc] else: self.fs_focus_col = tc self.fs_other = cur[1 - tc] self.raise_win(win) def switch_last(self): """Switch to the previous window (or the other column in full x2).""" if len(self.windows) <= 1: return if self.fullscreen and self.fullcols == 2 and self.fs_other is not None: self.move_column(1 - self.fs_focus_col) else: self.raise_win(self.windows[-2]) self.flash = ("switched", time.monotonic() + 1.0) # -- time search (history viewer) ---------------------------------------- def enter_timesearch(self): """Snapshot the focused window's logged history and open the viewer.""" win = self.focused if not win: return lines = list(win.screen.scrollback) # scrolled-off history now = time.time() for y in range(win.screen.lines): # plus what's on screen now text = "".join(c.data for c in win.screen.buffer[y]).rstrip() lines.append((now, text)) while lines and lines[-1][1] == "": # drop trailing blanks lines.pop() page = max(1, self.rows - 1) self.timesearch = {"win": win, "lines": lines, "pos": max(0, len(lines) - page)} self.selection = self.drag = None self._ts_esc_time = None self.flash = None self.write("\033[2J") self.comp.prev = [None] * self.rows def exit_timesearch(self): self.timesearch = None self._ts_esc_time = None self.write("\033[2J") self.comp.prev = [None] * self.rows def ts_move(self, kind, d): ts = self.timesearch page = max(1, self.rows - 1) step = d * (page if kind == "page" else 1) ts["pos"] = max(0, min(ts["pos"] + step, max(0, len(ts["lines"]) - page))) def ts_key(self, b): """Handle a keystroke while in the history viewer.""" nav = ((b"\x1b[A", ("move", -1)), (b"\x1bOA", ("move", -1)), (b"\x1b[B", ("move", 1)), (b"\x1bOB", ("move", 1)), (b"\x1b[5~", ("page", -1)), (b"\x1b[6~", ("page", 1))) for seq, (kind, d) in nav: if b[:len(seq)] == seq: self.ts_move(kind, d) return len(seq) if len(b) < len(seq) and seq[:len(b)] == b: return "wait" if b[:1] == b"\x1b": if len(b) == 1: return "wait" # lone ESC: resolved by idle timer self.exit_timesearch() # ESC + other -> leave the viewer return 1 return 1 # any other key is ignored here def window_at(self, col, row): for win in reversed(self.windows): if win.x <= col <= win.x + win.w - 1 and win.y <= row <= win.y + win.h - 1: return win return None # -- rendering ----------------------------------------------------------- def draw(self): self.comp.begin() self.fs_content = None self.draw_menu() if self.timesearch is not None: self.draw_timesearch() elif self.fullscreen and self.windows: self.draw_fullscreen() else: self.draw_background() for win in self.windows: self.draw_window(win, win is self.focused) if self.show_help: self.draw_help() self.comp.flush() self.place_cursor() def draw_menu(self): self.comp.text(1, 1, " " * self.cols, "0;7") self.menu = [] if self.timesearch is not None: # history-viewer status bar lines, pos = self.timesearch["lines"], self.timesearch["pos"] if lines: t = lines[min(pos, len(lines) - 1)][0] when = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(t)) else: when = "(empty log)" msg = f" TIME SEARCH {when} up/down PgUp/PgDn ESC=exit " self.comp.text(1, 1, msg[:self.cols], "0;7;1") self.menu = [(1, self.cols, "tsearch")] # clicking the bar exits return x = 2 if not self.fullscreen: full = "Full" elif self.fullcols == 1 and self.cols > TWO_COL_MIN: full = "Full2" else: full = "Float" buttons = [("New", "new"), ("Grid", "tile"), (full, "full"), ("Search", "tsearch"), ("Detach", "detach"), ("Help", "help"), ("Quit", "quit")] for i, (label, action) in enumerate(buttons): if i: self.comp.text(x, 1, " | ", "0;7") x += 3 self.comp.text(x, 1, label, "0;7;1") self.menu.append((x, x + len(label) - 1, action)) x += len(label) if self.flash: hint = self.flash[0] else: hint = "drag title=move border=resize [x]=close body=select" if x + len(hint) < self.cols: self.comp.text(self.cols - len(hint), 1, hint, "0;7") def draw_background(self): """Paint the cheesy 3D WM-TERM banner on the empty desktop.""" glyphs = [LOGO_FONT[c] for c in LOGO_WORD] base_w = sum(len(g[0]) for g in glyphs) + (len(glyphs) - 1) # +1 col gaps scale = 0 for sc in (4, 3, 2, 1): # biggest size that fits if base_w * sc + 2 <= self.cols and 5 * sc + 4 <= self.rows - 1: scale = sc break if not scale: return total_w, total_h = base_w * scale + 2, 5 * scale x0 = 1 + (self.cols - total_w) // 2 y0 = max(2, 2 + ((self.rows - 1) - (total_h + 2)) // 2) pixels, cx = [], x0 for gi, g in enumerate(glyphs): gw = len(g[0]) face = LOGO_COLORS[gi % len(LOGO_COLORS)] for ry in range(5): for rx in range(gw): if g[ry][rx] != " ": for dy in range(scale): for dx in range(scale): pixels.append((cx + rx * scale + dx, y0 + ry * scale + dy, face)) cx += (gw + 1) * scale for px, py, _ in pixels: # 3D extrusion (shaded blocks) self.comp.put(px + 2, py + 2, "▓", "0;90") for px, py, _ in pixels: self.comp.put(px + 1, py + 1, "▓", "0;90") for px, py, face in pixels: # bright solid faces on top self.comp.put(px, py, "█", face) def draw_window(self, win, focused): x, y, w, h = win.x, win.y, win.w, win.h bar = "0;7" if focused else "0" # reverse video marks the focused window edge = "0" # title bar: box top with the title and [x] overlaid self.comp.put(x, y, "┌", bar) for xx in range(x + 1, x + w - 1): self.comp.put(xx, y, "─", bar) self.comp.put(x + w - 1, y, "┐", bar) title = (" " + win.title + " ")[:max(0, w - 6)] self.comp.text(x + 2, y, title, bar) self.comp.text(x + w - 4, y, "[x]", bar) # sides for yy in range(y + 1, y + h - 1): self.comp.put(x, yy, "│", edge) self.comp.put(x + w - 1, yy, "│", edge) # bottom self.comp.put(x, y + h - 1, "└", edge) for xx in range(x + 1, x + w - 1): self.comp.put(xx, y + h - 1, "─", edge) self.comp.put(x + w - 1, y + h - 1, "◢", edge) # resize handle # interior content from the child's screen buffer iw, ih = win.inner() buf = win.screen.buffer sel = self.selection is not None and self.selection["win"] is win for yy in range(ih): row = buf[yy] for xx in range(iw): c = row[xx] attr = cell_sgr(c) if sel and self.sel_contains(xx, yy): attr += ";7" # reverse video over the selection self.comp.put(x + 1 + xx, y + 1 + yy, c.data, attr) def _fs_panes(self): """[(window, x_start, width)] content columns; 2-col leaves a divider gap.""" focused = self.focused if self.fullcols == 2 and self.cols > 3: leftw = (self.cols - 1) // 2 rightw = self.cols - leftw - 1 wins = [None, None] wins[self.fs_focus_col] = focused wins[1 - self.fs_focus_col] = self.fs_other return [(wins[0], 1, leftw), (wins[1], leftw + 2, rightw)] return [(focused, 1, self.cols)] def draw_fullscreen(self): """Each column shows the full title list (its own displayed window highlighted) over that window's content, split by a vertical divider.""" order = self._ordered() nbars = min(len(order), max(1, self.rows - 4)) oy = 2 + nbars ch = self.rows - oy + 1 self.fs_content = None self.fs_cells = [] panes = self._fs_panes() if self.fullcols == 2 and len(panes) == 2: div = (self.cols - 1) // 2 + 1 # divider, full height for r in range(2, self.rows + 1): self.comp.put(div, r, "│", "0") for ci, (win, xs, w) in enumerate(panes): self._draw_fs_column(ci, win, xs, w, order, nbars, oy, ch) def _draw_fs_column(self, ci, displayed, xs, w, order, nbars, oy, ch): for i in range(nbars): # title list (per column) win = order[i] bar = "0;7" if win is displayed else "0" # highlight this column's win self.comp.text(xs, 2 + i, "─" * w, bar) self.comp.text(xs + 1, 2 + i, f" {win.title} "[:max(0, w - 1)], bar) if displayed is None or ch < 1: return if displayed.screen.columns != w or displayed.screen.lines != ch: displayed.resize_inner(w, ch) buf = displayed.screen.buffer sel = self.selection is not None and self.selection["win"] is displayed for yy in range(min(ch, displayed.screen.lines)): cells = buf[yy] for xx in range(min(w, displayed.screen.columns)): c = cells[xx] attr = cell_sgr(c) if sel and self.sel_contains(xx, yy): attr += ";7" self.comp.put(xs + xx, oy + yy, c.data, attr) self.fs_cells.append((displayed, xs, oy, w, ch, ci)) if displayed is self.focused: self.fs_content = (xs, oy, w, ch) def draw_help(self): """Opaque centered box listing the controls (any key closes it).""" lines = HELP_LINES w = min(self.cols, max(len(s) for s in lines) + 4) h = min(self.rows, len(lines) + 2) x0 = max(1, 1 + (self.cols - w) // 2) y0 = max(1, 1 + (self.rows - h) // 2) self.comp.put(x0, y0, "┌") self.comp.put(x0 + w - 1, y0, "┐") for xx in range(x0 + 1, x0 + w - 1): self.comp.put(xx, y0, "─") self.comp.text(x0 + 2, y0, " termwm help ", "0;7") for i in range(1, h - 1): self.comp.put(x0, y0 + i, "│") for xx in range(x0 + 1, x0 + w - 1): self.comp.put(xx, y0 + i, " ") self.comp.put(x0 + w - 1, y0 + i, "│") if i - 1 < len(lines): self.comp.text(x0 + 2, y0 + i, lines[i - 1][:w - 4]) self.comp.put(x0, y0 + h - 1, "└") self.comp.put(x0 + w - 1, y0 + h - 1, "┘") for xx in range(x0 + 1, x0 + w - 1): self.comp.put(xx, y0 + h - 1, "─") def draw_timesearch(self): """Render the timestamped history viewer (the menu bar shows the date).""" ts = self.timesearch lines, pos = ts["lines"], ts["pos"] for i in range(self.rows - 1): row = 2 + i idx = pos + i if idx < len(lines): t, text = lines[idx] s = time.strftime("%H:%M:%S ", time.localtime(t)) + text attr = "0;7" if i == 0 else "0" # highlight the anchor line else: s, attr = "~", "0" if i == 0: s = s.ljust(self.cols) # full-width anchor bar self.comp.text(1, row, s[:self.cols], attr) # -- selection ----------------------------------------------------------- def _sel_range(self): """Return ordered (start_x, start_y, end_x, end_y) of the selection.""" s = self.selection a, b = (s["ay"], s["ax"]), (s["by"], s["bx"]) (sy, sx), (ey, ex) = (a, b) if a <= b else (b, a) return sx, sy, ex, ey def sel_contains(self, xx, yy): sx, sy, ex, ey = self._sel_range() if yy < sy or yy > ey: return False if sy == ey: return sx <= xx <= ex if yy == sy: return xx >= sx if yy == ey: return xx <= ex return True # a full middle line def selection_text(self): s = self.selection win = s["win"] buf = win.screen.buffer iw = win.screen.columns sx, sy, ex, ey = self._sel_range() lines = [] for yy in range(sy, ey + 1): x0 = sx if yy == sy else 0 x1 = ex if yy == ey else iw - 1 row = buf[yy] lines.append("".join(row[xx].data or " " for xx in range(x0, x1 + 1)).rstrip()) return "\n".join(lines) def select_word(self, win, ix, iy, ox, oy): """Expand to the whitespace-delimited word at screen-cell (ix, iy).""" iw = win.screen.columns if not (0 <= iy < win.screen.lines) or not (0 <= ix < iw): return False row = win.screen.buffer[iy] is_word = lambda x: bool(row[x].data) and not row[x].data.isspace() if not is_word(ix): return False l = r = ix while l > 0 and is_word(l - 1): l -= 1 while r < iw - 1 and is_word(r + 1): r += 1 self.selection = {"win": win, "ox": ox, "oy": oy, "ax": l, "ay": iy, "bx": r, "by": iy, "moved": True} self.copy_selection() return True def copy_selection(self): if not self.selection: return txt = self.selection_text() if not txt: self.selection = None return copied = False try: # system clipboard (macOS) subprocess.run(["pbcopy"], input=txt.encode(), check=True) copied = True except Exception: pass try: # OSC 52: also works over SSH / other OSes import base64 b64 = base64.b64encode(txt.encode()).decode() self.write(f"\033]52;c;{b64}\a") copied = True except Exception: pass self.flash = ("copied %d char%s" % (len(txt), "" if len(txt) == 1 else "s"), time.monotonic() + 1.5) if copied else None def place_cursor(self): if self.show_help or self.timesearch is not None: self.write(HIDE_CUR) return win = self.focused if not win: self.write(HIDE_CUR) return cur = win.screen.cursor if self.fullscreen and self.fs_content: ox, oy, cw, ch = self.fs_content if cur.hidden or not (0 <= cur.x < cw and 0 <= cur.y < ch): self.write(HIDE_CUR) return self.write(f"\033[{oy + cur.y};{ox + cur.x}H" + SHOW_CUR) return iw, ih = win.inner() if cur.hidden or not (0 <= cur.x < iw and 0 <= cur.y < ih): self.write(HIDE_CUR) return self.write(f"\033[{win.y + 1 + cur.y};{win.x + 1 + cur.x}H" + SHOW_CUR) # -- input --------------------------------------------------------------- def feed_stdin(self, data): self.inbuf += data while self.inbuf: b = self.inbuf if self.show_help: # any input closes help self.show_help = False self.write("\033[2J") self.comp.prev = [None] * self.rows self.inbuf = b"" break if self.timesearch is not None: # history viewer eats keys if b[:3] == b"\x1b[<": # but mouse still drives menu i = 3 while i < len(b) and b[i:i + 1] not in (b"M", b"m"): i += 1 if i >= len(b): break self.mouse(b[3:i], b[i:i + 1]) self.inbuf = b[i + 1:] continue r = self.ts_key(b) if r == "wait": break self.inbuf = b[r:] continue if self.prefix: self.prefix = False self.command(b[:1]) self.inbuf = b[1:] continue if b[:1] == PREFIX: self.prefix = True self.inbuf = b[1:] continue if b[:1] == b"\x00" and self.fullscreen: # Ctrl-Space: circulate self.circulate() self.inbuf = b[1:] continue if self.fullscreen and self.fullcols == 2: # Ctrl-Left/Right: columns n, target = self._match_keymap(b, FS_COLKEYS) if n == "wait": break if n: self.move_column(target) self.inbuf = b[n:] continue if b[:3] == b"\x1b[<": # SGR mouse report i = 3 while i < len(b) and b[i:i + 1] not in (b"M", b"m"): i += 1 if i >= len(b): break # incomplete; wait self.mouse(b[3:i], b[i:i + 1]) self.inbuf = b[i + 1:] continue sw = self._match_switch(b) # Ctrl-Enter -> switch_last if sw == "wait": break if sw: self.switch_last() self.inbuf = b[sw:] continue if b in (b"\x1b", b"\x1b["): # possible partial mouse break # forward a run of ordinary bytes to the focused shell j = 1 while j < len(b) and b[j:j + 1] not in (b"\x1b", PREFIX): j += 1 if self.focused: try: os.write(self.focused.fd, b[:j]) except OSError: pass self.inbuf = b[j:] def _match_keymap(self, b, keymap): """Match b against {seq: value}; return (consumed_len|'wait'|0, value).""" for seq, val in keymap.items(): if b[:len(seq)] == seq: return len(seq), val if len(b) < len(seq) and seq[:len(b)] == b: return "wait", None return 0, None def _match_switch(self, b): """Return consumed length if b starts with a Ctrl-Enter sequence, 'wait' if b is a partial prefix of one, else 0.""" for seq in KEY_SWITCH: if b[:len(seq)] == seq: return len(seq) if len(b) < len(seq) and seq[:len(b)] == b: return "wait" return 0 def command(self, c): if c in (b"n", b"N"): self.spawn() elif c in (b"q", b"Q"): self.quit_requested = True elif c in (b"w", b"W"): if self.focused: self.close_win(self.focused) elif c == b"\t": if len(self.windows) > 1: self.windows.insert(0, self.windows.pop()) elif c in (b"t", b"T"): self.tile() elif c in (b"m", b"M"): if self.focused: self.toggle_max(self.focused) elif c in (b"f", b"F"): self.toggle_full() elif c in (b"d", b"D"): # detach (daemon keeps running) self.detach_requested = True elif c in (b"h", b"H", b"?"): # show help overlay self.show_help = True elif c in (b"\r", b"\n"): # Ctrl-G Enter: switch_last self.switch_last() elif c == b"g": # send a literal Ctrl-G if self.focused: try: os.write(self.focused.fd, PREFIX) except OSError: pass def mouse(self, params, final): try: btn, col, row = (int(v) for v in params.split(b";")) except ValueError: return pressed = final == b"M" base = btn & 0b11 if not pressed: # button release if self.drag and self.drag.get("mode") == "select": if self.drag.get("moved"): self.copy_selection() else: self.selection = None # a plain click clears it self.drag = None return if self.drag: # motion while dragging self.update_drag(col, row) return if base != 0: # only left button acts return self.selection = None # any fresh press clears now = time.monotonic() # double-click detection dbl = (self._last_click is not None and now - self._last_click[0] < 0.4 and abs(col - self._last_click[1]) <= 1 and abs(row - self._last_click[2]) <= 1) self._last_click = (now, col, row) if row == 1: # menu bar for c0, c1, action in self.menu: if c0 <= col <= c1: if action == "new": self.spawn() elif action == "tile": self.tile() elif action == "full": self.toggle_full() elif action == "tsearch": if self.timesearch is None: self.enter_timesearch() else: self.exit_timesearch() elif action == "detach": self.detach_requested = True elif action == "help": self.show_help = True elif action == "quit": self.quit_requested = True return return if self.timesearch is not None: # ignore clicks in the log return if self.fullscreen: # stacked view order = self._ordered() nbars = min(len(order), max(1, self.rows - 4)) if 2 <= row <= 1 + nbars: # click a title -> show it win = order[row - 2] if self.fullcols == 2: self.fs_click_bar(win, self._fs_col_at(col)) else: self.raise_win(win) return for win, ox, oy, cw, ch, ci in self.fs_cells: # click content -> select if ox <= col <= ox + cw - 1 and oy <= row <= oy + ch - 1: if self.fullcols == 2: self.fs_click_bar(win, ci) else: self.raise_win(win) ix, iy = col - ox, row - oy if dbl and self.select_word(win, ix, iy, ox, oy): self._last_click = None return self.selection = {"win": win, "ox": ox, "oy": oy, "ax": ix, "ay": iy, "bx": ix, "by": iy, "moved": False} self.drag = {"mode": "select", "win": win} return return win = self.window_at(col, row) if not win: return self.raise_win(win) # close box on the title bar if row == win.y and win.x + win.w - 4 <= col <= win.x + win.w - 2: self.close_win(win) return # title bar -> double-click maximizes, single-click-drag moves if row == win.y: if dbl: self._last_click = None self.toggle_max(win) else: self.drag = {"mode": "move", "win": win, "dx": col - win.x, "dy": row - win.y} return # any border / corner -> resize (top row is reserved for move, above) edges = set() if col == win.x: edges.add("l") if col == win.x + win.w - 1: edges.add("r") if row == win.y + win.h - 1: edges.add("b") if edges: self.drag = {"mode": "resize", "win": win, "edges": edges, "right": win.x + win.w - 1} return # interior body if (win.x + 1 <= col <= win.x + win.w - 2 and win.y + 1 <= row <= win.y + win.h - 2): ox, oy = win.x + 1, win.y + 1 ix, iy = col - ox, row - oy if dbl and self.select_word(win, ix, iy, ox, oy): # double-click a word self._last_click = None return # otherwise start a text selection anchored to this window self.selection = {"win": win, "ox": ox, "oy": oy, "ax": ix, "ay": iy, "bx": ix, "by": iy, "moved": False} self.drag = {"mode": "select", "win": win} return # otherwise just focus (already raised) def update_drag(self, col, row): d = self.drag win = d["win"] if d["mode"] == "select": # confine the moving end of the selection to its origin rectangle s = self.selection iw, ih = win.screen.columns, win.screen.lines cx = min(max(col, s["ox"]), s["ox"] + iw - 1) - s["ox"] cy = min(max(row, s["oy"]), s["oy"] + ih - 1) - s["oy"] cx = min(max(cx, 0), iw - 1) cy = min(max(cy, 0), ih - 1) if (cx, cy) != (s["bx"], s["by"]): s["bx"], s["by"] = cx, cy self.drag["moved"] = True return if d["mode"] == "move": nx = min(max(1, col - d["dx"]), self.cols - win.w + 1) ny = min(max(2, row - d["dy"]), self.rows - win.h + 1) win.x, win.y = nx, ny else: # resize one or more edges edges = d["edges"] nx, ny, nw, nh = win.x, win.y, win.w, win.h if "r" in edges: # right edge: left stays put nw = min(max(MIN_W, col - win.x + 1), self.cols - win.x + 1) if "l" in edges: # left edge: right stays put right = d["right"] nx = max(1, min(col, right - MIN_W + 1)) nw = right - nx + 1 if "b" in edges: # bottom edge: top stays put nh = min(max(MIN_H, row - win.y + 1), self.rows - win.y + 1) if (nx, ny) != (win.x, win.y): win.x, win.y = nx, ny if (nw, nh) != (win.w, win.h): win.resize(nw, nh) win.maximized = False # -- terminal resize (size comes from the attached client) --------------- def resize_terminal(self, cols, rows): cols, rows = max(2, cols), max(2, rows) self.cols, self.rows = cols, rows if self.fullcols == 2 and cols <= TWO_COL_MIN: # too narrow for two columns self.fullcols, self.fs_other = 1, None self.comp.resize(cols, rows) self.write("\033[2J") for win in self.windows: if win.maximized: win.x, win.y = 1, 2 win.resize(cols, rows - 1) continue win.x = min(win.x, max(1, cols - win.w + 1)) win.y = min(max(2, win.y), max(2, rows - win.h + 1)) nw = min(win.w, cols - win.x + 1) nh = min(win.h, rows - win.y + 1) if (nw, nh) != (win.w, win.h): win.resize(max(MIN_W, nw), max(MIN_H, nh)) def tick(self): """Periodic upkeep (idle-ESC and flash expiry); returns True if redraw.""" dirty = False if self.timesearch is not None and self.inbuf == b"\x1b": if self._ts_esc_time is None: # lone ESC: wait briefly, self._ts_esc_time = time.monotonic() # then treat as exit elif time.monotonic() - self._ts_esc_time > 0.1: self.inbuf = b"" self.exit_timesearch() dirty = True elif self._ts_esc_time is not None: self._ts_esc_time = None if self.flash and time.monotonic() > self.flash[1]: self.flash = None dirty = True return dirty # --- client/server transport ------------------------------------------------ # # A length-prefixed message framing over a Unix socket. One byte type, a # 4-byte big-endian length, then the payload. # client -> server: 'i' input bytes 'r' resize (HH cols,rows) 'k' kill # server -> client: 'o' output bytes (terminal escape stream) def _frame(typ, payload=b""): return typ + struct.pack(">I", len(payload)) + payload def _unframe(buf): """Yield (type, payload) messages from buf; return the leftover bytes.""" msgs = [] while len(buf) >= 5: ln = struct.unpack(">I", buf[1:5])[0] if len(buf) < 5 + ln: break msgs.append((buf[0:1], buf[5:5 + ln])) buf = buf[5 + ln:] return msgs, buf def default_sock(): return os.path.join("/tmp", f"termwm-{os.getuid()}.sock") def server_alive(sockpath): if not os.path.exists(sockpath): return False s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) try: s.connect(sockpath) s.close() return True except OSError: return False class Server: """The daemon: owns the shells and the WM, serves one attached client.""" def __init__(self, lsock, sockpath): self.lsock = lsock self.sockpath = sockpath self.client = None self.cbuf = b"" self.wm = WM(80, 24, self._write) def _write(self, s): if self.client is not None: data = s.encode("utf-8", "replace") try: self.client.sendall(_frame(b"o", data)) except OSError: self.drop_client() def accept(self): try: c, _ = self.lsock.accept() except OSError: return if self.client is not None: # a new client steals the session try: self.client.close() except OSError: pass self.client = c self.cbuf = b"" self.wm.comp.prev = [None] * self.wm.rows # force a full repaint on attach def drop_client(self): if self.client is not None: try: self.client.close() except OSError: pass self.client = None self.cbuf = b"" self.wm.detach_requested = False def pump_client(self): try: data = self.client.recv(65536) except OSError: data = b"" if not data: # client/terminal went away self.drop_client() return False self.cbuf += data msgs, self.cbuf = _unframe(self.cbuf) dirty = False for typ, payload in msgs: if typ == b"i": self.wm.feed_stdin(payload) dirty = True elif typ == b"r" and len(payload) == 4: cols, rows = struct.unpack(">HH", payload) self.wm.resize_terminal(cols, rows) dirty = True elif typ == b"k": self.wm.quit_requested = True return dirty def run(self): while True: if self.wm.quit_requested or not self.wm.windows: break fds = [self.lsock] + ([self.client] if self.client else []) \ + [w.fd for w in self.wm.windows] try: ready, _, _ = select.select(fds, [], [], 0.05) except (InterruptedError, OSError): continue dirty = False if self.lsock in ready: self.accept() if self.client is not None and self.client in ready: dirty |= self.pump_client() for win in list(self.wm.windows): if win.fd in ready: try: d = os.read(win.fd, 65536) except OSError: d = b"" if d: win.feed(d) dirty = True else: self.wm.close_win(win) dirty = True if self.wm.detach_requested: self.drop_client() dirty |= self.wm.tick() if self.client is not None and (dirty or self.wm.drag): self.wm.draw() for win in self.wm.windows: win.close() try: self.lsock.close() finally: try: os.unlink(self.sockpath) except OSError: pass def start_server(sockpath): """Bind the socket, then double-fork a detached daemon that serves it.""" lsock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) try: os.unlink(sockpath) except FileNotFoundError: pass lsock.bind(sockpath) lsock.listen(1) pid = os.fork() if pid > 0: # original process -> client os.waitpid(pid, 0) lsock.close() return os.setsid() # detach from the terminal if os.fork() > 0: os._exit(0) signal.signal(signal.SIGHUP, signal.SIG_IGN) devnull = os.open(os.devnull, os.O_RDWR) for fd in (0, 1, 2): os.dup2(devnull, fd) try: Server(lsock, sockpath).run() finally: os._exit(0) def client(sockpath): s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) for _ in range(100): # wait for the daemon to listen try: s.connect(sockpath) break except OSError: time.sleep(0.03) else: sys.exit("termwm: could not connect to the server") fd = sys.stdin.fileno() old = termios.tcgetattr(fd) def send(typ, payload=b""): try: s.sendall(_frame(typ, payload)) except OSError: raise SystemExit try: tty.setraw(fd) out(ALT_ON + HIDE_CUR + MOUSE_ON + "\033[2J") sz = os.get_terminal_size() send(b"r", struct.pack(">HH", sz.columns, sz.lines)) signal.signal(signal.SIGWINCH, lambda *_: _send_size(s)) buf = b"" while True: try: ready, _, _ = select.select([0, s], [], []) except (InterruptedError, OSError): continue if 0 in ready: data = os.read(0, 65536) if not data: break send(b"i", data) if s in ready: data = s.recv(65536) if not data: # detached or server gone break buf += data msgs, buf = _unframe(buf) for typ, payload in msgs: if typ == b"o": os.write(1, payload) except SystemExit: pass finally: out(MOUSE_OFF + SHOW_CUR + ALT_OFF) termios.tcsetattr(fd, termios.TCSADRAIN, old) try: s.close() except OSError: pass def _send_size(s): try: sz = os.get_terminal_size() s.sendall(_frame(b"r", struct.pack(">HH", sz.columns, sz.lines))) except OSError: pass def main(): args = sys.argv[1:] sockpath = default_sock() if "--kill" in args: if not server_alive(sockpath): sys.exit("termwm: no running server") s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) s.connect(sockpath) s.sendall(_frame(b"k")) s.close() return if "--attach" in args or "-attach" in args: if not server_alive(sockpath): sys.exit("termwm: no running server to attach to") client(sockpath) return if not server_alive(sockpath): start_server(sockpath) client(sockpath) if __name__ == "__main__": main()