|
#!/usr/bin/env python3 |
|
"""memtop — a macOS memory TUI. |
|
|
|
Visual language is borrowed from btop: gradient meters that run green→red |
|
across their length, a stacked composition bar, scrolling history graphs, and |
|
values tinted by magnitude so the shape of the problem reads at a glance. |
|
""" |
|
import curses |
|
import subprocess |
|
import time |
|
import os |
|
import sys |
|
from collections import defaultdict, deque |
|
from dataclasses import dataclass |
|
from typing import List, Dict, Optional, Sequence |
|
|
|
APP_NAME = "memtop" |
|
|
|
# Braille dot bits: [column][row from top]. Two sample columns and four |
|
# vertical steps pack into every cell. |
|
BRAILLE_DOTS = ((0x01, 0x02, 0x04, 0x40), (0x08, 0x10, 0x20, 0x80)) |
|
DEFAULT_INTERVAL = 5.0 |
|
MIN_INTERVAL = 0.5 |
|
MAX_INTERVAL = 60.0 |
|
HISTORY_LEN = 600 |
|
|
|
|
|
|
|
def set_cursor_visibility(visibility: int) -> None: |
|
"""Change cursor visibility when the terminal supports it. |
|
|
|
Some macOS Terminal/Python combinations return ERR for curs_set even |
|
though the rest of curses works normally. Cursor visibility is cosmetic, |
|
so it must never prevent memtop from starting. |
|
""" |
|
try: |
|
curses.curs_set(visibility) |
|
except curses.error: |
|
pass |
|
|
|
|
|
@dataclass |
|
class Proc: |
|
pid: int |
|
ppid: int |
|
rss_kb: int |
|
vsz_kb: int |
|
cpu: float |
|
user: str |
|
command: str |
|
|
|
|
|
|
|
def fmt_bytes(kb: float) -> str: |
|
b = kb * 1024.0 |
|
units = ["B", "K", "M", "G", "T"] |
|
i = 0 |
|
while abs(b) >= 1024 and i < len(units) - 1: |
|
b /= 1024.0 |
|
i += 1 |
|
if i == 0: |
|
return f"{b:.0f}{units[i]}" |
|
if b >= 100: |
|
return f"{b:.0f}{units[i]}" |
|
if b >= 10: |
|
return f"{b:.1f}{units[i]}" |
|
return f"{b:.2f}{units[i]}" |
|
|
|
|
|
# ---------------------------------------------------------------- theme |
|
|
|
|
|
class Theme: |
|
"""Palette drawn from the terminal's own theme. |
|
|
|
By default every colour is an ANSI index 0-15, which the terminal resolves |
|
through ITS OWN PALETTE - so memtop wears whatever theme the terminal is |
|
wearing, and follows it when it changes, with nothing to sync. Index 8 |
|
("bright black") is the theme's grey and carries all the chrome; 7 and 15 |
|
are its normal and bright foreground. Backgrounds stay -1, the terminal's |
|
own, so the panels sit on the theme's background rather than on black. |
|
|
|
The alternative, MEMTOP_COLORS=256, picks fixed values out of the 256-colour |
|
cube: more shades for the ramp, but the same look in every theme. Two rules |
|
hold either way. Colour is spent on values, not on chrome, so the eye lands |
|
on numbers. And the ramp runs dark->bright, so a low meter recedes. |
|
|
|
Set MEMTOP_COLORS=none for monochrome. Everything below survives COLORS |
|
being small or start_color failing outright. |
|
""" |
|
|
|
# Desaturated green→olive→rust→red. The low end is deliberately dark. |
|
RAMP_256 = [22, 65, 71, 107, 143, 179, 173, 167, 203, 196, 196] |
|
# The same green→yellow→red arc built only from theme colours. Depth comes |
|
# from normal vs bright (2→10, 3→11, 1→9) rather than from extra hues, so |
|
# the ramp still climbs in a theme that has redefined all six. |
|
RAMP_16 = [2, 2, 2, 10, 10, 3, 3, 11, 1, 9, 9] |
|
RAMP_8 = [curses.COLOR_GREEN] * 4 + [curses.COLOR_YELLOW] * 4 + [curses.COLOR_RED] * 3 |
|
|
|
# Greys and accents. Chrome is grey; segments are low-saturation hues. |
|
ROLES_256 = { |
|
"border": 236, "title": 245, "titlenum": 131, "label": 242, |
|
"unit": 240, "value": 250, "bright": 253, "track": 236, |
|
"accent": 66, "selbg": 237, |
|
"wired": 95, "active": 66, "compressed": 137, |
|
"inactive": 60, "speculative": 239, "free": 235, |
|
"swap": 96, |
|
} |
|
# Theme colours. 8 is the theme's grey and does all the chrome, 7 and 15 |
|
# its normal and bright foreground; nothing here is a literal RGB value. |
|
ROLES_16 = { |
|
"border": 8, "title": 7, "titlenum": 5, "label": 7, |
|
"unit": 8, "value": 7, "bright": 15, "track": 8, |
|
"accent": 6, "selbg": 8, |
|
"wired": 5, "active": 6, "compressed": 3, |
|
"inactive": 4, "speculative": 8, "free": 8, |
|
"swap": 5, |
|
} |
|
# With only 8 colours there is no grey to spend on chrome: index 0 is the |
|
# background itself, so painting borders and labels with it makes them |
|
# vanish. DEFAULT (-1, the terminal's own foreground) plus A_DIM is the |
|
# only way to push chrome back and still see it. |
|
DEFAULT = -1 |
|
ROLES_8 = { |
|
"border": DEFAULT, "title": curses.COLOR_WHITE, |
|
"titlenum": curses.COLOR_MAGENTA, "label": DEFAULT, |
|
"unit": DEFAULT, "value": curses.COLOR_WHITE, |
|
"bright": curses.COLOR_WHITE, "track": DEFAULT, |
|
"accent": curses.COLOR_CYAN, "selbg": curses.COLOR_BLUE, |
|
"wired": curses.COLOR_MAGENTA, "active": curses.COLOR_CYAN, |
|
"compressed": curses.COLOR_YELLOW, "inactive": curses.COLOR_BLUE, |
|
"speculative": DEFAULT, "free": DEFAULT, |
|
"swap": curses.COLOR_MAGENTA, |
|
} |
|
|
|
DIM_ROLES = frozenset({"border", "label", "unit", "track", |
|
"speculative", "free", "inactive"}) |
|
|
|
def __init__(self, enabled: bool = True): |
|
self.enabled = False |
|
self._pairs: Dict[tuple, int] = {} |
|
self._next = 1 |
|
self._max_pairs = 1 |
|
self.colors = 0 |
|
self.rich = False |
|
self.themed = False |
|
self.has_dim = False |
|
self.ramp: List[int] = [] |
|
self.roles: Dict[str, int] = {} |
|
mode = os.environ.get("MEMTOP_COLORS", "").strip().lower() |
|
if not enabled or mode == "none": |
|
return |
|
try: |
|
curses.start_color() |
|
curses.use_default_colors() |
|
except curses.error: |
|
return |
|
self.colors = getattr(curses, "COLORS", 0) |
|
if self.colors < 8: |
|
return |
|
self.enabled = True |
|
# Some terminfo entries have no `dim` and curses drops A_DIM silently. |
|
# Bold renders brighter rather than greyer, but it is the only other |
|
# lever, so it stands in where dim is missing. |
|
try: |
|
self.has_dim = bool(curses.tigetstr("dim")) |
|
except Exception: |
|
self.has_dim = False |
|
self._max_pairs = min(getattr(curses, "COLOR_PAIRS", 64), 256) |
|
# Theme colours by default, and they need indices 8-15 to exist: a |
|
# terminal claiming 8 has no grey, so it falls back to dim DEFAULT. |
|
if mode == "256" and self.colors >= 256: |
|
self.ramp, self.roles = self.RAMP_256, self.ROLES_256 |
|
self.rich = True |
|
elif self.colors >= 16: |
|
self.ramp, self.roles = self.RAMP_16, self.ROLES_16 |
|
self.themed = True |
|
else: |
|
self.ramp, self.roles = self.RAMP_8, self.ROLES_8 |
|
|
|
def pair(self, fg: int, bg: int = -1) -> int: |
|
"""Return a colour-pair attribute, allocating the pair on first use.""" |
|
if not self.enabled: |
|
return 0 |
|
key = (fg, bg) |
|
cached = self._pairs.get(key) |
|
if cached is not None: |
|
return curses.color_pair(cached) |
|
if self._next >= self._max_pairs: |
|
return 0 |
|
try: |
|
curses.init_pair(self._next, fg, bg) |
|
except curses.error: |
|
return 0 |
|
self._pairs[key] = self._next |
|
self._next += 1 |
|
return curses.color_pair(self._pairs[key]) |
|
|
|
def c(self, role: str, bold: bool = False, bg: str = None) -> int: |
|
"""Attribute for a named role, optionally over a named background.""" |
|
if not self.enabled: |
|
# Monochrome still needs hierarchy, so map roles onto dim/bold. |
|
if role in self.DIM_ROLES: |
|
return curses.A_DIM |
|
return curses.A_BOLD if (bold or role == "bright") else 0 |
|
bgc = self.roles.get(bg, -1) if bg else -1 |
|
attr = self.pair(self.roles.get(role, self.roles["value"]), bgc) |
|
# Index 8 is already the theme's grey, so dimming it a second time is |
|
# how chrome disappeared. Only the 8-colour fallback still needs A_DIM. |
|
if not self.rich and not self.themed and role in self.DIM_ROLES: |
|
attr |= curses.A_DIM if self.has_dim else 0 |
|
return attr | (curses.A_BOLD if bold else 0) |
|
|
|
def ramp_attr(self, t: float, bold: bool = False, bg: str = None) -> int: |
|
"""Colour for a 0..1 position along the dark→bright ramp.""" |
|
if not self.enabled or not self.ramp: |
|
return curses.A_BOLD if t >= 0.8 else (curses.A_DIM if t < 0.35 else 0) |
|
t = max(0.0, min(1.0, t)) |
|
idx = int(round(t * (len(self.ramp) - 1))) |
|
bgc = self.roles.get(bg, -1) if bg else -1 |
|
attr = self.pair(self.ramp[idx], bgc) |
|
if not self.rich and not self.themed: |
|
# Eight colours have no bright half to climb into, so the low end |
|
# has to recede with A_DIM and the top has to shout with A_BOLD. |
|
if t < 0.55 and self.has_dim: |
|
attr |= curses.A_DIM |
|
elif t > 0.9: |
|
attr |= curses.A_BOLD |
|
return attr | (curses.A_BOLD if bold else 0) |
|
|
|
|
|
# ---------------------------------------------------------------- sampling |
|
|
|
|
|
def parse_vm_stat() -> Dict[str, int]: |
|
out = subprocess.check_output(["vm_stat"], text=True, stderr=subprocess.DEVNULL) |
|
page_size = 4096 |
|
first = out.splitlines()[0] if out else "" |
|
if "page size of" in first: |
|
try: |
|
page_size = int(first.split("page size of", 1)[1].split("bytes", 1)[0].strip()) |
|
except Exception: |
|
pass |
|
vals = {} |
|
for line in out.splitlines()[1:]: |
|
if ":" not in line: |
|
continue |
|
k, v = line.split(":", 1) |
|
try: |
|
vals[k.strip()] = int(v.strip().rstrip(".").replace(".", "")) |
|
except ValueError: |
|
continue |
|
vals["_page_size"] = page_size |
|
return vals |
|
|
|
|
|
def pressure_level() -> int: |
|
"""Kernel memory pressure: 1 normal, 2 warn, 4 critical.""" |
|
try: |
|
out = subprocess.check_output( |
|
["sysctl", "-n", "kern.memorystatus_vm_pressure_level"], |
|
text=True, stderr=subprocess.DEVNULL).strip() |
|
return int(out) |
|
except Exception: |
|
return 1 |
|
|
|
|
|
def system_memory() -> Dict[str, float]: |
|
vm = parse_vm_stat() |
|
ps = vm.get("_page_size", 4096) |
|
active = vm.get("Pages active", 0) |
|
inactive = vm.get("Pages inactive", 0) |
|
wired = vm.get("Pages wired down", 0) |
|
speculative = vm.get("Pages speculative", 0) |
|
compressed = vm.get("Pages occupied by compressor", 0) |
|
purgeable = vm.get("Pages purgeable", 0) |
|
free = vm.get("Pages free", 0) |
|
|
|
try: |
|
total_bytes = int(subprocess.check_output(["sysctl", "-n", "hw.memsize"], text=True).strip()) |
|
except Exception: |
|
total_bytes = (active + inactive + wired + speculative + free + compressed) * ps |
|
|
|
# Activity Monitor's "Memory Used" is app memory + wired + compressed. |
|
# Inactive and speculative pages are reclaimable on demand, so counting |
|
# them as used pins the reading near 100% on any long-uptime machine and |
|
# disagrees with memsnap.py. Keep the two tools telling the same story. |
|
used_bytes = (active + wired + compressed) * ps |
|
used_bytes = max(0, min(used_bytes, total_bytes)) |
|
|
|
swap_used = swap_total = 0.0 |
|
try: |
|
s = subprocess.check_output(["sysctl", "vm.swapusage"], text=True).strip() |
|
# vm.swapusage: total = 4096.00M used = 123.50M free = ... |
|
parts = s.replace("=", " ").split() |
|
|
|
def parse_num(tok: str) -> float: |
|
mult = 1 |
|
if tok.endswith("K"): mult = 1024 |
|
elif tok.endswith("M"): mult = 1024 ** 2 |
|
elif tok.endswith("G"): mult = 1024 ** 3 |
|
elif tok.endswith("T"): mult = 1024 ** 4 |
|
return float(tok[:-1]) * mult if tok[-1].isalpha() else float(tok) |
|
|
|
for i, p in enumerate(parts): |
|
if p == "total" and i + 1 < len(parts): swap_total = parse_num(parts[i + 1]) |
|
if p == "used" and i + 1 < len(parts): swap_used = parse_num(parts[i + 1]) |
|
except Exception: |
|
pass |
|
|
|
return { |
|
"total_kb": total_bytes / 1024, |
|
"used_kb": used_bytes / 1024, |
|
# Real free pages. total-used would double-count inactive, which |
|
# appears as its own segment in the composition bar. |
|
"free_kb": free * ps / 1024, |
|
"available_kb": max(0.0, total_bytes / 1024 - used_bytes / 1024), |
|
"active_kb": active * ps / 1024, |
|
"inactive_kb": inactive * ps / 1024, |
|
"speculative_kb": speculative * ps / 1024, |
|
"compressed_kb": compressed * ps / 1024, |
|
"purgeable_kb": purgeable * ps / 1024, |
|
"wired_kb": wired * ps / 1024, |
|
"swap_total_kb": swap_total / 1024, |
|
"swap_used_kb": swap_used / 1024, |
|
"pressure": pressure_level(), |
|
} |
|
|
|
|
|
def process_list() -> List[Proc]: |
|
# comm gives executable name/path without argument noise; RSS is resident set KB. |
|
cmd = ["ps", "-axo", "pid=,ppid=,rss=,vsz=,%cpu=,user=,comm="] |
|
out = subprocess.check_output(cmd, text=True, errors="replace") |
|
procs = [] |
|
for line in out.splitlines(): |
|
parts = line.strip().split(None, 6) |
|
if len(parts) < 7: |
|
continue |
|
try: |
|
pid, ppid, rss, vsz = map(int, parts[:4]) |
|
cpu = float(parts[4]) |
|
except ValueError: |
|
continue |
|
procs.append(Proc(pid, ppid, rss, vsz, cpu, parts[5], parts[6])) |
|
return procs |
|
|
|
|
|
def basename(cmd: str) -> str: |
|
b = os.path.basename(cmd.rstrip("/")) |
|
return b or cmd |
|
|
|
|
|
def appish_name(cmd: str) -> str: |
|
"""Name a process by its outermost .app bundle, else the executable. |
|
|
|
Taking the *first* ".app/" component rolls helper processes up into the |
|
app that owns them, so "Discord Helper (Renderer)" counts as Discord. |
|
That is the whole point of the grouped view: multi-process apps are only |
|
visible once their helpers are added together. |
|
""" |
|
marker = ".app/" |
|
idx = cmd.find(marker) |
|
if idx != -1: |
|
return os.path.basename(cmd[:idx]) |
|
return basename(cmd) |
|
|
|
|
|
def group_procs(procs: List[Proc], mode: str): |
|
if mode == "pid": |
|
return [{ |
|
"name": f"{appish_name(p.command)} [{p.pid}]", |
|
"rss": p.rss_kb, "vsz": p.vsz_kb, "cpu": p.cpu, |
|
"count": 1, "pids": [p.pid], "user": p.user, |
|
} for p in procs] |
|
|
|
groups = defaultdict(lambda: {"rss": 0, "vsz": 0, "cpu": 0.0, "count": 0, "pids": [], "user": ""}) |
|
for p in procs: |
|
if mode == "user": |
|
key = p.user |
|
elif mode == "path": |
|
key = p.command |
|
else: |
|
key = appish_name(p.command) |
|
g = groups[key] |
|
g["rss"] += p.rss_kb |
|
g["vsz"] += p.vsz_kb |
|
g["cpu"] += p.cpu |
|
g["count"] += 1 |
|
if len(g["pids"]) < 8: g["pids"].append(p.pid) |
|
g["user"] = p.user if g["count"] == 1 else "multiple" |
|
rows = [] |
|
for name, g in groups.items(): |
|
g = dict(g); g["name"] = name; rows.append(g) |
|
return rows |
|
|
|
|
|
def print_snapshot() -> None: |
|
"""Print a useful report when no interactive terminal is available.""" |
|
mem = system_memory() |
|
total = mem["total_kb"] |
|
used = mem["used_kb"] |
|
rows = sorted(group_procs(process_list(), "name"), |
|
key=lambda row: row["rss"], reverse=True) |
|
|
|
percent = used / total * 100 if total else 0 |
|
verdict = {1: "ok", 2: "warn", 4: "critical"}.get(int(mem["pressure"]), "ok") |
|
print(f"Memory: {fmt_bytes(used)} / {fmt_bytes(total)} ({percent:.1f}%) pressure: {verdict}") |
|
print(f"Wired: {fmt_bytes(mem['wired_kb'])} " |
|
f"Compressed: {fmt_bytes(mem['compressed_kb'])} " |
|
f"Swap: {fmt_bytes(mem['swap_used_kb'])} / " |
|
f"{fmt_bytes(mem['swap_total_kb'])}") |
|
print() |
|
print(f"{'NAME':<42} {'MEM':>9} {'%MEM':>6} {'CPU':>6} {'N':>4}") |
|
for row in rows[:20]: |
|
name = row["name"] |
|
if len(name) > 42: |
|
name = name[:41] + "…" |
|
mem_percent = row["rss"] / total * 100 if total else 0 |
|
print(f"{name:<42} {fmt_bytes(row['rss']):>9} " |
|
f"{mem_percent:5.1f}% {row['cpu']:5.1f}% {row['count']:>4}") |
|
|
|
|
|
# ---------------------------------------------------------------- app |
|
|
|
|
|
class App: |
|
def __init__(self, stdscr, interval, color=True): |
|
self.s = stdscr |
|
self.interval = interval |
|
self.paused = False |
|
self.group_mode = "name" # name, pid, user, path |
|
self.sort = "rss" # rss, cpu, count, name |
|
self.reverse = True |
|
self.filter = "" |
|
self.show_help = False |
|
self.show_graph = True |
|
self.graph_symbol = "braille" # braille, block, shade |
|
self.last_refresh = 0.0 |
|
self.rows = [] |
|
self.mem = {} |
|
self.scroll = 0 |
|
self.selected = 0 |
|
self.status = "" |
|
self.want_color = color |
|
self.theme = Theme(color) |
|
self.hist = defaultdict(lambda: deque(maxlen=HISTORY_LEN)) |
|
|
|
# ---- data |
|
|
|
def refresh_data(self): |
|
try: |
|
self.mem = system_memory() |
|
total = self.mem.get("total_kb", 1) or 1 |
|
swap_t = self.mem.get("swap_total_kb", 0) |
|
for key, kb, denom in self.metric_rows(): |
|
self.hist[key].append((kb / denom) if denom else 0.0) |
|
procs = process_list() |
|
rows = group_procs(procs, self.group_mode) |
|
if self.filter: |
|
f = self.filter.lower() |
|
rows = [r for r in rows if f in r["name"].lower()] |
|
keyfn = { |
|
"rss": lambda r: r["rss"], |
|
"cpu": lambda r: r["cpu"], |
|
"count": lambda r: r["count"], |
|
"name": lambda r: r["name"].lower(), |
|
}[self.sort] |
|
self.rows = sorted(rows, key=keyfn, reverse=self.reverse) |
|
self.last_refresh = time.monotonic() |
|
self.status = "" |
|
self.selected = min(self.selected, max(0, len(self.rows) - 1)) |
|
except Exception as e: |
|
self.status = f"Refresh error: {e}" |
|
|
|
# ---- primitives |
|
|
|
def add(self, y, x, text, attr=0): |
|
h, w = self.s.getmaxyx() |
|
if y < 0 or y >= h or x >= w: |
|
return |
|
try: |
|
self.s.addnstr(y, x, text, max(0, w - x - 1), attr) |
|
except curses.error: |
|
pass |
|
|
|
def sparkline(self, y, x, width, samples: Sequence[float], attr=0): |
|
"""Single-row history track for one metric.""" |
|
if self.graph_symbol != "braille": |
|
chars = " ▁▂▃▄▅▆▇█" if self.graph_symbol == "block" else " ░░▒▒▓▓██" |
|
data = list(samples)[-width:] |
|
pad = width - len(data) |
|
for col, value in enumerate(data): |
|
level = int(max(0, min(8, round(max(0.0, min(1.0, value)) * 8)))) |
|
if level: |
|
self.add(y, x + pad + col, chars[level], attr) |
|
return |
|
need = width * 2 |
|
data = list(samples)[-need:] |
|
pad = need - len(data) |
|
masks = [0] * width |
|
for i, value in enumerate(data): |
|
value = max(0.0, min(1.0, value)) |
|
col = pad + i |
|
cell_x, sub = col // 2, col % 2 |
|
if cell_x >= width: |
|
continue |
|
for d in range(max(1, int(round(value * 4))) if value > 0 else 0): |
|
from_top = 3 - d |
|
masks[cell_x] |= BRAILLE_DOTS[sub][from_top] |
|
for cell_x, mask in enumerate(masks): |
|
if mask: |
|
self.add(y, x + cell_x, chr(0x2800 + mask), attr) |
|
|
|
def box(self, y, x, height, width, num="", title="", right=""): |
|
if height < 2 or width < 2: |
|
return |
|
border = self.theme.c("border") |
|
self.add(y, x, "╭" + "─" * (width - 2) + "╮", border) |
|
for i in range(1, height - 1): |
|
self.add(y + i, x, "│", border) |
|
self.add(y + i, x + width - 1, "│", border) |
|
self.add(y + height - 1, x, "╰" + "─" * (width - 2) + "╯", border) |
|
if title: |
|
cx = x + 2 |
|
if num: |
|
self.add(y, cx, num, self.theme.c("titlenum", bold=True)); cx += len(num) |
|
self.add(y, cx, title, self.theme.c("title", bold=True)) |
|
if right: |
|
rx = x + width - len(right) - 2 |
|
if rx > x + 2: |
|
self.add(y, rx, right, self.theme.c("label")) |
|
|
|
# ---- drawing |
|
|
|
def pressure_state(self): |
|
"""(label, 0..1 severity) from the kernel pressure level and swap use.""" |
|
level = int(self.mem.get("pressure", 1) or 1) |
|
swap_t = self.mem.get("swap_total_kb", 0) |
|
swap_frac = (self.mem.get("swap_used_kb", 0) / swap_t) if swap_t else 0.0 |
|
if level >= 4: |
|
return "critical", 1.0 |
|
if level == 2: |
|
return "warn", 0.72 |
|
if swap_frac > 0.85: |
|
return "swap full", 0.6 |
|
return "ok", 0.1 |
|
|
|
def span_label(self): |
|
longest = max((len(v) for v in self.hist.values()), default=0) |
|
span_s = longest * self.interval |
|
if span_s < 90: |
|
return f"{span_s:.0f}s history" |
|
return f"{span_s / 60:.0f}m history" |
|
|
|
def metric_rows(self): |
|
"""(key, kilobytes, denominator) for every row of the memory panel. |
|
|
|
Ordered most to least important; the panel shows as many as fit. |
|
""" |
|
m = self.mem |
|
total = m.get("total_kb", 1) or 1 |
|
swap_t = m.get("swap_total_kb", 0) |
|
return [ |
|
("used", m.get("used_kb", 0), total), |
|
("swap", m.get("swap_used_kb", 0), swap_t), |
|
("compressed", m.get("compressed_kb", 0), total), |
|
("wired", m.get("wired_kb", 0), total), |
|
("active", m.get("active_kb", 0), total), |
|
("inactive", m.get("inactive_kb", 0), total), |
|
("free", m.get("free_kb", 0), total), |
|
] |
|
|
|
def stat(self, y, x, width, label, kb, frac, spark_key=None): |
|
"""label · value · percent · thin history line — the panel's one row. |
|
|
|
Text-forward on purpose. Chunky meters turn the panel into slabs of |
|
colour, and stacked one per row they merge into a single block; btop |
|
keeps this panel almost entirely text and gives each metric a thin |
|
dotted trace instead. |
|
""" |
|
self.add(y, x, label, self.theme.c("label")) |
|
self.add(y, x + 12, f"{fmt_bytes(kb):>8}", self.theme.c("value")) |
|
self.add(y, x + 21, f"{frac * 100:3.0f}%", |
|
self.theme.ramp_attr(frac, bold=frac > 0.9)) |
|
if spark_key is not None: |
|
sx = x + 27 |
|
sw = max(0, min(40, width - 29)) |
|
if sw > 4: |
|
self.sparkline(y, sx, sw, self.hist[spark_key], |
|
self.theme.ramp_attr(frac * 0.75)) |
|
|
|
def draw(self): |
|
self.s.erase() |
|
h, w = self.s.getmaxyx() |
|
if h < 14 or w < 70: |
|
self.add(0, 0, "memtop needs a terminal at least 70x14. Resize to continue.", |
|
self.theme.c("bright")) |
|
self.s.refresh(); return |
|
|
|
total = self.mem.get("total_kb", 1) or 1 |
|
used = self.mem.get("used_kb", 0) |
|
comp = self.mem.get("compressed_kb", 0) |
|
wired = self.mem.get("wired_kb", 0) |
|
swap_u = self.mem.get("swap_used_kb", 0) |
|
swap_t = self.mem.get("swap_total_kb", 0) |
|
|
|
# --- title row |
|
self.add(0, 2, APP_NAME, self.theme.c("title", bold=True)) |
|
state, severity = self.pressure_state() |
|
live = "paused" if self.paused else "live" |
|
tail = f"{state} {live} {self.interval:g}s" |
|
rx = max(14, w - len(tail) - 3) |
|
self.add(0, rx, state, self.theme.ramp_attr(severity, bold=severity > 0.5)) |
|
self.add(0, rx + len(state) + 3, live, |
|
self.theme.c("value") if self.paused else self.theme.c("label")) |
|
self.add(0, rx + len(state) + 3 + len(live) + 1, f" {self.interval:g}s", |
|
self.theme.c("unit")) |
|
|
|
# --- memory panel |
|
# Text-forward, like btop's: one line per metric with a thin dotted |
|
# trace, rather than a column of chunky meters that merge into a slab. |
|
inner_x, inner_w = 3, w - 6 |
|
rows_all = self.metric_rows() |
|
room = h - 12 # leave the process panel usable |
|
n_rows = max(2, min(len(rows_all), room)) |
|
shown = rows_all[:n_rows] |
|
panel_h = n_rows + 2 |
|
self.box(1, 1, panel_h, w - 2, "1 ", "memory", |
|
f"{fmt_bytes(total)} total {self.span_label()}") |
|
for i, (key, kb, denom) in enumerate(shown): |
|
frac = (kb / denom) if denom else 0.0 |
|
label = "swap" if key == "swap" else key |
|
self.stat(2 + i, inner_x, inner_w, label, kb, frac, |
|
spark_key=key if self.show_graph else None) |
|
|
|
# --- process panel |
|
table_y = 1 + panel_h |
|
table_h = h - table_y - 2 |
|
if table_h < 4: |
|
self.s.refresh(); return |
|
|
|
mode_label = {"name": "by app", "pid": "by process", |
|
"user": "by user", "path": "by path"}[self.group_mode] |
|
right = f"{mode_label} sort {self.sort}{' ↓' if self.reverse else ' ↑'}" |
|
if self.filter: |
|
right += f" /{self.filter}" |
|
self.box(table_y, 1, table_h, w - 2, "2 ", "processes", right) |
|
|
|
# No per-row bar. Stacked down the list they merged into one solid |
|
# block of colour, and btop's process list has no such bar either -- |
|
# the percentage column already carries magnitude. |
|
fixed = 35 |
|
name_w = max(12, inner_w - fixed) |
|
header = (f"{'#':>3} {'name':<{name_w}} {'mem':>9} {'%':>5} " |
|
f"{'cpu':>6} {'n':>3}") |
|
self.add(table_y + 1, inner_x, header, self.theme.c("label") | curses.A_UNDERLINE) |
|
|
|
top = table_y + 2 |
|
visible = max(1, table_y + table_h - 1 - top) |
|
if self.selected < self.scroll: self.scroll = self.selected |
|
if self.selected >= self.scroll + visible: self.scroll = self.selected - visible + 1 |
|
self.scroll = max(0, min(self.scroll, max(0, len(self.rows) - visible))) |
|
|
|
# Heat is relative to the largest row on screen, so the ranking stays |
|
# readable whether the top consumer is 40% of RAM or 4%. |
|
peak = max((r["rss"] for r in self.rows[:200]), default=1) or 1 |
|
|
|
for screen_i, idx in enumerate(range(self.scroll, min(len(self.rows), self.scroll + visible))): |
|
r = self.rows[idx] |
|
y = top + screen_i |
|
mempct = (r["rss"] / total * 100) if total else 0 |
|
heat = min(1.0, r["rss"] / peak) |
|
sel = idx == self.selected |
|
n = r["name"] |
|
if len(n) > name_w: n = n[:name_w - 1] + "…" |
|
|
|
# A_REVERSE composed with the bold-black chrome roles gives black |
|
# on black, so without a grey palette to draw a band in, the whole |
|
# selected row is drawn as one flat reverse attribute. |
|
cpu = r["cpu"] |
|
x = inner_x |
|
# A marker plus bold reads as selection everywhere. A full-width |
|
# reverse band is the loudest thing on the screen, and at 8 |
|
# colours there is no grey to draw a subtle band in. |
|
self.add(y, x - 1, "▸" if sel else " ", self.theme.c("titlenum")) |
|
self.add(y, x, f"{idx + 1:>3}", self.theme.c("unit")); x += 5 |
|
self.add(y, x, f"{n:<{name_w}}", |
|
self.theme.c("bright" if (sel or heat > 0.45) else "value", |
|
bold=sel)); x += name_w + 1 |
|
self.add(y, x, f"{fmt_bytes(r['rss']):>9}", |
|
self.theme.c("value", bold=heat > 0.45)); x += 10 |
|
self.add(y, x, f"{mempct:4.1f}%", |
|
self.theme.ramp_attr(heat * 0.8)); x += 6 |
|
self.add(y, x, f"{cpu:5.1f}%", |
|
self.theme.ramp_attr(min(1.0, cpu / 100.0)) |
|
if cpu >= 10 else self.theme.c("unit")); x += 7 |
|
self.add(y, x, f"{r['count']:>3}", |
|
self.theme.c("label" if r["count"] > 8 else "unit")) |
|
|
|
# --- footer |
|
keys = [("q", "quit"), ("?", "help"), ("g", "group"), ("s", "sort"), |
|
("r", "reverse"), ("+/-", "rate"), ("spc", "pause"), |
|
("/", "filter"), ("G", "graph")] |
|
x = 2 |
|
for key, desc in keys: |
|
if x + len(key) + len(desc) + 2 >= w: break |
|
self.add(h - 2, x, key, self.theme.c("titlenum")) |
|
self.add(h - 2, x + len(key) + 1, desc, self.theme.c("label")) |
|
x += len(key) + len(desc) + 3 |
|
if self.status: |
|
self.add(h - 1, 2, self.status, self.theme.ramp_attr(1.0, bold=True)) |
|
else: |
|
if self.rows: |
|
r = self.rows[self.selected] |
|
self.add(h - 1, 2, r["name"][:max(0, w - 46)], self.theme.c("value")) |
|
sx = 2 + min(len(r["name"]), max(0, w - 46)) + 2 |
|
detail = f"rss {fmt_bytes(r['rss'])} pids {','.join(map(str, r['pids']))}" |
|
self.add(h - 1, sx, detail[:max(0, w - sx - 12)], self.theme.c("unit")) |
|
if not self.paused: |
|
remaining = max(0.0, self.interval - (time.monotonic() - self.last_refresh)) |
|
self.add(h - 1, max(2, w - 10), f"{remaining:4.1f}s", self.theme.c("unit")) |
|
|
|
if self.show_help: |
|
self.draw_help() |
|
self.s.refresh() |
|
|
|
def draw_help(self): |
|
h, w = self.s.getmaxyx() |
|
color_note = (f"colour: {self.theme.colors} available" |
|
if self.theme.enabled else "colour: unavailable (monochrome)") |
|
rows = [ |
|
("q / Esc", "quit / close help"), |
|
("Space", "pause/resume refresh"), |
|
("+ / =", "faster refresh (down to 0.5s)"), |
|
("- / _", "slower refresh (up to 60s)"), |
|
("g", "grouping: app → pid → user → path"), |
|
("s", "sort: memory → CPU → count → name"), |
|
("r", "reverse sort direction"), |
|
("/", "set a name filter; Enter applies"), |
|
("c", "clear filter"), |
|
("G", "show/hide the history graph"), |
|
("S", "graph symbol: braille → block → shade"), |
|
("C", "toggle colour"), |
|
("↑/↓ or j/k", "move selection"), |
|
("PgUp/PgDn", "move a page"), |
|
("Home/End", "first/last row"), |
|
] |
|
notes = [ |
|
"Memory is RSS from macOS ps. The default view sums RSS across", |
|
"every process belonging to the same .app bundle, so helper", |
|
"processes count toward the app that spawned them.", |
|
"", |
|
"Used = active + wired + compressed, matching Activity Monitor.", |
|
"Meters run dark→bright across their length; row shading is", |
|
"relative to the largest process group on screen.", |
|
"", |
|
color_note, |
|
"For the full 256-colour ramp run with TERM=xterm-256color.", |
|
] |
|
body_w = max([len(k) + 3 + len(d) for k, d in rows] + [len(n) for n in notes]) |
|
box_w = min(w - 6, body_w + 4) |
|
box_h = min(h - 4, len(rows) + len(notes) + 3) |
|
y0 = max(1, (h - box_h) // 2); x0 = max(2, (w - box_w) // 2) |
|
for i in range(box_h): |
|
self.add(y0 + i, x0, " " * box_w) |
|
self.box(y0, x0, box_h, box_w, "", "help") |
|
y = y0 + 1 |
|
keyw = max(len(k) for k, _ in rows) |
|
for key, desc in rows: |
|
if y >= y0 + box_h - 1: break |
|
self.add(y, x0 + 2, f"{key:<{keyw}}", self.theme.c("titlenum")) |
|
self.add(y, x0 + 2 + keyw + 2, desc, self.theme.c("value")) |
|
y += 1 |
|
y += 1 |
|
for note in notes: |
|
if y >= y0 + box_h - 1: break |
|
self.add(y, x0 + 2, note[:box_w - 4], self.theme.c("label")) |
|
y += 1 |
|
|
|
def prompt_filter(self): |
|
h, w = self.s.getmaxyx() |
|
prompt = "/ filter: " |
|
curses.echo(); set_cursor_visibility(1) |
|
self.add(h - 1, 0, " " * (w - 1)) |
|
self.add(h - 1, 0, prompt, self.theme.c("title", bold=True)) |
|
self.s.refresh() |
|
try: |
|
raw = self.s.getstr(h - 1, len(prompt), max(1, w - len(prompt) - 2)) |
|
self.filter = raw.decode(errors="replace").strip() |
|
self.selected = self.scroll = 0 |
|
self.refresh_data() |
|
except Exception: |
|
pass |
|
finally: |
|
curses.noecho(); set_cursor_visibility(0) |
|
|
|
def handle(self, ch): |
|
if self.show_help: |
|
if ch in (ord('?'), ord('q'), 27): self.show_help = False |
|
return True |
|
if ch in (ord('q'), 27): return False |
|
if ch == ord('?'): self.show_help = True |
|
elif ch == ord(' '): self.paused = not self.paused |
|
elif ch in (ord('+'), ord('=')): |
|
self.interval = max(MIN_INTERVAL, round(self.interval - 0.5, 1)); self.last_refresh = time.monotonic() |
|
elif ch in (ord('-'), ord('_')): |
|
self.interval = min(MAX_INTERVAL, round(self.interval + 0.5, 1)); self.last_refresh = time.monotonic() |
|
elif ch == ord('g'): |
|
modes = ["name", "pid", "user", "path"]; self.group_mode = modes[(modes.index(self.group_mode) + 1) % len(modes)]; self.selected = self.scroll = 0; self.refresh_data() |
|
elif ch == ord('s'): |
|
sorts = ["rss", "cpu", "count", "name"]; self.sort = sorts[(sorts.index(self.sort) + 1) % len(sorts)]; self.refresh_data() |
|
elif ch == ord('r'): self.reverse = not self.reverse; self.refresh_data() |
|
elif ch == ord('/'): self.prompt_filter() |
|
elif ch == ord('c'): self.filter = ""; self.selected = self.scroll = 0; self.refresh_data() |
|
elif ch == ord('G'): self.show_graph = not self.show_graph |
|
elif ch == ord('S'): |
|
syms = ["braille", "block", "shade"] |
|
self.graph_symbol = syms[(syms.index(self.graph_symbol) + 1) % len(syms)] |
|
elif ch == ord('C'): |
|
self.want_color = not self.want_color |
|
self.theme = Theme(self.want_color) |
|
elif ch in (curses.KEY_DOWN, ord('j')): self.selected = min(len(self.rows) - 1, self.selected + 1) if self.rows else 0 |
|
elif ch in (curses.KEY_UP, ord('k')): self.selected = max(0, self.selected - 1) |
|
elif ch == curses.KEY_NPAGE: self.selected = min(len(self.rows) - 1, self.selected + 10) if self.rows else 0 |
|
elif ch == curses.KEY_PPAGE: self.selected = max(0, self.selected - 10) |
|
elif ch == curses.KEY_HOME: self.selected = 0 |
|
elif ch == curses.KEY_END: self.selected = max(0, len(self.rows) - 1) |
|
return True |
|
|
|
def run(self): |
|
set_cursor_visibility(0) |
|
self.s.keypad(True); self.s.timeout(100) |
|
self.refresh_data() |
|
running = True |
|
while running: |
|
if not self.paused and time.monotonic() - self.last_refresh >= self.interval: |
|
self.refresh_data() |
|
self.draw() |
|
ch = self.s.getch() |
|
if ch != -1: running = self.handle(ch) |
|
|
|
|
|
def main(): |
|
interval = DEFAULT_INTERVAL |
|
color = True |
|
args = [a for a in sys.argv[1:]] |
|
if "--no-color" in args: |
|
color = False |
|
args.remove("--no-color") |
|
if args: |
|
if args[0] in ("-h", "--help"): |
|
print("Usage: memtop.py [refresh-seconds] [--no-color]\nExample: memtop.py 2") |
|
return |
|
try: |
|
interval = max(MIN_INTERVAL, min(MAX_INTERVAL, float(args[0]))) |
|
except ValueError: |
|
print("Refresh interval must be a number between 0.5 and 60 seconds.", file=sys.stderr); sys.exit(2) |
|
if sys.platform != "darwin": |
|
print("memtop is designed for macOS.", file=sys.stderr); sys.exit(1) |
|
if not (sys.stdin.isatty() and sys.stdout.isatty()): |
|
try: |
|
print_snapshot() |
|
except (OSError, subprocess.SubprocessError) as exc: |
|
print(f"memtop could not read system information: {exc}", file=sys.stderr) |
|
sys.exit(1) |
|
return |
|
curses.wrapper(lambda s: App(s, interval, color).run()) |
|
|
|
|
|
if __name__ == "__main__": |
|
main() |