Skip to content

Instantly share code, notes, and snippets.

@Creased

Creased/emmc_check.py

Last active July 26, 2026 23:06
Show Gist options

  • Select an option

    Learn more about clone URLs
  • Save Creased/b98e02bb0a4904b8023b8a6837977ab9 to your computer and use it in GitHub Desktop.

Select an option

Learn more about clone URLs
Save Creased/b98e02bb0a4904b8023b8a6837977ab9 to your computer and use it in GitHub Desktop.
Download ZIP
eMMC checking script for Nintendo Switch eMMC module
Raw
emmc_check.py
#!/usr/bin/env python3
"""
emmcheck - qualification bench for eMMC modules.
Reads identity and health registers, verifies the partition layout, profiles
read throughput across the whole address space, and isolates unreadable
sectors down to the LBA. Knows the Nintendo Switch eMMC layout.
Read-only unless you pass --write-test.
Requires: python3, rich, root. Optional: mmc-utils (for EXT_CSD health).
"""
from __future__ import annotations
import argparse
import json
import mmap
import os
import random
import shutil
import subprocess
import sys
import time
import zlib
from dataclasses import dataclass, field, asdict
from datetime import datetime, timezone
from pathlib import Path
try:
from rich.align import Align
from rich.console import Console, Group
from rich.panel import Panel
from rich.progress import (
BarColumn, Progress, SpinnerColumn, TextColumn,
TimeElapsedColumn, TimeRemainingColumn,
)
from rich.rule import Rule
from rich.table import Table
from rich.text import Text
except ImportError:
sys.exit("emmcheck needs the 'rich' library.\n pip install rich (or: apt install python3-rich)")
console = Console()
SECTOR = 512
MIB = 1024 * 1024
GIB = 1024 * 1024 * 1024
# ---------------------------------------------------------------------------
# Reference data
# ---------------------------------------------------------------------------
# JEDEC MID values seen in the wild. Unlisted IDs are reported as raw hex.
MANFIDS = {
0x02: "SanDisk",
0x11: "Toshiba / Kioxia",
0x13: "Micron",
0x15: "Samsung",
0x2C: "Kingston",
0x45: "SanDisk",
0x70: "Kingston",
0x88: "Foresee (Longsys)",
0x90: "SK Hynix",
0x9B: "YMTC",
0xD6: "Foresee (Longsys)",
0xFE: "Micron",
}
# Nintendo Switch internal eMMC GPT, in order. Sizes are MiB, approximate.
SWITCH_LAYOUT = [
("PRODINFO", 4),
("PRODINFOF", 4),
("BCPKG2-1-Normal-Main", 8),
("BCPKG2-2-Normal-Sub", 8),
("BCPKG2-3-SafeMode-Main", 8),
("BCPKG2-4-SafeMode-Sub", 8),
("BCPKG2-5-Repair-Main", 8),
("BCPKG2-6-Repair-Sub", 8),
("SAFE", 64),
("SYSTEM", 2560),
("USER", 26000), # varies with total capacity; matched loosely
]
PRE_EOL = {
0x00: ("Not supported", "dim", 0),
0x01: ("Normal - under 80% of reserve blocks used", "green", 0),
0x02: ("WARNING - 80% of reserve blocks used", "yellow", 1),
0x03: ("URGENT - 90% of reserve blocks used", "red", 2),
}
# ---------------------------------------------------------------------------
# Small helpers
# ---------------------------------------------------------------------------
def human(n: float) -> str:
for unit, div in (("GiB", GIB), ("MiB", MIB), ("KiB", 1024)):
if n >= div:
return f"{n / div:.2f} {unit}"
return f"{int(n)} B"
def read_sysfs(path: str) -> str | None:
try:
return Path(path).read_text().strip()
except (OSError, UnicodeDecodeError):
return None
def have(binary: str) -> bool:
return shutil.which(binary) is not None
# ---------------------------------------------------------------------------
# Device discovery
# ---------------------------------------------------------------------------
@dataclass
class Partition:
name: str
label: str | None
start_lba: int
sectors: int
@property
def size(self) -> int:
return self.sectors * SECTOR
def contains(self, byte_offset: int) -> bool:
lo = self.start_lba * SECTOR
return lo <= byte_offset < lo + self.size
@dataclass
class Device:
name: str # mmcblk1
path: str # /dev/mmcblk1
size: int = 0
partitions: list[Partition] = field(default_factory=list)
mounted: list[str] = field(default_factory=list)
is_file: bool = False
# identity
manfid: int | None = None
oemid: int | None = None
model: str | None = None
serial: str | None = None
fwrev: str | None = None
hwrev: str | None = None
date: str | None = None
card_type: str | None = None
cid: str | None = None
csd: str | None = None
# health
life_a: int | None = None
life_b: int | None = None
pre_eol: int | None = None
ext_csd_rev: str | None = None
health_source: str = "none"
# link
bus_width: str | None = None
clock: str | None = None
timing: str | None = None
host: str | None = None
driver: str | None = None
transport: str = "unknown" # usb2 | usb | native
usb_speed: str | None = None
usb_id: str | None = None
@property
def vendor(self) -> str:
if self.manfid is None:
return "unknown"
return MANFIDS.get(self.manfid, f"unrecognised (0x{self.manfid:02x})")
@property
def link_ceiling_mbps(self) -> float | None:
"""Theoretical MB/s of the negotiated MMC bus: clock x width / 8.
On a card reader this is usually a harder limit than USB, because the
reader negotiates plain high-speed 4-bit rather than HS200/HS400.
"""
if not self.clock or not self.bus_width:
return None
try:
hz = float(self.clock.split()[0])
except (ValueError, IndexError):
return None
bits = 1
for token in self.bus_width.replace("(", " ").split():
if token.isdigit() and int(token) in (1, 4, 8):
bits = int(token)
return hz * bits / 8 / 1_000_000
def partition_at(self, byte_offset: int) -> str:
for p in self.partitions:
if p.contains(byte_offset):
return p.label or p.name
return "(unpartitioned)"
def natural_key(name: str) -> tuple:
"""Sort block device names by trailing number, not lexicographically.
Plain sorted() orders mmcblk1p10 immediately after mmcblk1p1, which
silently scrambles the partition table against any expected layout.
"""
head = name.rstrip("0123456789")
tail = name[len(head):]
return (head, int(tail) if tail else -1)
def list_mmc_devices() -> list[Device]:
devs = []
names = [p.name for p in Path("/sys/block").glob("mmcblk*")]
for name in sorted(names, key=natural_key):
# skip the boot0/boot1/rpmb hardware partitions, they are separate nodes
if any(s in name for s in ("boot", "rpmb", "gp")):
continue
devs.append(load_device(f"/dev/{name}"))
return devs
def load_device(path: str) -> Device:
name = os.path.basename(path)
dev = Device(name=name, path=path)
if not Path(f"/sys/block/{name}").exists():
# allow pointing at an image file for offline testing
dev.is_file = True
dev.size = os.path.getsize(path) if os.path.exists(path) else 0
return dev
base = f"/sys/block/{name}"
nsec = read_sysfs(f"{base}/size")
dev.size = int(nsec) * SECTOR if nsec else 0
# partitions, ordered by GPT entry number
part_dirs = sorted(Path(base).glob(f"{name}p*"),
key=lambda p: natural_key(p.name))
for part in part_dirs:
start = read_sysfs(f"{part}/start")
size = read_sysfs(f"{part}/size")
if start is None or size is None:
continue
dev.partitions.append(Partition(part.name, None, int(start), int(size)))
_attach_labels(dev)
# mount check
try:
for line in Path("/proc/mounts").read_text().splitlines():
src = line.split()[0]
if src.startswith(f"/dev/{name}"):
dev.mounted.append(line.split()[1])
except OSError:
pass
# identity from sysfs
d = f"{base}/device"
dev.model = read_sysfs(f"{d}/name")
dev.serial = read_sysfs(f"{d}/serial")
dev.fwrev = read_sysfs(f"{d}/fwrev")
dev.hwrev = read_sysfs(f"{d}/hwrev")
dev.date = read_sysfs(f"{d}/date")
dev.card_type = read_sysfs(f"{d}/type")
dev.cid = read_sysfs(f"{d}/cid")
dev.csd = read_sysfs(f"{d}/csd")
for attr, key in (("manfid", "manfid"), ("oemid", "oemid")):
raw = read_sysfs(f"{d}/{attr}")
if raw:
try:
setattr(dev, key, int(raw, 16))
except ValueError:
pass
# health: sysfs first (cheap), EXT_CSD later (authoritative)
for attr, key in (("life_time", None), ("pre_eol_info", "pre_eol")):
raw = read_sysfs(f"{d}/{attr}")
if not raw:
continue
if attr == "life_time":
parts = raw.split()
try:
dev.life_a = int(parts[0], 16)
dev.life_b = int(parts[1], 16)
dev.health_source = "sysfs"
except (ValueError, IndexError):
pass
else:
try:
dev.pre_eol = int(raw, 16)
dev.health_source = "sysfs"
except ValueError:
pass
# link parameters
try:
real = os.path.realpath(f"{base}/device")
host = None
for part in Path(real).parts:
if part.startswith("mmc") and ":" not in part:
host = part
dev.host = host
if host:
ios = read_sysfs(f"/sys/kernel/debug/{host}/ios")
if ios:
for line in ios.splitlines():
if ":" not in line:
continue
k, v = (x.strip() for x in line.split(":", 1))
kl = k.lower()
if kl == "clock":
dev.clock = v
elif kl == "bus width":
dev.bus_width = v
elif kl == "timing spec":
dev.timing = v
# driver name
drv = os.path.realpath(f"{real}/../driver") if os.path.exists(f"{real}/../driver") else None
if drv:
dev.driver = os.path.basename(drv)
_detect_transport(dev, real)
except OSError:
pass
return dev
def _detect_transport(dev: Device, real_path: str) -> None:
"""Walk up the sysfs tree to find the USB device this card reader sits on.
The negotiated USB speed is the real ceiling on sequential throughput, so
it belongs in the report next to the benchmark numbers.
"""
node = Path(real_path)
for parent in [node] + list(node.parents):
if (parent / "idVendor").exists() and (parent / "speed").exists():
speed = read_sysfs(str(parent / "speed"))
vid = read_sysfs(str(parent / "idVendor"))
pid = read_sysfs(str(parent / "idProduct"))
dev.usb_speed = speed
if vid and pid:
dev.usb_id = f"{vid}:{pid}"
try:
mbit = float(speed) if speed else 0.0
except ValueError:
mbit = 0.0
dev.transport = "usb2" if 0 < mbit <= 480 else "usb"
return
if parent.name == "sys":
break
dev.transport = "native"
def _attach_labels(dev: Device) -> None:
"""Pull GPT partition labels via lsblk if available."""
if not have("lsblk"):
return
try:
out = subprocess.run(
["lsblk", "-J", "-b", "-o", "NAME,PARTLABEL", dev.path],
capture_output=True, text=True, timeout=10,
)
if out.returncode != 0:
return
data = json.loads(out.stdout)
labels = {}
for node in data.get("blockdevices", []):
for child in node.get("children", []) or []:
labels[child.get("name")] = child.get("partlabel") or None
for p in dev.partitions:
p.label = labels.get(p.name)
except (subprocess.SubprocessError, json.JSONDecodeError, OSError):
pass
def read_extcsd(dev: Device) -> None:
"""Authoritative health read via mmc-utils. Sysfs can be a stale snapshot
from card init on older kernels, so prefer this when available."""
if dev.is_file or not have("mmc"):
return
try:
out = subprocess.run(["mmc", "extcsd", "read", dev.path],
capture_output=True, text=True, timeout=30)
if out.returncode != 0:
return
except (subprocess.SubprocessError, OSError):
return
for line in out.stdout.splitlines():
stripped = line.strip()
if stripped.startswith("Extended CSD rev"):
dev.ext_csd_rev = stripped.replace("Extended CSD rev", "").strip()
continue
if ":" not in line:
continue
key, _, val = line.partition(":")
val = val.strip()
try:
num = int(val, 16) if val.startswith("0x") else None
except ValueError:
num = None
if "DEVICE_LIFE_TIME_EST_TYP_A" in key and num is not None:
dev.life_a = num
dev.health_source = "EXT_CSD"
elif "DEVICE_LIFE_TIME_EST_TYP_B" in key and num is not None:
dev.life_b = num
dev.health_source = "EXT_CSD"
elif "PRE_EOL_INFO" in key and num is not None:
dev.pre_eol = num
dev.health_source = "EXT_CSD"
# ---------------------------------------------------------------------------
# eMMC hardware partitions (boot0/boot1/rpmb/gp)
# ---------------------------------------------------------------------------
@dataclass
class HwPartition:
name: str
path: str
size: int
force_ro: bool | None = None
readable: bool = False
errors: list[int] = field(default_factory=list)
mbps: float = 0.0
note: str = ""
def discover_hw_partitions(dev: Device) -> list[HwPartition]:
"""Boot and RPMB areas are separate address spaces with their own nodes.
Nothing in a user-area scan reaches them, so a module can pass a full
surface scan and still hold an unreadable bootloader.
"""
out = []
if dev.is_file:
return out
for suffix in ("boot0", "boot1", "rpmb", "gp0", "gp1", "gp2", "gp3"):
name = f"{dev.name}{suffix}"
base = Path(f"/sys/block/{name}")
if not base.exists():
continue
nsec = read_sysfs(f"{base}/size")
size = int(nsec) * SECTOR if nsec else 0
ro = read_sysfs(f"{base}/force_ro")
out.append(HwPartition(
name=suffix, path=f"/dev/{name}", size=size,
force_ro=(ro == "1") if ro is not None else None))
return out
def scan_hw_partitions(parts: list[HwPartition], progress: Progress,
task) -> None:
"""Read each hardware partition end to end. They are a few MiB at most."""
for p in parts:
if p.size == 0:
p.note = "zero length"
progress.update(task, advance=1)
continue
if p.name == "rpmb":
# RPMB is not a plain block device; it answers only to the
# authenticated protocol, so a raw read proves nothing either way
p.note = "authenticated access only, not raw-readable"
progress.update(task, advance=1)
continue
chunk = min(MIB, p.size)
total, elapsed = 0, 0.0
try:
with RawReader(p.path, chunk) as r:
for off in range(0, (p.size // chunk) * chunk, chunk):
t0 = time.perf_counter()
try:
n, _ = r.read(off, chunk)
elapsed += time.perf_counter() - t0
total += n
except OSError:
p.errors.extend(isolate_bad_sectors(r, off, chunk))
p.readable = not p.errors
p.mbps = (total / MIB / elapsed) if elapsed > 0 else 0.0
except OSError as e:
p.note = f"could not open ({e.strerror or e})"
progress.update(task, advance=1)
def hw_partition_panel(dev: Device, parts: list[HwPartition],
is_switch: bool) -> Panel | None:
if not parts:
return None
t = Table(box=None, padding=(0, 2), show_edge=False)
t.add_column("area")
t.add_column("size", justify="right")
t.add_column("write", justify="left")
t.add_column("read", justify="right")
t.add_column("")
problems = 0
for p in parts:
wp = "-" if p.force_ro is None else ("protected" if p.force_ro else
"[yellow]writable[/]")
if p.errors:
status, rate = f"[red]{len(p.errors)} bad sector(s)[/]", "-"
problems += 1
elif p.readable:
status, rate = "[green]ok[/]", f"{p.mbps:.1f} MB/s"
else:
status, rate = f"[dim]{p.note or 'not tested'}[/]", "-"
t.add_row(p.name, human(p.size) if p.size else "-", wp, rate, status)
body: list = [t]
if is_switch:
boots = {p.name: p for p in parts if p.name in ("boot0", "boot1")}
missing = [b for b in ("boot0", "boot1") if b not in boots]
if missing:
body.append(Text())
body.append(Text(f" • {', '.join(missing)} absent - a Switch module "
"should expose both boot areas", style="yellow"))
for name, p in boots.items():
if p.size and abs(p.size - 4 * MIB) > 64 * 1024:
body.append(Text())
body.append(Text(f" • {name} is {human(p.size)}, not the "
"expected 4 MiB", style="yellow"))
border = "red" if problems else "green"
return Panel(Group(*body), title="[bold]hardware partitions[/]",
border_style=border)
# ---------------------------------------------------------------------------
# GPT
# ---------------------------------------------------------------------------
GPT_SIG = b"EFI PART"
@dataclass
class GptInfo:
present: bool = False
primary_ok: bool = False
backup_ok: bool = False
disk_guid: str | None = None
backup_disk_guid: str | None = None
names: dict[int, str] = field(default_factory=dict) # first_lba -> name
problems: list[str] = field(default_factory=list)
def _guid(raw: bytes) -> str:
"""Format a GPT GUID: first three fields little-endian, last two big."""
a = int.from_bytes(raw[0:4], "little")
b = int.from_bytes(raw[4:6], "little")
c = int.from_bytes(raw[6:8], "little")
return (f"{a:08X}-{b:04X}-{c:04X}-"
f"{raw[8:10].hex().upper()}-{raw[10:16].hex().upper()}")
def _parse_gpt_header(buf: bytes) -> tuple[dict, bool] | None:
if len(buf) < 92 or buf[:8] != GPT_SIG:
return None
hdr_size = int.from_bytes(buf[12:16], "little")
if not (92 <= hdr_size <= len(buf)):
return None
stored = int.from_bytes(buf[16:20], "little")
tmp = bytearray(buf[:hdr_size])
tmp[16:20] = b"\x00\x00\x00\x00"
crc_ok = (zlib.crc32(bytes(tmp)) & 0xFFFFFFFF) == stored
return ({
"my_lba": int.from_bytes(buf[24:32], "little"),
"alt_lba": int.from_bytes(buf[32:40], "little"),
"first_usable": int.from_bytes(buf[40:48], "little"),
"last_usable": int.from_bytes(buf[48:56], "little"),
"disk_guid": _guid(buf[56:72]),
"entries_lba": int.from_bytes(buf[72:80], "little"),
"entry_count": int.from_bytes(buf[80:84], "little"),
"entry_size": int.from_bytes(buf[84:88], "little"),
"entries_crc": int.from_bytes(buf[88:92], "little"),
}, crc_ok)
def parse_gpt(path: str, size: int) -> GptInfo:
"""Read the GPT from the device without going through libblkid.
libblkid refuses to report partition labels when a GPT fails its
validation, so a disagreement between the primary and backup copies
silently costs us the partition names. Reading it directly keeps the
names and turns the disagreement into a reported finding.
"""
info = GptInfo()
try:
fd = os.open(path, os.O_RDONLY)
except OSError:
return info
def pread(off: int, length: int) -> bytes:
try:
return os.pread(fd, length, off)
except OSError:
return b""
try:
head = _parse_gpt_header(pread(SECTOR, SECTOR))
if head is None:
return info
hdr, crc_ok = head
info.present = True
info.primary_ok = crc_ok
info.disk_guid = hdr["disk_guid"]
if not crc_ok:
info.problems.append("primary GPT header CRC does not match")
count = min(hdr["entry_count"], 256)
esize = hdr["entry_size"]
if 0 < esize <= 1024 and count:
blob = pread(hdr["entries_lba"] * SECTOR, count * esize)
if len(blob) == count * esize:
if (zlib.crc32(blob) & 0xFFFFFFFF) != hdr["entries_crc"]:
info.problems.append("partition entry array CRC does not match")
for i in range(count):
e = blob[i * esize:(i + 1) * esize]
if e[0:16] == b"\x00" * 16:
continue
first = int.from_bytes(e[32:40], "little")
name = e[56:128].decode("utf-16-le", "replace").rstrip("\x00")
if name:
info.names[first] = name
else:
info.problems.append("could not read the partition entry array")
# backup header sits in the last sector
last_lba = size // SECTOR - 1
back = _parse_gpt_header(pread(last_lba * SECTOR, SECTOR))
if back is None:
info.problems.append(
"backup GPT missing or unreadable at the end of the device")
else:
bhdr, bcrc = back
info.backup_ok = bcrc
info.backup_disk_guid = bhdr["disk_guid"]
if not bcrc:
info.problems.append("backup GPT header CRC does not match")
if bhdr["disk_guid"] != hdr["disk_guid"]:
info.problems.append(
"primary and backup GPT report different disk GUIDs - the "
"two copies came from different images")
if hdr["alt_lba"] != last_lba:
info.problems.append(
f"primary GPT expects its backup at LBA {hdr['alt_lba']:,} "
f"but the device ends at {last_lba:,}")
finally:
os.close(fd)
return info
def load_history(path: str) -> list[dict]:
"""Prior runs from the JSON log, for cross-module and repeat-test checks."""
records = []
try:
with open(path) as fh:
for line in fh:
line = line.strip()
if line:
try:
records.append(json.loads(line))
except json.JSONDecodeError:
continue
except OSError:
pass
return records
def history_findings(dev: Device, gpt: GptInfo,
history: list[dict]) -> list[str]:
"""Compare this module against everything the bench has seen before.
Two things matter across a batch: whether this exact part has been tested
before (so wear can be tracked over time), and whether its partition table
is a clone of another part's (so duplicated console data gets noticed
before the module is deployed).
"""
out: list[str] = []
if not history:
return out
serial = dev.serial
if serial:
prior = [r for r in history
if (r.get("device") or {}).get("serial") == serial]
if prior:
last = prior[-1]
when = (last.get("timestamp") or "")[:10]
note = f"tested {len(prior)} time(s) before, most recently {when} " \
f"({last.get('verdict', '?')})"
prev_dev = last.get("device") or {}
for label, key in (("SLC", "life_a"), ("MLC/TLC", "life_b")):
was, now = prev_dev.get(key), getattr(dev, key)
if was and now and now > was:
note += f"; {label} wear rose 0x{was:02x} to 0x{now:02x}"
out.append(note)
if gpt.disk_guid:
twins = {(r.get("device") or {}).get("serial")
for r in history
if r.get("gpt_disk_guid") == gpt.disk_guid}
twins.discard(serial)
twins.discard(None)
if twins:
out.append(
f"partition table is identical to {len(twins)} other module(s) "
f"on this bench - all were written from the same image and carry "
f"the same console data")
return out
def gpt_panel(gpt: GptInfo) -> Panel | None:
if not gpt.present:
return None
t = Table.grid(padding=(0, 2))
t.add_column(style="dim", justify="right")
t.add_column()
t.add_row("primary", "[green]valid[/]" if gpt.primary_ok else "[red]bad CRC[/]")
t.add_row("backup", "[green]valid[/]" if gpt.backup_ok else "[red]bad or missing[/]")
if gpt.disk_guid:
t.add_row("disk GUID", gpt.disk_guid)
if gpt.backup_disk_guid and gpt.backup_disk_guid != gpt.disk_guid:
t.add_row("backup GUID", f"[yellow]{gpt.backup_disk_guid}[/]")
body: list = [t]
if gpt.problems:
body.append(Text())
for p in gpt.problems:
body.append(Text(f" • {p}", style="yellow"))
return Panel(Group(*body), title="[bold]partition table[/]",
border_style="yellow" if gpt.problems else "green")
# ---------------------------------------------------------------------------
# Kernel log correlation
# ---------------------------------------------------------------------------
# What the mmc/usb stacks say when a transfer was retried or the link reset.
# These mean the problem is on the wire, not in the flash.
LINK_ERROR_HINTS = (
"error -110", "error -84", "error -71", "error -32", "error -5",
"timeout", "timed out", "crc err", "data crc", "response crc",
"reset high-speed", "reset full-speed", "reset superspeed",
"device descriptor read", "cmd response", "req failed",
"tuning failed", "retrying", "controller never released",
"card is busy", "recovery", "reset usb",
)
# These mean the media itself could not return the data.
MEDIA_ERROR_HINTS = (
"critical medium error", "unrecovered read error", "uncorrectable",
"buffer i/o error", "i/o error, dev", "ecc error",
)
def capture_dmesg() -> list[str]:
"""Snapshot the kernel ring buffer so it can be diffed after the scan."""
if not have("dmesg"):
return []
try:
out = subprocess.run(["dmesg", "-t"], capture_output=True,
text=True, timeout=15)
return out.stdout.splitlines() if out.returncode == 0 else []
except (subprocess.SubprocessError, OSError):
return []
def classify_kernel_line(raw: str) -> str | None:
"""Return 'media', 'link' or None for a kernel log line."""
low = raw.lower()
if not any(t in low for t in ("mmc", "usb", "rtsx", "blk_update", "i/o error")):
return None
if any(h in low for h in MEDIA_ERROR_HINTS):
return "media"
if any(h in low for h in LINK_ERROR_HINTS):
return "link"
return None
def dmesg_delta(before: list[str]) -> tuple[list[str], int, int]:
"""New kernel lines since `before`, split into link and media faults.
This is what separates 'the module is failing' from 'the reader is
dropping transfers': the flash controller and the USB bridge complain
in completely different vocabularies.
"""
after = capture_dmesg()
if not after:
return [], 0, 0
new = after[len(before):]
if before:
# the ring buffer may have wrapped; re-anchor on the last known line
try:
anchor = len(after) - 1 - after[::-1].index(before[-1])
new = after[anchor + 1:]
except ValueError:
new = after[-500:]
lines, link, media = [], 0, 0
for raw in new:
kind = classify_kernel_line(raw)
if kind is None:
continue
if kind == "media":
media += 1
else:
link += 1
lines.append(raw.strip())
return lines, link, media
# ---------------------------------------------------------------------------
# I/O engine
# ---------------------------------------------------------------------------
class RawReader:
"""Direct-I/O reader with a page-aligned buffer, falling back to buffered
reads with cache eviction if O_DIRECT is refused."""
def __init__(self, path: str, chunk: int, write: bool = False):
self.path = path
self.chunk = chunk
self.direct = True
mode = os.O_RDWR if write else os.O_RDONLY
try:
self.fd = os.open(path, mode | os.O_DIRECT)
except OSError:
self.fd = os.open(path, mode)
self.direct = False
self._buf = mmap.mmap(-1, chunk)
self._mv = memoryview(self._buf)
def read(self, offset: int, length: int) -> tuple[int, bytes]:
n = os.preadv(self.fd, [self._mv[:length]], offset)
if not self.direct:
os.posix_fadvise(self.fd, offset, length, os.POSIX_FADV_DONTNEED)
return n, bytes(self._mv[:n])
def write(self, offset: int, data: bytes) -> int:
self._mv[:len(data)] = data
n = os.pwritev(self.fd, [self._mv[:len(data)]], offset)
return n
def sync(self) -> None:
try:
os.fsync(self.fd)
except OSError:
pass
def close(self) -> None:
self._mv.release()
self._buf.close()
os.close(self.fd)
def __enter__(self):
return self
def __exit__(self, *a):
self.close()
def isolate_bad_sectors(reader: RawReader, offset: int, length: int,
limit: int = 64) -> list[int]:
"""Re-read a failed chunk sector by sector to find the exact bad LBAs."""
bad = []
step = SECTOR
for pos in range(offset, offset + length, step):
try:
reader.read(pos, step)
except OSError:
bad.append(pos // SECTOR)
if len(bad) >= limit:
break
return bad
# ---------------------------------------------------------------------------
# Tests
# ---------------------------------------------------------------------------
@dataclass
class Sample:
offset: int
mbps: float
error: bool = False
warmup: bool = False # first bucket: link negotiation skews it, ignore
@dataclass
class Results:
samples: list[Sample] = field(default_factory=list)
bad_lbas: list[int] = field(default_factory=list)
error_regions: list[tuple[int, str]] = field(default_factory=list)
seq_mbps: float = 0.0
rand_iops: float = 0.0
rand_mbps: float = 0.0
bytes_read: int = 0
scan_seconds: float = 0.0
write_verified: int = 0
write_mismatches: list[int] = field(default_factory=list)
write_mbps: float = 0.0
kernel_lines: list[str] = field(default_factory=list)
link_errors: int = 0
media_errors: int = 0
direct_io: bool = True
sweep: list[tuple[int, float]] = field(default_factory=list)
cmd_overhead_ms: float | None = None
asymptotic_mbps: float | None = None
knee_bytes: int | None = None
peak_sweep_mbps: float | None = None
sweep_cliff: bool = False
chunk_used: int = 0
def plan_offsets(dev: Device, chunk: int, full: bool,
points: int) -> tuple[list[int], int]:
"""Decide which offsets to read and how wide a display bucket is.
Full mode walks every chunk and buckets the timings for the chart.
Sample mode reads `points` evenly spaced chunks.
"""
usable = (dev.size // chunk) * chunk
if usable == 0:
return [], chunk
if full:
return list(range(0, usable, chunk)), max(chunk, usable // max(1, points))
stride = max(chunk, (usable // max(1, points) // chunk) * chunk)
return list(range(0, usable, stride)), stride
def sequential_profile(dev: Device, reader: RawReader, results: Results,
chunk: int, offsets: list[int], bucket_size: int,
progress: Progress, task) -> None:
"""Read across the address space, recording throughput per position."""
if not offsets:
return
bucket_bytes = 0
bucket_time = 0.0
bucket_index = 0
bucket_error = False
total_time = 0.0
total_bytes = 0
for off in offsets:
t0 = time.perf_counter()
failed = False
try:
n, _ = reader.read(off, chunk)
except OSError:
failed = True
n = 0
dt = time.perf_counter() - t0
if failed:
bucket_error = True
bad = isolate_bad_sectors(reader, off, chunk)
results.bad_lbas.extend(bad)
results.error_regions.append((off, dev.partition_at(off)))
else:
bucket_bytes += n
bucket_time += dt
total_bytes += n
total_time += dt
progress.update(task, advance=chunk)
if off // bucket_size != bucket_index:
mbps = (bucket_bytes / MIB / bucket_time) if bucket_time > 0 else 0.0
results.samples.append(
Sample(bucket_index * bucket_size, mbps, bucket_error,
warmup=(len(results.samples) == 0)))
bucket_index = off // bucket_size
bucket_bytes, bucket_time, bucket_error = 0, 0.0, False
if bucket_bytes or bucket_error:
mbps = (bucket_bytes / MIB / bucket_time) if bucket_time > 0 else 0.0
results.samples.append(Sample(bucket_index * bucket_size, mbps, bucket_error))
results.bytes_read = total_bytes
results.scan_seconds = total_time
results.seq_mbps = (total_bytes / MIB / total_time) if total_time else 0.0
def random_read_test(dev: Device, reader4k: RawReader, results: Results,
seconds: float, progress: Progress, task) -> None:
"""4 KiB random reads. Not USB-bandwidth-limited, so this is the number
that actually characterises the flash rather than the reader."""
size = dev.size
if size < 64 * MIB:
return
blocks = size // 4096
rng = random.Random(0xC0FFEE)
count = 0
start = time.perf_counter()
deadline = start + seconds
while True:
now = time.perf_counter()
if now >= deadline:
break
off = rng.randrange(blocks) * 4096
try:
reader4k.read(off, 4096)
count += 1
except OSError:
results.error_regions.append((off, dev.partition_at(off)))
if count % 64 == 0:
progress.update(task, completed=min(seconds, now - start))
elapsed = time.perf_counter() - start
progress.update(task, completed=seconds)
results.rand_iops = count / elapsed if elapsed else 0.0
results.rand_mbps = results.rand_iops * 4096 / MIB
def write_verify_test(dev: Device, results: Results, chunk: int,
offsets: list[int], progress: Progress, task) -> None:
"""DESTRUCTIVE. Writes a position-derived pattern and reads it back.
Catches write-side faults and fake-capacity modules, which cheap
replacement eMMC is prone to: a module that lies about its size will
wrap writes onto earlier addresses and fail verification in the
upper range.
"""
with RawReader(dev.path, chunk, write=True) as rw:
total_bytes, total_time = 0, 0.0
for off in offsets:
rng = random.Random(off)
pattern = rng.randbytes(chunk)
t0 = time.perf_counter()
try:
rw.write(off, pattern)
except OSError:
results.write_mismatches.append(off)
progress.update(task, advance=1)
continue
total_time += time.perf_counter() - t0
total_bytes += chunk
progress.update(task, advance=1)
rw.sync()
# read back in a second pass so the controller cache cannot mask a failure
with RawReader(dev.path, chunk) as ro:
for off in offsets:
if off in results.write_mismatches:
continue
expect = random.Random(off).randbytes(chunk)
try:
_, got = ro.read(off, chunk)
except OSError:
results.write_mismatches.append(off)
continue
if got != expect:
results.write_mismatches.append(off)
else:
results.write_verified += 1
progress.update(task, advance=1)
results.write_mbps = (total_bytes / MIB / total_time) if total_time else 0.0
# ---------------------------------------------------------------------------
# Rendering
# ---------------------------------------------------------------------------
BLOCKS = "▁▂▃▄▅▆▇█"
def profile_chart(dev: Device, results: Results, width: int | None = None) -> Panel:
samples = results.samples
if not samples:
return Panel("No samples collected.", title="read profile", border_style="dim")
# panel borders and padding eat 4 columns; never exceed the terminal
if width is None:
width = max(20, min(160, console.width - 6))
# resample to the target width
if len(samples) > width:
step = len(samples) / width
buckets = []
for i in range(width):
lo, hi = int(i * step), max(int(i * step) + 1, int((i + 1) * step))
group = samples[lo:hi]
if not group:
continue
buckets.append(Sample(
group[0].offset,
sum(s.mbps for s in group) / len(group),
any(s.error for s in group),
))
else:
buckets = samples
good = [s.mbps for s in buckets if not s.error and s.mbps > 0]
if not good:
return Panel("All reads failed.", title="read profile", border_style="red")
peak = max(good)
median = sorted(good)[len(good) // 2]
bar = Text()
for s in buckets:
if s.error:
bar.append("✗", style="bold white on red")
continue
level = min(7, max(0, int(s.mbps / peak * 7.99)))
ratio = s.mbps / median if median else 1.0
if ratio < 0.55:
style = "bold red"
elif ratio < 0.85:
style = "yellow"
else:
style = "green"
bar.append(BLOCKS[level], style=style)
label_l, label_r = "0", human(dev.size)
pad = len(bar) - len(label_l) - len(label_r)
axis = Text(no_wrap=True, overflow="crop")
axis.append(label_l, style="dim")
axis.append(" " * max(1, pad), style="dim")
axis.append(label_r, style="dim")
legend = Text(no_wrap=True, overflow="ellipsis")
legend.append("█", style="green")
legend.append(" at speed ", style="dim")
legend.append("█", style="yellow")
legend.append(" 55-85% of median ", style="dim")
legend.append("█", style="bold red")
legend.append(" under 55% ", style="dim")
legend.append("✗", style="bold white on red")
legend.append(" read error", style="dim")
stats = Text(
f"peak {peak:.1f} MB/s median {median:.1f} MB/s "
f"slowest {min(good):.1f} MB/s",
style="dim", no_wrap=True, overflow="ellipsis",
)
bar.no_wrap = True
bar.overflow = "crop"
return Panel(Group(bar, axis, Text(), stats, legend),
title="[bold]read throughput across the address space[/]",
border_style="cyan")
def identity_panel(dev: Device) -> Panel:
t = Table.grid(padding=(0, 2))
t.add_column(style="dim", justify="right")
t.add_column()
t.add_row("device", f"[bold]{dev.path}[/]")
t.add_row("capacity", f"{human(dev.size)} ({dev.size:,} bytes)")
t.add_row("vendor", dev.vendor)
if dev.model:
t.add_row("model", f"[bold]{dev.model}[/]")
if dev.serial:
t.add_row("serial", dev.serial)
if dev.date:
t.add_row("mfg date", dev.date)
rev = " / ".join(x for x in (dev.hwrev, dev.fwrev) if x)
if rev:
t.add_row("hw / fw rev", rev)
if dev.card_type:
t.add_row("type", dev.card_type)
link = " · ".join(x for x in (dev.clock, dev.bus_width, dev.timing) if x)
if link:
t.add_row("mmc link", link)
if dev.transport in ("usb", "usb2"):
speed = f"{dev.usb_speed} Mbit/s" if dev.usb_speed else "unknown speed"
gen = "USB 2.0" if dev.transport == "usb2" else "USB 3.x"
extra = f" [dim]{dev.usb_id}[/]" if dev.usb_id else ""
t.add_row("host link", f"{gen} · {speed}{extra}")
if dev.driver:
t.add_row("driver", dev.driver)
return Panel(t, title="[bold]identity[/]", border_style="blue")
def health_panel(dev: Device) -> tuple[Panel, int]:
"""Returns the panel and a severity score: 0 ok, 1 warn, 2 fail."""
t = Table.grid(padding=(0, 2))
t.add_column(style="dim", justify="right")
t.add_column()
severity = 0
if dev.pre_eol is None and dev.life_a is None:
t.add_row("status", "[dim]not reported - eMMC 5.0 or newer required[/]")
if not have("mmc"):
t.add_row("", "[dim]install mmc-utils for a second opinion[/]")
return Panel(t, title="[bold]health[/]", border_style="dim"), 0
if dev.pre_eol is not None:
label, style, sev = PRE_EOL.get(
dev.pre_eol, (f"undefined value 0x{dev.pre_eol:02x}", "yellow", 1))
severity = max(severity, sev)
t.add_row("reserve blocks", f"[{style}]{label}[/]")
for name, val, kind in (("wear (SLC)", dev.life_a, "A"),
("wear (MLC/TLC)", dev.life_b, "B")):
if val is None:
continue
if val == 0:
t.add_row(name, "[dim]not reported[/]")
continue
if val >= 0x0B:
style, sev, txt = "red", 2, "past rated endurance"
else:
lo, hi = (val - 1) * 10, val * 10
txt = f"{lo}-{hi}% consumed"
if val >= 0x09:
style, sev = "red", 2
elif val >= 0x08:
style, sev = "yellow", 1
else:
style, sev = "green", 0
severity = max(severity, sev)
gauge = "█" * min(10, val) + "░" * max(0, 10 - val)
t.add_row(name, f"[{style}]{gauge}[/] {txt} [dim](0x{val:02x})[/]")
t.add_row("source", f"[dim]{dev.health_source}[/]")
border = {0: "green", 1: "yellow", 2: "red"}[severity]
return Panel(t, title="[bold]health[/]", border_style=border), severity
def layout_panel(dev: Device) -> tuple[Panel, bool, int]:
"""Verify the partition table. Returns (panel, is_switch, mismatch_count)."""
if not dev.partitions:
return (Panel("[dim]No partition table found - blank or raw module.[/]",
title="[bold]layout[/]", border_style="dim"), False, 0)
labels = [(p.label or "") for p in dev.partitions]
expected_labels = [n for n, _ in SWITCH_LAYOUT]
label_hits = sum(1 for l in labels if l in expected_labels)
size_shape = [round(p.size / MIB) for p in dev.partitions]
expected_shape = [s for _, s in SWITCH_LAYOUT[:-1]]
shape_match = size_shape[:len(expected_shape)] == expected_shape
is_switch = label_hits >= 6 or (len(dev.partitions) == 11 and shape_match)
t = Table(box=None, padding=(0, 2), show_edge=False)
t.add_column("#", style="dim", justify="right")
t.add_column("partition")
t.add_column("size", justify="right")
t.add_column("expected", justify="right", style="dim")
t.add_column("", justify="left")
mismatches = 0
for i, p in enumerate(dev.partitions):
name = p.label or p.name
size_mib = p.size / MIB
exp = ""
mark = ""
if is_switch and i < len(SWITCH_LAYOUT):
exp_name, exp_mib = SWITCH_LAYOUT[i]
exp = f"{exp_mib} MiB" if i < len(SWITCH_LAYOUT) - 1 else "remainder"
name_ok = (p.label is None) or (p.label == exp_name)
# last partition (USER) absorbs whatever capacity is left
size_ok = (i == len(SWITCH_LAYOUT) - 1) or abs(size_mib - exp_mib) <= 1
if name_ok and size_ok:
mark = "[green]ok[/]"
else:
mark = "[yellow]differs[/]"
mismatches += 1
t.add_row(str(i + 1), name, f"{size_mib:,.0f} MiB", exp, mark)
if is_switch:
title = "[bold]layout[/] [green]· Nintendo Switch eMMC recognised[/]"
border = "green" if mismatches == 0 else "yellow"
if len(dev.partitions) != 11:
border = "yellow"
mismatches += 1
else:
title = "[bold]layout[/] [dim]· not a Switch layout[/]"
border = "blue"
return Panel(t, title=title, border_style=border), is_switch, mismatches
def errors_panel(dev: Device, results: Results) -> Panel | None:
if not results.error_regions and not results.write_mismatches:
return None
t = Table(box=None, padding=(0, 2), show_edge=False)
t.add_column("offset", justify="right")
t.add_column("LBA", justify="right")
t.add_column("partition")
t.add_column("kind")
shown = 0
for off, part in results.error_regions[:20]:
t.add_row(f"0x{off:012x}", f"{off // SECTOR:,}", part, "[red]read error[/]")
shown += 1
for off in results.write_mismatches[:20]:
t.add_row(f"0x{off:012x}", f"{off // SECTOR:,}",
dev.partition_at(off), "[red]write/verify mismatch[/]")
shown += 1
extra = len(results.error_regions) + len(results.write_mismatches) - shown
body: list = [t]
if extra > 0:
body.append(Text(f" ... and {extra} more", style="dim"))
if results.bad_lbas:
preview = ", ".join(str(x) for x in results.bad_lbas[:12])
more = "" if len(results.bad_lbas) <= 12 else f" (+{len(results.bad_lbas) - 12})"
body.append(Text())
body.append(Text(f" unreadable sectors: {preview}{more}", style="red"))
return Panel(Group(*body), title="[bold red]faults[/]", border_style="red")
def kernel_panel(results: Results) -> Panel | None:
if not results.kernel_lines:
return None
summary = []
if results.link_errors:
summary.append(f"[yellow]{results.link_errors} link-level[/]")
if results.media_errors:
summary.append(f"[red]{results.media_errors} media-level[/]")
body: list = [
Text.from_markup(" " + " · ".join(summary) +
" message(s) logged during the scan"),
Text(),
]
for line in results.kernel_lines[:12]:
body.append(Text(f" {line[:160]}", style="dim", overflow="ellipsis"))
if len(results.kernel_lines) > 12:
body.append(Text(f" ... and {len(results.kernel_lines) - 12} more",
style="dim"))
border = "red" if results.media_errors else "yellow"
return Panel(Group(*body), title="[bold]kernel log[/]", border_style=border)
def diagnose_link(dev: Device, results: Results, health_sev: int) -> str | None:
"""Decide whether poor numbers indict the module or the measurement path.
A module doing heavy ECC correction wears its reserve pool down, so
pristine health registers alongside poor throughput is contradictory -
and the contradiction resolves toward the reader, not the flash.
"""
if results.media_errors or results.error_regions:
return None # real media faults, not a link story
if dev.health_source == "none" or health_sev != 0:
return None # can't use registers as an alibi
ceiling = dev.link_ceiling_mbps
slow = bool(ceiling and results.seq_mbps and results.seq_mbps < ceiling * 0.4)
scored = sorted(s.mbps for s in results.samples
if not s.error and not s.warmup and s.mbps > 0)
erratic = False
if len(scored) >= 8:
median = scored[len(scored) // 2]
p25 = scored[len(scored) // 4]
erratic = bool(median and p25 < median * 0.55)
# eMMC random 4K reads land in the thousands of IOPS; low hundreds means
# per-transaction cost is dominating, which is a bus property
starved = bool(0 < results.rand_iops < 500)
if not (slow or erratic or starved):
return None
if results.link_errors:
return (f"the kernel logged {results.link_errors} link-level error(s) "
"during the scan while the module's own health registers stayed "
"clean - the fault is in the reader, adapter or wiring, not the "
"flash")
if sum([slow, erratic, starved]) >= 2:
return ("throughput is poor and erratic but the module reports a clean "
"reserve pool and no wear - heavy ECC correction would have "
"consumed spare blocks, so suspect the reader, adapter seating "
"or signal integrity before the module")
return None
# Transfer sizes for the latency sweep. Spread over two orders of magnitude
# so a straight-line fit can separate fixed cost from marginal cost.
SWEEP_SIZES = (64 * 1024, 256 * 1024, MIB, 4 * MIB, 16 * MIB)
def warm_up(dev: Device, seconds: float = 2.0) -> None:
"""Read and discard until the link settles.
rtsx readers use runtime power management, so the first reads after probe
pay a wake-and-settle cost. Measuring small transfers in that window makes
the sweep report a fixed per-command cost that does not exist once the
device is awake.
"""
size = 4 * MIB
if dev.size < 8 * MIB:
return
span = max(1, (dev.size - size) // size)
try:
with RawReader(dev.path, size) as r:
deadline = time.perf_counter() + seconds
i = 0
while time.perf_counter() < deadline:
try:
r.read((i % span) * size, size)
except OSError:
return
i += 1
except OSError:
return
def latency_sweep(dev: Device, results: Results,
progress: Progress, task) -> None:
"""Time reads at several transfer sizes and fit t = overhead + bytes/rate.
This settles the question the throughput number cannot: a link paying a
large fixed cost per command looks identical to slow flash at one block
size, but the two diverge as the block grows. A big intercept means the
reader is the problem and larger reads will fix it; a low slope means the
path genuinely cannot move data faster.
"""
if dev.size < 128 * MIB:
return
rng = random.Random(0x5EED)
points: list[tuple[int, float]] = []
for size in SWEEP_SIZES:
if size > dev.size // 8:
continue
# small transfers need more samples: their per-read time is short
# enough that scheduler noise can move the median several ms
reps = max(5, min(40, (24 * MIB) // size))
span = (dev.size - size) // size
times = []
try:
with RawReader(dev.path, size) as r:
# two discarded reads per size: the transition between block
# sizes can itself cost a request or two to stabilise
for _ in range(2):
try:
r.read(rng.randrange(span) * size, size)
except OSError:
pass
for _ in range(reps):
off = rng.randrange(span) * size
t0 = time.perf_counter()
try:
r.read(off, size)
except OSError:
continue
times.append(time.perf_counter() - t0)
progress.update(task, advance=1)
except OSError:
continue
if times:
times.sort()
points.append((size, times[len(times) // 2])) # median resists outliers
results.sweep = points
if len(points) < 3:
return
# Find the knee before fitting. A marginal link runs at full rate up to
# some transfer size and then collapses as retries start dominating; a
# straight line through that cliff describes neither regime.
rates = [(size, (size / MIB) / secs) for size, secs in points if secs > 0]
if rates:
best = max(r for _, r in rates)
healthy = [size for size, r in rates if r >= best * 0.85]
results.knee_bytes = max(healthy) if healthy else rates[0][0]
results.peak_sweep_mbps = best
worst_large = min(r for size, r in rates if size >= results.knee_bytes)
results.sweep_cliff = worst_large < best * 0.6
# fit only the healthy regime so the numbers mean something
points = [p for p in points if p[0] <= results.knee_bytes]
if len(points) < 3:
return
n = len(points)
mx = sum(p[0] for p in points) / n
my = sum(p[1] for p in points) / n
denom = sum((p[0] - mx) ** 2 for p in points)
if denom <= 0:
return
slope = sum((p[0] - mx) * (p[1] - my) for p in points) / denom
intercept = my - slope * mx
results.cmd_overhead_ms = max(0.0, intercept * 1000)
if slope > 0:
results.asymptotic_mbps = (1.0 / slope) / 1_000_000
def sweep_panel(dev: Device, results: Results) -> Panel | None:
if not results.sweep:
return None
t = Table(box=None, padding=(0, 2), show_edge=False)
t.add_column("transfer", justify="right")
t.add_column("median time", justify="right")
t.add_column("effective", justify="right")
t.add_column("")
best = 0.0
for size, secs in results.sweep:
mbps = (size / MIB) / secs if secs > 0 else 0.0
best = max(best, mbps)
for size, secs in results.sweep:
mbps = (size / MIB) / secs if secs > 0 else 0.0
width = int(round(mbps / best * 24)) if best else 0
t.add_row(human(size), f"{secs * 1000:.1f} ms", f"{mbps:.1f} MB/s",
"[cyan]" + "▇" * max(1, width) + "[/]")
body: list = [t]
if results.cmd_overhead_ms is not None and results.asymptotic_mbps:
body.append(Text())
g = Table.grid(padding=(0, 2))
g.add_column(style="dim", justify="right")
g.add_column()
g.add_row("fixed cost per read", f"[bold]{results.cmd_overhead_ms:.1f} ms[/]")
g.add_row("marginal rate",
f"[bold]{results.asymptotic_mbps:.1f} MB/s[/] "
f"[dim]once the command cost is paid[/]")
if results.knee_bytes:
style = "yellow" if results.sweep_cliff else "green"
g.add_row("largest clean transfer",
f"[{style}]{human(results.knee_bytes)}[/]")
body.append(g)
ceiling = dev.link_ceiling_mbps
body.append(Text())
if results.sweep_cliff and results.knee_bytes:
body.append(Text.from_markup(
f"[yellow]Throughput collapses above {human(results.knee_bytes)} "
f"per transfer.[/]\n"
f"[dim]Reads at or below that size run at "
f"{results.peak_sweep_mbps:.1f} MB/s - full speed. Larger ones "
f"fall to a fraction\nof it. Flash does not behave this way; a "
f"link that accumulates errors over a\nlong burst and retries "
f"the transfer does. Check module seating, connector\ncontacts "
f"and cable before suspecting the media.[/]"))
elif results.cmd_overhead_ms > 3.0 and ceiling and \
results.asymptotic_mbps > ceiling * 0.6:
body.append(Text.from_markup(
"[dim]Large fixed cost, healthy marginal rate: the flash keeps up "
"once a\ntransfer is underway and the time is going to command "
"turnaround.\nThis is a reader/link property. Use a larger --chunk "
"for real work.[/]"))
elif ceiling and results.asymptotic_mbps < ceiling * 0.5:
body.append(Text.from_markup(
"[dim]The marginal rate itself is well under the bus ceiling, so "
"bigger reads\nwill not rescue it. The bottleneck is in the data "
"path, not the command\nturnaround.[/]"))
return Panel(Group(*body), title="[bold]transfer size sweep[/]",
border_style="cyan")
def speed_panel(dev: Device, results: Results) -> Panel:
t = Table.grid(padding=(0, 2))
t.add_column(style="dim", justify="right")
t.add_column()
t.add_row("sequential read", f"[bold]{results.seq_mbps:.1f} MB/s[/]")
if results.rand_iops:
t.add_row("random 4K read",
f"[bold]{results.rand_iops:,.0f} IOPS[/] "
f"[dim]({results.rand_mbps:.1f} MB/s)[/]")
if results.write_mbps:
t.add_row("sequential write", f"[bold]{results.write_mbps:.1f} MB/s[/]")
t.add_row("data read",
f"{human(results.bytes_read)} in {results.scan_seconds:.1f}s")
if results.chunk_used:
t.add_row("transfer size", human(results.chunk_used))
ceiling = dev.link_ceiling_mbps
if ceiling:
pct = (results.seq_mbps / ceiling * 100) if results.seq_mbps else 0
style = "green" if pct >= 70 else "yellow" if pct >= 40 else "red"
t.add_row("bus ceiling",
f"{ceiling:.0f} MB/s [dim]({dev.clock}, {dev.bus_width})[/]")
t.add_row("link efficiency", f"[{style}]{pct:.0f}% of ceiling[/]")
lines = []
if not results.direct_io:
lines.append(
"[yellow]O_DIRECT was refused; reads went through the page cache.[/]\n"
"[dim]Throughput above is inflated and should not be trusted.[/]")
peak = max((s.mbps for s in results.samples
if not s.error and not s.warmup), default=0.0)
if ceiling and peak > ceiling * 1.3:
lines.append(
f"[yellow]Peak {peak:.1f} MB/s is well above the ~{ceiling:.0f} MB/s "
"estimated bus ceiling.[/]\n"
"[dim]The estimate is a naive clock x width figure and reader "
"buffering lets short\nbursts beat it slightly, but not by this "
"margin. Suspect cached reads or\nmisreported link parameters.[/]")
if not lines:
if ceiling and results.seq_mbps and results.seq_mbps < ceiling * 0.4:
lines.append(
"[dim]Well under the bus ceiling. See the transfer size sweep "
"below - it separates\ncommand turnaround cost from actual "
"transfer rate.[/]")
elif dev.transport == "usb2":
lines.append(
"[dim]Attached over USB 2.0. Sequential throughput describes the "
"link, not the eMMC.[/]")
body = Group(t, Text(), *[Text.from_markup(l) for l in lines]) if lines else t
return Panel(body, title="[bold]speed[/]", border_style="cyan")
def verdict_panel(dev: Device, results: Results, health_sev: int,
layout_mismatches: int, is_switch: bool,
full: bool, wrote: bool,
gpt_problems: list[str] | None = None,
strict: bool = False,
hw_parts: list | None = None) -> tuple[Panel, str]:
reasons: list[str] = []
level = 0
if health_sev == 2:
level = 2
reasons.append("controller reports the reserve block pool is nearly exhausted")
elif health_sev == 1:
level = max(level, 1)
reasons.append("wear indicators are elevated")
if results.error_regions:
level = 2
reasons.append(f"{len(results.error_regions)} unreadable region(s) on the media")
if results.write_mismatches:
level = 2
reasons.append(f"{len(results.write_mismatches)} write/verify mismatch(es) - "
"suspect fake capacity or failing cells")
link_note = diagnose_link(dev, results, health_sev)
for p in (hw_parts or []):
if p.errors:
level = 2
reasons.append(f"{len(p.errors)} unreadable sector(s) in the {p.name} "
"hardware partition - this area holds the boot chain "
"and is not covered by the user-area scan")
if results.media_errors:
level = 2
reasons.append(f"{results.media_errors} media-level kernel error(s) "
"logged during the scan")
scored = [s for s in results.samples if not s.error and not s.warmup and s.mbps > 0]
if len(scored) >= 8:
median = sorted(s.mbps for s in scored)[len(scored) // 2]
slow = [s for s in scored if s.mbps < median * 0.55]
if len(slow) >= 3 and len(slow) >= len(scored) * 0.05:
level = max(level, 1)
if link_note:
reasons.append(f"{len(slow)} region(s) reading below 55% of median")
else:
reasons.append(
f"{len(slow)} region(s) reading below 55% of median - "
"consistent with heavy internal retry or ECC correction")
if results.sweep_cliff and results.knee_bytes:
level = max(level, 1)
reasons.append(
f"transfers above {human(results.knee_bytes)} collapse while smaller "
"ones run at full speed - a transfer-length dependent fault points at "
"the connection, not the flash")
if link_note and not results.sweep_cliff:
level = max(level, 1)
reasons.append(link_note)
if not results.direct_io:
level = max(level, 1)
reasons.append("O_DIRECT was unavailable, so throughput figures include "
"page cache effects and cannot be compared to the bus ceiling")
# Content findings describe what is written on the module, not whether the
# silicon is sound. A reflash fixes them; a worn NAND die does not. Keeping
# them out of the hardware verdict stops every second-hand module from
# being flagged as suspect hardware.
content: list[str] = list(gpt_problems or [])
if is_switch and layout_mismatches:
content.append("partition table deviates from the stock Switch layout")
if strict:
for c in content:
level = max(level, 1)
reasons.append(c)
if level == 0:
title, style = "PASS", "bold green"
headline = ("Hardware is sound." if content else "No faults found.")
elif level == 1:
title, style = "INSPECT", "bold yellow"
headline = ("Measurement is unreliable - retest before judging the module."
if (link_note or results.sweep_cliff) else
"Usable, but something is worth a second look.")
else:
title, style = "FAIL", "bold red"
headline = "Do not deploy this module."
body = [Text(headline, style=style)]
for r in reasons:
body.append(Text(f" • {r}"))
if not reasons:
body.append(Text(" • health registers nominal", style="dim"))
body.append(Text(" • no read errors across the scanned range", style="dim"))
body.append(Text(" • throughput flat across the address space", style="dim"))
if content and not strict:
body.append(Text())
body.append(Text("content notes - fixed by reflashing, not hardware faults:",
style="cyan"))
for c in content:
body.append(Text(f" • {c}", style="cyan"))
caveats = []
if not full:
caveats.append("sampled scan only - run --full to read every sector")
if not wrote:
caveats.append("read-only - write-side faults and fake capacity are not covered")
if hw_parts and all(not p.readable and not p.errors for p in hw_parts):
caveats.append("boot areas not read")
if dev.health_source == "none":
caveats.append("no health registers; this part predates eMMC 5.0")
if caveats:
body.append(Text())
body.append(Text("coverage: " + "; ".join(caveats), style="dim italic"))
return Panel(Group(*body), title=f"[{style}]{title}[/]",
border_style=style.split()[-1]), title
# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------
def pick_device(devices: list[Device]) -> Device | None:
if not devices:
console.print("[red]No mmcblk devices found.[/]")
console.print("[dim]Check that rtsx_usb_sdmmc is loaded: lsmod | grep rtsx[/]")
return None
if len(devices) == 1:
return devices[0]
console.print("\n[bold]Attached MMC devices[/]\n")
t = Table(box=None, padding=(0, 2))
t.add_column("#", justify="right", style="cyan")
t.add_column("device")
t.add_column("size", justify="right")
t.add_column("model")
t.add_column("mounted")
for i, d in enumerate(devices):
t.add_row(str(i + 1), d.path, human(d.size), d.model or "-",
"[yellow]yes[/]" if d.mounted else "no")
console.print(t)
try:
raw = console.input("\nSelect device number: ").strip()
idx = int(raw) - 1
if 0 <= idx < len(devices):
return devices[idx]
except (ValueError, EOFError, KeyboardInterrupt):
pass
console.print("[red]No device selected.[/]")
return None
def main() -> int:
ap = argparse.ArgumentParser(
prog="emmcheck",
description="Qualification bench for eMMC modules. Read-only by default.",
)
ap.add_argument("device", nargs="?", help="e.g. /dev/mmcblk1 (omit to choose)")
ap.add_argument("--full", action="store_true",
help="read every sector instead of sampling")
ap.add_argument("--points", type=int, default=160,
help="sample positions across the device (default 160)")
ap.add_argument("--chunk", type=int, default=0,
help="read size in MiB; default 0 picks the largest size "
"the link handles cleanly, measured by the sweep")
ap.add_argument("--random-seconds", type=float, default=10.0,
help="duration of the 4K random read test (0 to skip)")
ap.add_argument("--write-test", action="store_true",
help="DESTROYS ALL DATA. Write/verify pass for fake-capacity "
"and write-fault detection.")
ap.add_argument("--strict", action="store_true",
help="treat content findings (GPT drift, non-stock layout) "
"as verdict-affecting; use when qualifying blank modules")
ap.add_argument("--skip-boot", action="store_true",
help="do not read the boot0/boot1 hardware partitions")
ap.add_argument("--no-sweep", action="store_true",
help="skip the transfer size sweep")
ap.add_argument("--json", metavar="FILE", help="append a JSON record of the run")
ap.add_argument("--force", action="store_true",
help="proceed even if the device has mounted partitions")
args = ap.parse_args()
chunk = args.chunk * MIB # 0 means auto-select from the sweep
console.print()
console.print(Rule("[bold]emmcheck[/] [dim]eMMC qualification bench[/]",
style="cyan"))
if os.geteuid() != 0:
console.print("\n[yellow]Not running as root.[/] Raw device reads and "
"EXT_CSD access will fail. Re-run with sudo.\n")
if args.device:
if not os.path.exists(args.device):
console.print(f"[red]{args.device} does not exist.[/]")
return 1
dev = load_device(args.device)
else:
dev = pick_device(list_mmc_devices())
if dev is None:
return 1
read_extcsd(dev)
history = load_history(args.json) if args.json else []
hw_parts = discover_hw_partitions(dev)
gpt = parse_gpt(dev.path, dev.size) if not dev.is_file else GptInfo()
if gpt.names:
for p in dev.partitions:
if not p.label:
p.label = gpt.names.get(p.start_lba)
if dev.mounted and not args.force:
console.print(f"\n[red]{dev.path} has mounted partitions:[/] "
f"{', '.join(dev.mounted)}")
console.print("Unmount them first, or pass --force to test anyway.\n")
return 1
if args.write_test:
console.print()
console.print(Panel(
Text.from_markup(
f"The write test overwrites [bold]every sampled position[/] on "
f"[bold]{dev.path}[/].\n"
"On a Switch module this destroys PRODINFO, which holds "
"console-unique\ncertificates and cannot be regenerated. Dump the "
"module first if it\nhas ever been paired to a console.\n\n"
"Type the device path to confirm."),
title="[bold red]destructive[/]", border_style="red"))
try:
if console.input("\nConfirm: ").strip() != dev.path:
console.print("[yellow]Cancelled.[/]\n")
return 1
except (EOFError, KeyboardInterrupt):
console.print("\n[yellow]Cancelled.[/]\n")
return 1
console.print()
console.print(identity_panel(dev))
hp, health_sev = health_panel(dev)
console.print(hp)
lp, is_switch, layout_mismatches = layout_panel(dev)
console.print(lp)
gp = gpt_panel(gpt)
if gp:
console.print(gp)
results = Results()
console.print()
dmesg_before = capture_dmesg()
progress = Progress(
SpinnerColumn(style="cyan"),
TextColumn("[bold]{task.description}[/]"),
BarColumn(bar_width=40, complete_style="cyan", finished_style="green"),
TextColumn("{task.percentage:>3.0f}%"),
TimeElapsedColumn(),
TimeRemainingColumn(),
console=console,
)
try:
with progress:
# The sweep runs first: it costs ~15s and tells us what transfer
# size this link actually handles, so the main scan is measured on
# the right side of any cliff instead of straddling it.
wtask = progress.add_task("warming up link", total=1)
warm_up(dev)
progress.update(wtask, completed=1)
if not args.no_sweep:
total_reads = sum(max(5, min(40, (24 * MIB) // s))
for s in SWEEP_SIZES if s <= dev.size // 8)
stask = progress.add_task("transfer size sweep",
total=max(1, total_reads))
latency_sweep(dev, results, progress, stask)
if hw_parts and not args.skip_boot:
htask = progress.add_task("boot / rpmb areas",
total=len(hw_parts))
scan_hw_partitions(hw_parts, progress, htask)
if chunk == 0:
chunk = results.knee_bytes or 4 * MIB
chunk = max(64 * 1024, min(16 * MIB, chunk))
results.chunk_used = chunk
offsets, bucket_size = plan_offsets(dev, chunk, args.full, args.points)
if not offsets:
console.print("[red]Device is smaller than one read chunk.[/]\n")
return 1
scan_total = len(offsets) * chunk
label = "surface scan" if args.full else "read profile"
task = progress.add_task(label, total=scan_total)
with RawReader(dev.path, chunk) as reader:
results.direct_io = reader.direct
sequential_profile(dev, reader, results, chunk, offsets,
bucket_size, progress, task)
progress.update(task, completed=scan_total)
if args.random_seconds > 0:
rtask = progress.add_task("random 4K read",
total=args.random_seconds)
with RawReader(dev.path, 4096) as r4k:
random_read_test(dev, r4k, results,
args.random_seconds, progress, rtask)
if args.write_test:
w_offsets, _ = plan_offsets(dev, chunk, args.full, args.points)
wtask = progress.add_task("write + verify",
total=len(w_offsets) * 2)
write_verify_test(dev, results, chunk, w_offsets,
progress, wtask)
except KeyboardInterrupt:
console.print("\n[yellow]Interrupted. Reporting partial results.[/]\n")
except OSError as e:
console.print(f"\n[red]I/O setup failed:[/] {e}")
console.print("[dim]Root is required for raw device access.[/]\n")
return 1
results.kernel_lines, results.link_errors, results.media_errors = \
dmesg_delta(dmesg_before)
hwp = hw_partition_panel(dev, hw_parts, is_switch)
if hwp:
console.print(hwp)
console.print()
console.print(profile_chart(dev, results))
console.print(speed_panel(dev, results))
sp = sweep_panel(dev, results)
if sp:
console.print(sp)
kp = kernel_panel(results)
if kp:
console.print(kp)
ep = errors_panel(dev, results)
if ep:
console.print(ep)
vp, verdict = verdict_panel(dev, results, health_sev, layout_mismatches,
is_switch, args.full, args.write_test,
gpt_problems=gpt.problems + history_findings(
dev, gpt, history),
strict=args.strict, hw_parts=hw_parts)
console.print()
console.print(vp)
console.print()
if args.json:
record = {
"timestamp": datetime.now(timezone.utc).isoformat(),
"verdict": verdict,
"device": {k: v for k, v in asdict(dev).items()
if k not in ("partitions",)},
"vendor": dev.vendor,
"is_switch_layout": is_switch,
"layout_mismatches": layout_mismatches,
"partitions": [asdict(p) for p in dev.partitions],
"seq_mbps": round(results.seq_mbps, 2),
"rand_iops": round(results.rand_iops, 1),
"bytes_read": results.bytes_read,
"error_regions": results.error_regions,
"bad_lbas": results.bad_lbas,
"write_mismatches": results.write_mismatches,
"link_errors": results.link_errors,
"media_errors": results.media_errors,
"kernel_lines": results.kernel_lines[:50],
"bus_ceiling_mbps": dev.link_ceiling_mbps,
"direct_io": results.direct_io,
"cmd_overhead_ms": results.cmd_overhead_ms,
"asymptotic_mbps": results.asymptotic_mbps,
"sweep": results.sweep,
"knee_bytes": results.knee_bytes,
"sweep_cliff": results.sweep_cliff,
"chunk_used": results.chunk_used,
"gpt_problems": gpt.problems,
"gpt_disk_guid": gpt.disk_guid,
"hw_partitions": [asdict(p) for p in hw_parts],
"full_scan": args.full,
"write_test": args.write_test,
}
with open(args.json, "a") as fh:
fh.write(json.dumps(record) + "\n")
console.print(f"[dim]Record appended to {args.json}[/]\n")
return {"PASS": 0, "INSPECT": 1, "FAIL": 2}[verdict]
if __name__ == "__main__":
try:
sys.exit(main())
except KeyboardInterrupt:
console.print("\n[yellow]Aborted.[/]\n")
sys.exit(130)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment