from typing import Any

import psutil
from psutil._ntuples import sdiskio


def counters() -> dict[str, sdiskio]:
    """Current per-disk IO counters.

    Returns:
        A device-name to sdiskio mapping, or an empty dict on systems
        without disk statistics.
    """
    return psutil.disk_io_counters(perdisk=True) or {}


def rates(prev: dict[str, sdiskio], dt: float) -> tuple[float, float]:
    """Aggregate read/write rates in bytes/s since a previous counters call.

    Disks that were not present in `prev` (hot-plugged) contribute
    nothing, and negative byte deltas (counter wrap, reboots) are clamped
    to zero.

    Args:
        prev: counters() result from the previous sample.
        dt: seconds between the two snapshots.

    Returns:
        (read_bytes_per_s, write_bytes_per_s).
    """
    cur = counters()
    r = 0
    w = 0
    for name, c in cur.items():
        p = prev.get(name)
        if p is not None and dt > 0:
            r += max(0, int(c.read_bytes) - int(p.read_bytes))
            w += max(0, int(c.write_bytes) - int(p.write_bytes))
    return (r / dt if dt > 0 else 0.0, w / dt if dt > 0 else 0.0)


def partitions() -> list[dict[str, Any]]:
    """Mounted real filesystems, grouped by device.

    All partitions on the same device are merged into one entry; usage
    stats come from the first readable mountpoint, unreadable ones are
    skipped. When a device has more than 3 mountpoints, mounts_disp shows
    the first three plus "+N more".

    Returns:
        One entry per device (device, fstype, usage, mounts, mounts_disp),
        sorted by device name.
    """
    groups: dict[str, dict[str, Any]] = {}
    order: list[str] = []
    for p in psutil.disk_partitions(all=False):
        if p.device in groups:
            g = groups[p.device]
            if p.mountpoint not in g["mounts"]:
                g["mounts"].append(p.mountpoint)
            continue
        try:
            u = psutil.disk_usage(p.mountpoint)
        except (OSError, PermissionError):
            continue
        g = {
            "device": p.device,
            "fstype": p.fstype,
            "total": u.total,
            "used": u.used,
            "free": u.free,
            "pct": u.percent,
            "mounts": [p.mountpoint],
        }
        groups[p.device] = g
        order.append(p.device)
    out = [groups[d] for d in sorted(order)]
    for g in out:
        g["mounts"] = sorted(g["mounts"])
        mounts = g["mounts"]
        if len(mounts) > 3:
            g["mounts_disp"] = " · ".join(mounts[:3]) + f" +{len(mounts) - 3} more"
        else:
            g["mounts_disp"] = " · ".join(mounts)
    return out