termtree.py

#!/usr/bin/env python3
"""termtree — ASCII disk usage visualizer with truecolor heat gradient.

TreeSize-inspired CLI tool. Shows what takes space: tree with gradient bars,
file-type clustering, and a squarified ASCII treemap. Stdlib only.

Homepage: https://github.com/daniellauding/termtree
License:  MIT
"""
from __future__ import annotations
import os
import sys
import re
import argparse
import subprocess
import time
from collections import defaultdict

__version__ = "0.3.0"

UNITS = ['B', 'KB', 'MB', 'GB', 'TB', 'PB']
BLOCKS = ' ▏▎▍▌▋▊▉█'

# 24-bit truecolor heat stops: cool blue → cyan → green → yellow → orange → red
HEAT_STOPS = [
    (0.00, (70, 90, 130)),
    (0.05, (80, 180, 220)),
    (0.15, (120, 220, 130)),
    (0.30, (240, 220, 90)),
    (0.55, (245, 160, 60)),
    (1.00, (240, 80, 80)),
]

RESET = '\033[0m'
BOLD = '\033[1m'
DIM = '\033[2m'

USE_COLOR = True
USE_TRUECOLOR = True


def supports_truecolor():
    return os.environ.get('COLORTERM', '').lower() in ('truecolor', '24bit')


def rgb(r, g, b):
    if not USE_COLOR:
        return ''
    if USE_TRUECOLOR:
        return f'\033[38;2;{r};{g};{b}m'
    # 256-color approximation
    idx = 16 + 36 * (r * 5 // 255) + 6 * (g * 5 // 255) + (b * 5 // 255)
    return f'\033[38;5;{idx}m'


def reset():
    return RESET if USE_COLOR else ''


def bold():
    return BOLD if USE_COLOR else ''


def dim():
    return DIM if USE_COLOR else ''


def _lerp(a, b, t):
    return tuple(int(a[i] + (b[i] - a[i]) * t) for i in range(3))


def heat_rgb(pct):
    """Smooth heat-gradient color for value in [0,1]."""
    pct = max(0.0, min(1.0, pct))
    for i in range(len(HEAT_STOPS) - 1):
        p0, c0 = HEAT_STOPS[i]
        p1, c1 = HEAT_STOPS[i + 1]
        if pct <= p1:
            t = (pct - p0) / (p1 - p0) if p1 > p0 else 0
            return _lerp(c0, c1, t)
    return HEAT_STOPS[-1][1]


def heat(pct):
    r, g, b = heat_rgb(pct)
    return rgb(r, g, b)


def fmt_size(n):
    f, i = float(n), 0
    while f >= 1024 and i < len(UNITS) - 1:
        f /= 1024
        i += 1
    if i == 0:
        return f"{int(f)} {UNITS[i]}"
    return f"{f:.1f} {UNITS[i]}"


def gradient_bar(pct, width=20):
    """Bar with cold→hot gradient across filled cells."""
    pct = max(0.0, min(1.0, pct))
    filled = pct * width
    full = int(filled)
    parts = []
    for i in range(full):
        # Position in 0..1 across the FULL bar (so a 100% bar shows full rainbow)
        cell_pct = (i + 1) / width
        r, g, b = heat_rgb(cell_pct)
        parts.append(f"{rgb(r, g, b)}█")
    rendered = full
    if full < width:
        idx = int((filled - full) * 8)
        if idx > 0:
            cell_pct = (full + 1) / width
            r, g, b = heat_rgb(cell_pct)
            parts.append(f"{rgb(r, g, b)}{BLOCKS[idx]}")
            rendered += 1
    if USE_COLOR:
        parts.append(RESET)
    parts.append(' ' * (width - rendered))
    return ''.join(parts)


def icon(node):
    if not node.is_dir:
        return '  '
    n = node.name
    if n.endswith('.app'):
        return '📱'
    if n.endswith('.framework'):
        return '⚙️ '
    if n.endswith('.bundle'):
        return '📦'
    if n.startswith('.'):
        return '🔒'
    return '📁'


class Node:
    __slots__ = ('name', 'path', 'size', 'is_dir', 'children', 'file_count')

    def __init__(self, name, path, is_dir):
        self.name, self.path, self.is_dir = name, path, is_dir
        self.size = 0
        self.children = []
        self.file_count = 0


def _file_bytes(st, apparent):
    """Bytes the file actually occupies (default: physical via st_blocks).
    For sparse files (Docker.raw, VM images, time machine bundles) physical can
    be vastly smaller than logical. `apparent=True` returns logical size like
    `du --apparent-size` / `ls -l`."""
    if apparent:
        return st.st_size
    blocks = getattr(st, 'st_blocks', None)
    if blocks is None:  # non-POSIX
        return st.st_size
    return blocks * 512


def scan(root_path, follow_symlinks=False, on_progress=None, apparent=False):
    ext_stats = defaultdict(lambda: [0, 0])  # ext -> [size, count]
    counter = [0]

    def _scan(path, name, is_link):
        if is_link and not follow_symlinks:
            try:
                st = os.lstat(path)
            except OSError:
                return None
            n = Node(name, path, False)
            n.size = _file_bytes(st, apparent)
            n.file_count = 1
            return n
        try:
            is_dir = os.path.isdir(path)
        except OSError:
            return None

        if is_dir:
            n = Node(name, path, True)
            try:
                with os.scandir(path) as it:
                    entries = list(it)
            except (PermissionError, OSError):
                return n
            # Opaque bundles: sum content, don't show internals (keeps macOS sane)
            opaque = name.endswith(('.app', '.framework', '.bundle'))
            for entry in entries:
                try:
                    is_l = entry.is_symlink()
                except OSError:
                    is_l = False
                child = _scan(entry.path, entry.name, is_l)
                if child:
                    n.size += child.size
                    n.file_count += child.file_count
                    if not opaque:
                        n.children.append(child)
            return n
        else:
            try:
                st = os.lstat(path) if is_link else os.stat(path)
            except OSError:
                return None
            n = Node(name, path, False)
            n.size = _file_bytes(st, apparent)
            n.file_count = 1
            ext = os.path.splitext(name)[1].lower() or '(no ext)'
            ext_stats[ext][0] += n.size
            ext_stats[ext][1] += 1
            counter[0] += 1
            if on_progress and counter[0] % 5000 == 0:
                on_progress(counter[0])
            return n

    abs_path = os.path.abspath(root_path)
    name = os.path.basename(abs_path) or abs_path
    return _scan(abs_path, name, os.path.islink(abs_path)), ext_stats


def disk_context(path):
    """Return (total, used, free) bytes for the filesystem containing path."""
    try:
        fs = os.statvfs(path)
    except OSError:
        return None
    total = fs.f_blocks * fs.f_frsize
    free = fs.f_bavail * fs.f_frsize
    used = total - free
    return total, used, free


BANNER = [
    "████████╗███████╗██████╗ ███╗   ███╗████████╗██████╗ ███████╗███████╗",
    "╚══██╔══╝██╔════╝██╔══██╗████╗ ████║╚══██╔══╝██╔══██╗██╔════╝██╔════╝",
    "   ██║   █████╗  ██████╔╝██╔████╔██║   ██║   ██████╔╝█████╗  █████╗  ",
    "   ██║   ██╔══╝  ██╔══██╗██║╚██╔╝██║   ██║   ██╔══██╗██╔══╝  ██╔══╝  ",
    "   ██║   ███████╗██║  ██║██║ ╚═╝ ██║   ██║   ██║  ██║███████╗███████╗",
    "   ╚═╝   ╚══════╝╚═╝  ╚═╝╚═╝     ╚═╝   ╚═╝   ╚═╝  ╚═╝╚══════╝╚══════╝",
]


def print_banner(path, size, count, disk_info=None, apparent=False):
    """Rainbow gradient banner + powerline-style breadcrumb."""
    if USE_COLOR:
        width = len(BANNER[0])
        for line in BANNER:
            out = []
            last = ''
            for i, ch in enumerate(line):
                if ch == ' ':
                    out.append(' ')
                    continue
                pos = i / max(1, width - 1)
                r, g, b = heat_rgb(pos)
                color = rgb(r, g, b)
                if color != last:
                    out.append(color)
                    last = color
                out.append(ch)
            out.append(RESET)
            print(''.join(out))
    else:
        for line in BANNER:
            print(line)
    r1, g1, b1 = heat_rgb(0.05)
    r2, g2, b2 = heat_rgb(0.55)
    sep = f"{rgb(180, 180, 200)} ▸ {reset()}"
    mode = 'apparent' if apparent else 'on-disk'
    crumb = (f"{bold()}{rgb(r1, g1, b1)}termtree{reset()}"
             f"{sep}{rgb(r2, g2, b2)}{path}{reset()}"
             f"{sep}{bold()}{fmt_size(size)}{reset()}"
             f"{dim()} · {count:,} files · {mode}{reset()}")
    print(crumb)
    if disk_info:
        total, used, free = disk_info
        pct_of_disk = (size / total) if total else 0
        disk_pct = (used / total) if total else 0
        ph = heat(disk_pct)
        scan_pct_color = heat(min(1.0, pct_of_disk * 2))  # amplify so 50% looks hot
        print(f"{dim()}  Volume: {reset()}"
              f"{ph}{fmt_size(used)} used{reset()}"
              f"{dim()} / {fmt_size(total)} total · {fmt_size(free)} free "
              f"({disk_pct * 100:.0f}% full){reset()}"
              f"  {dim()}·{reset()}  "
              f"{scan_pct_color}this scan = {pct_of_disk * 100:.1f}% of volume{reset()}")
    print()


def print_tree(root, max_depth=2, top_n=10):
    total = root.size or 1
    label = f"  {'BAR':<18}   {'%':>5}    {'SIZE':>9}     NAME"
    print(f"{bold()}{label}{reset()}")
    print(f"{dim()}{'─' * 78}{reset()}")
    _walk(root, total, max_depth, top_n, '', 0)


def _walk(node, total, max_depth, top_n, prefix, depth):
    if not node.is_dir or depth >= max_depth:
        return
    children = sorted(node.children, key=lambda c: c.size, reverse=True)
    rest = []
    if top_n and len(children) > top_n:
        rest = children[top_n:]
        children = children[:top_n]
    n_lines = len(children) + (1 if rest else 0)
    for i, c in enumerate(children):
        is_last = (i == n_lines - 1)
        connector = '└── ' if is_last else '├── '
        pct = c.size / total if total else 0
        h = heat(pct)
        size_s = fmt_size(c.size)
        print(f"{dim()}{prefix}{connector}{reset()}"
              f"{gradient_bar(pct, 18)} "
              f"{h}{pct * 100:5.1f}%{reset()} "
              f"{size_s:>9} {icon(c)} {c.name}")
        if c.is_dir:
            ext = '    ' if is_last else '│   '
            _walk(c, total, max_depth, top_n, prefix + ext, depth + 1)
    if rest:
        rest_size = sum(c.size for c in rest)
        rest_count = sum(c.file_count for c in rest)
        pct = rest_size / total if total else 0
        h = heat(pct)
        print(f"{dim()}{prefix}└── {reset()}"
              f"{gradient_bar(pct, 18)} "
              f"{h}{pct * 100:5.1f}%{reset()} "
              f"{fmt_size(rest_size):>9} {dim()}… {len(rest)} more "
              f"({rest_count:,} files){reset()}")


def print_extensions(ext_stats, total, top_n=15):
    if not ext_stats:
        return
    items = sorted(ext_stats.items(), key=lambda kv: kv[1][0], reverse=True)[:top_n]
    print(f"\n{bold()}File types{reset()}")
    print(f"{dim()}{'─' * 78}{reset()}")
    for ext, (size, count) in items:
        pct = size / total if total else 0
        h = heat(pct)
        print(f"  {gradient_bar(pct, 22)} "
              f"{h}{pct * 100:5.1f}%{reset()} "
              f"{fmt_size(size):>9} {count:>8,}× {ext}")


def squarify(sizes, x, y, w, h):
    """Squarified treemap layout. Returns list of (x,y,w,h) per input index."""
    n = len(sizes)
    boxes = [None] * n
    if not sizes or w <= 0 or h <= 0:
        return boxes
    total = sum(sizes)
    if total <= 0:
        return boxes
    norm = [s * w * h / total for s in sizes]
    indices = sorted(range(n), key=lambda i: -sizes[i])

    def worst(row, side):
        if not row or side <= 0:
            return float('inf')
        s = sum(row)
        if s <= 0:
            return float('inf')
        rmax, rmin = max(row), min(row)
        return max(side * side * rmax / (s * s), s * s / (side * side * rmin))

    def layout(idx_list, x, y, w, h):
        if not idx_list or w <= 0.01 or h <= 0.01:
            return
        if len(idx_list) == 1:
            boxes[idx_list[0]] = (x, y, w, h)
            return
        rest = list(idx_list)
        # Char cells are ~2:1 tall:wide → scale h by 2 for visual squareness
        side = min(w, h * 2.0)
        row = []
        row_idx = []
        while rest:
            new_row = row + [norm[rest[0]]]
            if not row or worst(new_row, side) <= worst(row, side):
                row = new_row
                row_idx.append(rest.pop(0))
            else:
                break
        rs = sum(row)
        if w >= h * 2.0:
            rw = rs / h
            off = 0
            for ri, sv in zip(row_idx, row):
                bh = sv / rw if rw > 0 else 0
                boxes[ri] = (x, y + off, rw, bh)
                off += bh
            layout(rest, x + rw, y, w - rw, h)
        else:
            rh = rs / w
            off = 0
            for ri, sv in zip(row_idx, row):
                bw = sv / rh if rh > 0 else 0
                boxes[ri] = (x + off, y, bw, rh)
                off += bw
            layout(rest, x, y + rh, w, h - rh)

    layout(indices, float(x), float(y), float(w), float(h))
    return boxes


def render_treemap(node, width=78, height=20):
    children = sorted([c for c in node.children if c.size > 0],
                      key=lambda c: c.size, reverse=True)
    if not children:
        return
    raw = squarify([c.size for c in children], 0, 0, width, height)
    grid = [[(' ', '')] * width for _ in range(height)]

    for idx, b in enumerate(raw):
        if b is None:
            continue
        bx, by, bw, bh = b
        x1, y1 = int(round(bx)), int(round(by))
        x2 = min(int(round(bx + bw)), width)
        y2 = min(int(round(by + bh)), height)
        if x2 - x1 < 2 or y2 - y1 < 1:
            continue
        c = children[idx]
        pct = c.size / node.size if node.size else 0
        color = heat(pct)
        for x in range(x1, x2):
            grid[y1][x] = ('─', color)
            grid[y2 - 1][x] = ('─', color)
        for y in range(y1, y2):
            grid[y][x1] = ('│', color)
            grid[y][x2 - 1] = ('│', color)
        grid[y1][x1] = ('┌', color)
        grid[y1][x2 - 1] = ('┐', color)
        grid[y2 - 1][x1] = ('└', color)
        grid[y2 - 1][x2 - 1] = ('┘', color)

        inner_w = x2 - x1 - 2
        if inner_w >= 3 and y2 - y1 >= 3:
            name = c.name
            if len(name) > inner_w:
                name = name[:inner_w - 1] + '…'
            label_color = color + (BOLD if USE_COLOR else '')
            for i, ch in enumerate(name):
                grid[y1 + 1][x1 + 1 + i] = (ch, label_color)
            if y2 - y1 >= 4:
                lab = f"{fmt_size(c.size)} {pct * 100:.0f}%"
                if len(lab) > inner_w:
                    lab = f"{pct * 100:.0f}%"
                for i, ch in enumerate(lab[:inner_w]):
                    grid[y1 + 2][x1 + 1 + i] = (ch, color)

    print(f"\n{bold()}Treemap{reset()} {dim()}(squarified, color = heat by size){reset()}")
    for row in grid:
        out, last = [], ''
        for ch, c in row:
            if c != last:
                if last:
                    out.append(RESET if USE_COLOR else '')
                out.append(c)
                last = c
            out.append(ch)
        if last:
            out.append(RESET if USE_COLOR else '')
        print(''.join(out))


# ─────────────────────────────────────────────────────────────────────
# System overview mode (--sys): CPU, memory, disk, network in one snapshot
# ─────────────────────────────────────────────────────────────────────


def fmt_rate(bps):
    if bps >= 1e9:
        return f"{bps / 1e9:.2f} GB/s"
    if bps >= 1e6:
        return f"{bps / 1e6:.2f} MB/s"
    if bps >= 1e3:
        return f"{bps / 1e3:.1f} KB/s"
    return f"{int(bps)} B/s"


def _parse_unit(s):
    """'16G', '2389M', '4523K' → bytes."""
    if not s:
        return 0
    s = s.strip().rstrip('+-')
    if not s:
        return 0
    mult = {'B': 1, 'K': 1024, 'M': 1024 ** 2, 'G': 1024 ** 3, 'T': 1024 ** 4}
    u = s[-1].upper()
    if u in mult:
        try:
            return int(float(s[:-1]) * mult[u])
        except ValueError:
            return 0
    try:
        return int(s)
    except ValueError:
        return 0


def _parse_top_header(text):
    info = {}
    for line in text.splitlines():
        if line.startswith('Load Avg:'):
            m = re.match(r'Load Avg:\s*([\d.]+),\s*([\d.]+),\s*([\d.]+)', line)
            if m:
                info['load'] = (float(m.group(1)), float(m.group(2)), float(m.group(3)))
        elif line.startswith('CPU usage:'):
            m = re.match(r'CPU usage:\s*([\d.]+)% user,\s*([\d.]+)% sys,\s*([\d.]+)% idle', line)
            if m:
                info['cpu_user'] = float(m.group(1))
                info['cpu_sys'] = float(m.group(2))
                info['cpu_busy'] = 100.0 - float(m.group(3))
        elif line.startswith('PhysMem:'):
            m = re.search(
                r'PhysMem:\s*(\S+) used \((\S+) wired,?\s*(\S+)?\s*compressor\)?,\s*(\S+) unused',
                line)
            if m:
                info['mem_used'] = _parse_unit(m.group(1))
                info['mem_wired'] = _parse_unit(m.group(2))
                info['mem_compressed'] = _parse_unit(m.group(3) or '0')
                info['mem_unused'] = _parse_unit(m.group(4))
        elif line.startswith('Networks:'):
            m = re.search(r'packets:\s*\d+/(\S+) in,\s*\d+/(\S+) out', line)
            if m:
                info['net_in_total'] = _parse_unit(m.group(1))
                info['net_out_total'] = _parse_unit(m.group(2))
        elif line.startswith('Disks:'):
            m = re.search(r'\d+/(\S+) read,\s*\d+/(\S+) written', line)
            if m:
                info['disk_read_total'] = _parse_unit(m.group(1))
                info['disk_write_total'] = _parse_unit(m.group(2))
    return info


def _parse_top_procs(text):
    procs = []
    in_proc = False
    for line in text.splitlines():
        if line.startswith('PID') and 'COMMAND' in line:
            in_proc = True
            continue
        if not in_proc or not line.strip():
            continue
        if line.startswith('Processes:'):
            break  # next snapshot
        parts = line.split(None, 4)
        if len(parts) < 5:
            continue
        try:
            procs.append({
                'pid': int(parts[0]),
                'cpu': float(parts[1]),
                'mem': _parse_unit(parts[2]),
                'user': parts[3],
                'cmd': parts[4].strip(),
            })
        except (ValueError, IndexError):
            continue
    return procs


def _swap_info():
    """macOS swap usage + cumulative swap I/O. Returns (total, used, ins, outs)
    or None if unavailable. Active swap I/O is the real "out of memory" signal —
    on macOS, high memory % alone means almost nothing."""
    try:
        out = subprocess.run(['sysctl', '-n', 'vm.swapusage'],
                             capture_output=True, text=True, check=True, timeout=5).stdout
    except (subprocess.CalledProcessError, subprocess.TimeoutExpired, FileNotFoundError):
        return None
    m = re.search(r'total\s*=\s*([\d.]+)([BKMGT])\s+used\s*=\s*([\d.]+)([BKMGT])', out)
    if not m:
        return None
    total = _parse_unit(m.group(1) + m.group(2))
    used = _parse_unit(m.group(3) + m.group(4))
    swap_ins = swap_outs = 0
    try:
        vm = subprocess.run(['vm_stat'], capture_output=True, text=True,
                            check=True, timeout=5).stdout
        si = re.search(r'Swapins:\s+(\d+)', vm)
        so = re.search(r'Swapouts:\s+(\d+)', vm)
        if si:
            swap_ins = int(si.group(1))
        if so:
            swap_outs = int(so.group(1))
    except Exception:
        pass
    return total, used, swap_ins, swap_outs


def _netstat_ib():
    """Per-interface (Ibytes, Obytes) since boot — exact bytes (not rounded)."""
    try:
        out = subprocess.run(['netstat', '-ibn'],
                             capture_output=True, text=True, check=True, timeout=5).stdout
    except (subprocess.CalledProcessError, subprocess.TimeoutExpired, FileNotFoundError):
        return {}
    result = {}
    lines = out.splitlines()
    if not lines:
        return result
    header = lines[0].split()
    try:
        ib_idx = header.index('Ibytes')
        ob_idx = header.index('Obytes')
    except ValueError:
        return result
    seen = set()
    for line in lines[1:]:
        parts = line.split()
        if len(parts) <= max(ib_idx, ob_idx):
            continue
        iface = parts[0]
        if iface in seen:
            continue
        seen.add(iface)
        try:
            result[iface] = (int(parts[ib_idx]), int(parts[ob_idx]))
        except ValueError:
            continue
    return result


def collect_system(sample_sec=2.0):
    """Run `top -l 2` and compute deltas. Use netstat -ib for exact network
    bytes (top rounds to MB so small traffic shows 0). Enrich procs via ps."""
    sample_sec = max(1.0, float(sample_sec))
    # Take netstat snapshot *before* top runs; second one after.
    net1 = _netstat_ib()
    t1 = time.monotonic()
    try:
        out = subprocess.run(
            ['top', '-l', '2', '-s', str(int(sample_sec)),
             '-n', '60', '-stats', 'pid,cpu,mem,user,command'],
            capture_output=True, text=True, check=True, timeout=int(sample_sec) + 15
        ).stdout
    except (subprocess.CalledProcessError, subprocess.TimeoutExpired, FileNotFoundError):
        return None
    net2 = _netstat_ib()
    elapsed = max(0.5, time.monotonic() - t1)

    splits = re.split(r'(?m)^Processes:', out)
    snapshots = ['Processes:' + s for s in splits if s.strip()]
    if len(snapshots) < 2:
        return None
    sys1 = _parse_top_header(snapshots[0])
    sys2 = _parse_top_header(snapshots[-1])
    procs2 = _parse_top_procs(snapshots[-1])

    # Network: prefer exact netstat deltas
    iface_rates = {}
    net_in_rate = 0.0
    net_out_rate = 0.0
    for iface, (i1, o1) in net1.items():
        if iface not in net2:
            continue
        i2, o2 = net2[iface]
        di = max(0, i2 - i1) / elapsed
        do = max(0, o2 - o1) / elapsed
        if di + do > 0 or iface in ('en0', 'en1'):
            iface_rates[iface] = (di, do)
        # Exclude loopback from the system total
        if iface.startswith('lo'):
            continue
        net_in_rate += di
        net_out_rate += do
    sys2['net_in_rate'] = net_in_rate
    sys2['net_out_rate'] = net_out_rate
    sys2['iface_rates'] = iface_rates

    # Disk I/O: top is rounded, but no easy stdlib alternative — keep as approximation
    sys2['disk_read_rate'] = max(0, sys2.get('disk_read_total', 0) - sys1.get('disk_read_total', 0)) / elapsed
    sys2['disk_write_rate'] = max(0, sys2.get('disk_write_total', 0) - sys1.get('disk_write_total', 0)) / elapsed

    # Enrich processes with full command lines and derived app names
    try:
        ps_out = subprocess.run(['ps', '-axwwwo', 'pid,command'],
                                capture_output=True, text=True, check=True, timeout=10).stdout
        full = {}
        for ln in ps_out.strip().split('\n')[1:]:
            ps_parts = ln.split(None, 1)
            if len(ps_parts) == 2:
                try:
                    full[int(ps_parts[0])] = ps_parts[1]
                except ValueError:
                    continue
        for p in procs2:
            fcmd = full.get(p['pid'], p['cmd'])
            p['full'] = fcmd
            m = re.search(r'/([^/]+)\.app/', fcmd)
            if m:
                p['app'] = m.group(1)
            else:
                first = fcmd.split()[0] if fcmd else ''
                p['app'] = os.path.basename(first) or p['cmd'] or 'unknown'
    except Exception:
        for p in procs2:
            p['full'] = p['cmd']
            p['app'] = p['cmd']

    return {'sys': sys2, 'procs': procs2}


def _print_banner_lines():
    if USE_COLOR:
        w = len(BANNER[0])
        for line in BANNER:
            out = []
            last = ''
            for i, ch in enumerate(line):
                if ch == ' ':
                    out.append(' ')
                    continue
                r, g, b = heat_rgb(i / max(1, w - 1))
                color = rgb(r, g, b)
                if color != last:
                    out.append(color)
                    last = color
                out.append(ch)
            out.append(RESET)
            print(''.join(out))
    else:
        for line in BANNER:
            print(line)


def print_system_overview(args):
    if sys.stderr.isatty():
        print(f"{dim()}Sampling system for {args.sample:.0f}s…{reset()}",
              file=sys.stderr, end='', flush=True)
    snap = collect_system(args.sample)
    if sys.stderr.isatty():
        print(f"\r{' ' * 40}\r", file=sys.stderr, end='', flush=True)
    if not snap:
        print("error: failed to collect system snapshot (top failed?)", file=sys.stderr)
        return 1

    sys_info = snap['sys']
    procs = snap['procs']

    if not args.no_banner:
        _print_banner_lines()
        r1, g1, b1 = heat_rgb(0.05)
        r2, g2, b2 = heat_rgb(0.55)
        sep = f"{rgb(180, 180, 200)} ▸ {reset()}"
        print(f"{bold()}{rgb(r1, g1, b1)}termtree{reset()}{sep}"
              f"{rgb(r2, g2, b2)}system overview{reset()}{sep}"
              f"{dim()}sampled {args.sample:.0f}s{reset()}")
        print()

    ncpu = os.cpu_count() or 1
    cpu_busy = sys_info.get('cpu_busy', 0)
    load = sys_info.get('load', (0, 0, 0))

    mem_used = sys_info.get('mem_used', 0)
    mem_unused = sys_info.get('mem_unused', 0)
    # Prefer sysctl hw.memsize as authoritative — top's "G/M" rounding can lose
    # several hundred MB; using sysctl keeps the percentage honest.
    try:
        mem_total = int(subprocess.run(['sysctl', '-n', 'hw.memsize'],
                                       capture_output=True, text=True,
                                       check=True, timeout=2).stdout.strip())
    except Exception:
        mem_total = mem_used + mem_unused or max(mem_used, 1)
    mem_pct = mem_used / mem_total if mem_total else 0

    disk = disk_context('/System/Volumes/Data') or disk_context('/')

    net_in = sys_info.get('net_in_rate', 0)
    net_out = sys_info.get('net_out_rate', 0)
    disk_r = sys_info.get('disk_read_rate', 0)
    disk_w = sys_info.get('disk_write_rate', 0)

    bw = 22

    print(f"  {bold()}CPU{reset()}     {gradient_bar(cpu_busy / 100, bw)} "
          f"{heat(cpu_busy / 100)}{cpu_busy:5.1f}%{reset()}   "
          f"{dim()}{ncpu} cores · load {load[0]:.2f} {load[1]:.2f} {load[2]:.2f}{reset()}")

    print(f"  {bold()}Memory{reset()}  {gradient_bar(mem_pct, bw)} "
          f"{heat(mem_pct)}{mem_pct * 100:5.1f}%{reset()}   "
          f"{dim()}{fmt_size(mem_used)} used / {fmt_size(mem_total)} · "
          f"{fmt_size(sys_info.get('mem_compressed', 0))} compressed · "
          f"{fmt_size(sys_info.get('mem_wired', 0))} wired{reset()}")

    # Swap — the *honest* pressure metric. Memory % can be 95% on macOS while
    # the system is perfectly fine, as long as swap stays at zero. If you see
    # swap > 0 with active swapins/outs, *now* you have a problem.
    swap = _swap_info()
    if swap is not None:
        s_total, s_used, swap_ins, swap_outs = swap
        s_pct = (s_used / s_total) if s_total else 0
        active = swap_ins > 0 or swap_outs > 0
        if s_total == 0 and not active:
            badge = f"{rgb(*heat_rgb(0.10))}● healthy{reset()} (no swap configured)"
            print(f"  {bold()}Swap{reset()}    {' ' * bw}    {dim()}0 B{reset()}    {badge}")
        else:
            badge = (f"{rgb(*heat_rgb(0.90))}● active{reset()}" if active
                     else f"{rgb(*heat_rgb(0.10))}● idle{reset()}")
            print(f"  {bold()}Swap{reset()}    {gradient_bar(s_pct, bw)} "
                  f"{heat(s_pct)}{s_pct * 100:5.1f}%{reset()}   "
                  f"{dim()}{fmt_size(s_used)} used / {fmt_size(s_total)}{reset()}    {badge}")

    if disk:
        d_total, d_used, d_free = disk
        d_pct = d_used / d_total if d_total else 0
        print(f"  {bold()}Disk{reset()}    {gradient_bar(d_pct, bw)} "
              f"{heat(d_pct)}{d_pct * 100:5.1f}%{reset()}   "
              f"{dim()}{fmt_size(d_used)} used / {fmt_size(d_total)} · "
              f"{fmt_size(d_free)} free{reset()}")

    # Network: scale 10 MB/s = "full"
    NET_REF = 10 * 1024 * 1024
    in_p = min(1.0, net_in / NET_REF)
    out_p = min(1.0, net_out / NET_REF)
    print(f"  {bold()}Net{reset()}     "
          f"↓ {gradient_bar(in_p, 10)} {heat(in_p)}{fmt_rate(net_in):>10}{reset()}   "
          f"↑ {gradient_bar(out_p, 10)} {heat(out_p)}{fmt_rate(net_out):>10}{reset()}")

    # Disk I/O: scale 100 MB/s = "full"
    IO_REF = 100 * 1024 * 1024
    r_p = min(1.0, disk_r / IO_REF)
    w_p = min(1.0, disk_w / IO_REF)
    print(f"  {bold()}I/O{reset()}     "
          f"R {gradient_bar(r_p, 10)} {heat(r_p)}{fmt_rate(disk_r):>10}{reset()}   "
          f"W {gradient_bar(w_p, 10)} {heat(w_p)}{fmt_rate(disk_w):>10}{reset()}")
    print()

    # Aggregate per app
    app = defaultdict(lambda: {'cpu': 0.0, 'mem': 0, 'count': 0})
    for p in procs:
        a = p.get('app', 'unknown')
        app[a]['cpu'] += p['cpu']
        app[a]['mem'] += p['mem']
        app[a]['count'] += 1

    top_cpu = sorted(app.items(), key=lambda kv: -kv[1]['cpu'])[:args.top]
    top_mem = sorted(app.items(), key=lambda kv: -kv[1]['mem'])[:args.top]
    # CPU bar reference: 100% = one core saturated (top's per-process convention).
    # A process using two full cores reads as 200% and the bar caps at full.
    cpu_ref = 100.0

    print(f"  {bold()}Top CPU{reset()}{' ' * 31}{bold()}Top Memory{reset()}")
    print(f"  {dim()}{'─' * 36}   {'─' * 36}{reset()}")
    n_rows = max(len(top_cpu), len(top_mem))
    for i in range(n_rows):
        left_cells = ' ' * 36
        right_cells = ''
        if i < len(top_cpu) and top_cpu[i][1]['cpu'] > 0:
            name, ai = top_cpu[i]
            pct = min(1.0, ai['cpu'] / cpu_ref)
            label = name[:16]
            left_cells = (f"{gradient_bar(pct, 10)} "
                          f"{heat(pct)}{ai['cpu']:6.1f}%{reset()} "
                          f"{label:<16}")
        if i < len(top_mem) and top_mem[i][1]['mem'] > 0:
            name, ai = top_mem[i]
            pct = ai['mem'] / mem_total if mem_total else 0
            label = name[:16]
            right_cells = (f"{gradient_bar(min(1.0, pct), 10)} "
                           f"{heat(min(1.0, pct))}{fmt_size(ai['mem']):>9}{reset()} "
                           f"{label:<16}")
        print(f"  {left_cells}   {right_cells}")

    return 0


# ─────────────────────────────────────────────────────────────────────
# main / argparse
# ─────────────────────────────────────────────────────────────────────

def main():
    global USE_COLOR, USE_TRUECOLOR
    p = argparse.ArgumentParser(
        prog='termtree',
        description="ASCII disk usage visualizer with truecolor heat gradient. "
                    "Use --sys for a system overview (CPU/memory/disk/network).",
    )
    p.add_argument('path', nargs='?', default='.', help="directory (default: .)")
    p.add_argument('-d', '--depth', type=int, default=2, help="tree depth (default: 2)")
    p.add_argument('-n', '--top', type=int, default=10, help="top-N per level (default: 10)")
    p.add_argument('-w', '--width', type=int, default=78, help="treemap width")
    p.add_argument('--height', type=int, default=20, help="treemap height")
    p.add_argument('--no-banner', action='store_true', help="hide ASCII banner")
    p.add_argument('--no-tree', action='store_true')
    p.add_argument('--no-treemap', action='store_true')
    p.add_argument('--no-types', action='store_true')
    p.add_argument('--no-color', action='store_true')
    p.add_argument('-L', '--follow-symlinks', action='store_true')
    p.add_argument('--apparent', action='store_true',
                   help="report logical file size (st_size) instead of allocated "
                        "on-disk blocks. Default is `du`-style allocated bytes — "
                        "important for sparse files like Docker.raw which can show "
                        "100s of GB logical but actually use a fraction.")
    p.add_argument('-S', '--sys', action='store_true',
                   help="system overview mode: CPU, memory, disk, network on one "
                        "screen. Skips disk scan; samples top + ps for ~1s.")
    p.add_argument('--sample', type=float, default=2.0,
                   help="sample window in seconds for --sys rate metrics (default: 2.0)")
    p.add_argument('--watch', type=float, metavar='SECS',
                   help="re-run periodically, clearing screen between (Ctrl+C to exit)")
    p.add_argument('-V', '--version', action='version', version=f'%(prog)s {__version__}')
    args = p.parse_args()

    if args.no_color or not sys.stdout.isatty():
        USE_COLOR = False
    USE_TRUECOLOR = USE_COLOR and supports_truecolor()

    # System overview mode: bypass disk scanning entirely
    if args.sys:
        if args.watch:
            try:
                while True:
                    print('\033[2J\033[H', end='')  # clear screen + home
                    print_system_overview(args)
                    print(f"\n{dim()}Refreshing every {args.watch:.0f}s · Ctrl+C to exit{reset()}")
                    time.sleep(args.watch)
            except KeyboardInterrupt:
                print()
                return 0
        return print_system_overview(args)

    path = os.path.expanduser(args.path)
    if not os.path.exists(path):
        print(f"error: path not found: {path}", file=sys.stderr)
        return 1

    show_progress = sys.stderr.isatty()
    if show_progress:
        print(f"Scanning {path}…", file=sys.stderr, end='', flush=True)

    def progress(c):
        if show_progress:
            print(f"\rScanning {path}… {c:,} files", file=sys.stderr, end='', flush=True)

    root, ext_stats = scan(path, follow_symlinks=args.follow_symlinks,
                           on_progress=progress, apparent=args.apparent)
    if show_progress:
        print(f"\r{' ' * (len(path) + 40)}\r", file=sys.stderr, end='', flush=True)

    if not root:
        print("error: scan failed", file=sys.stderr)
        return 1

    disk_info = disk_context(path)
    if not args.no_banner:
        print_banner(path, root.size, root.file_count, disk_info=disk_info, apparent=args.apparent)
    else:
        mode = 'apparent' if args.apparent else 'on-disk'
        print(f"{bold()}{path}{reset()}  {fmt_size(root.size)} · {root.file_count:,} files · {mode}\n")

    if not args.no_tree:
        print_tree(root, max_depth=args.depth, top_n=args.top)
    if not args.no_types:
        print_extensions(ext_stats, root.size or 1)
    if not args.no_treemap and root.is_dir:
        render_treemap(root, width=args.width, height=args.height)
    return 0


if __name__ == '__main__':
    sys.exit(main())