API reference

Everything on this page is generated from the source docstrings.

Top level

shapesandstrides.measure

measure(fn: Callable[[], object], *, warmup: int = DEFAULT_WARMUP, iters: int = MIN_ITERS, device: int = 0, flush_l2: bool = True, policy: ClockPolicy | None = None, min_duration_us: float = MIN_DURATION_US) -> TimingResult

Time a callable honestly.

Returns an interval and a quality tier, never a bare number.

min_duration_us sets the floor for the timed window (see resolve_inner_reps). It exists to make CUDA event quantization small relative to the effect size we claim to detect, not merely to clear the timer's own floor for registering an event at all: a window barely above the timer's resolution still ties heavily, and a bootstrap CI over tied data can land narrower than the timer can actually resolve. As a second line of defense, the CI is widened to at least one quantization step below (see quantization_step in shapesandstrides.stats).

Known limitation: the reported CI is a bootstrap over samples within a single measurement window, so it captures sampling error inside that window but not run-to-run variability between windows. On unlocked hardware the cross-run spread can exceed the reported interval by orders of magnitude — scripts/validate_timing.py measures exactly this. Folding between-window variance into Tier B intervals is planned for Phase 1.

Source code in src/shapesandstrides/timing.py

def measure(
    fn: Callable[[], object],
    *,
    warmup: int = DEFAULT_WARMUP,
    iters: int = MIN_ITERS,
    device: int = 0,
    flush_l2: bool = True,
    policy: ClockPolicy | None = None,
    min_duration_us: float = MIN_DURATION_US,
) -> TimingResult:
    """Time a callable honestly.

    Returns an interval and a quality tier, never a bare number.

    min_duration_us sets the floor for the timed window (see
    resolve_inner_reps). It exists to make CUDA event quantization small
    relative to the effect size we claim to detect, not merely to clear the
    timer's own floor for registering an event at all: a window barely above
    the timer's resolution still ties heavily, and a bootstrap CI over tied
    data can land narrower than the timer can actually resolve. As a second
    line of defense, the CI is widened to at least one quantization step
    below (see quantization_step in shapesandstrides.stats).

    Known limitation: the reported CI is a bootstrap over samples within a
    single measurement window, so it captures sampling error inside that
    window but not run-to-run variability between windows. On unlocked
    hardware the cross-run spread can exceed the reported interval by orders
    of magnitude — scripts/validate_timing.py measures exactly this. Folding
    between-window variance into Tier B intervals is planned for Phase 1.
    """
    import torch

    if iters < MIN_ITERS:
        raise ValueError(f"iters must be at least {MIN_ITERS}, got {iters}")
    if not torch.cuda.is_available():
        raise RuntimeError("measure() requires a CUDA device")

    torch.cuda.set_device(device)
    policy = policy or UnlockedClockPolicy()
    dev = torch.device(f"cuda:{device}")
    scratch = l2_flush_buffer(dev) if flush_l2 else None
    sampler = ClockSampler(device)

    policy.apply()
    # restore() runs in the finally below and clears policy.locked, so capture
    # the lock state now. Reading it after restore would make Tier A unreachable.
    was_locked = policy.locked
    try:
        # Warmup. The default of 200 exists because do_bench's default of 25
        # yields two calls and underestimates by ~30% (triton#2306).
        for _ in range(warmup):
            fn()
        torch.cuda.synchronize()

        # Calibrate the inner loop against one timed iteration.
        probe_start = torch.cuda.Event(enable_timing=True)
        probe_end = torch.cuda.Event(enable_timing=True)
        probe_start.record()
        fn()
        probe_end.record()
        torch.cuda.synchronize()
        inner_reps = resolve_inner_reps(
            probe_start.elapsed_time(probe_end), min_duration_us
        )

        # Events are allocated up front so allocation never lands inside a
        # timed region.
        starts = [torch.cuda.Event(enable_timing=True) for _ in range(iters)]
        ends = [torch.cuda.Event(enable_timing=True) for _ in range(iters)]

        throttle_before = throttle_snapshot(device)
        clock_samples: list[float] = []
        throttled_during = False

        # NVML runs in-process at microsecond latency, unlike the nvidia-smi
        # subprocess this used to shell out to, so the old bounded-sample cap
        # (clock_sample_stride) is unnecessary here — sample every iteration.
        for i in range(iters):
            if scratch is not None:
                scratch.zero_()
            starts[i].record()
            for _ in range(inner_reps):
                fn()
            ends[i].record()
            sm_clock = sampler.sample_clock_mhz()
            if sm_clock is not None:
                clock_samples.append(sm_clock)
            if sampler.throttled_now():
                throttled_during = True

        torch.cuda.synchronize()
        throttle_after = throttle_snapshot(device)
    finally:
        policy.restore()
        sampler.shutdown()

    samples = [
        starts[i].elapsed_time(ends[i]) / inner_reps for i in range(iters)
    ]
    # Hardware-asserted throttling gates the tier; driver-reported software
    # flags (sw_power_cap, sw_thermal_slowdown) do not, because they were
    # measured Active on real laptop hardware at idle, 55C, 18W, so none of
    # them are trustworthy evidence. They are still recorded below, as
    # metadata.
    hw_throttled = hw_throttle_active(throttle_before, throttle_after, throttled_during)
    power_capped = power_cap_active(throttle_before) or power_cap_active(throttle_after)
    sw_thermal_flagged = (
        sw_thermal_throttle_active(throttle_before)
        or sw_thermal_throttle_active(throttle_after)
    )
    hw_power_brake_flagged = (
        hw_power_brake_active(throttle_before) or hw_power_brake_active(throttle_after)
    )
    tier = assign_tier(was_locked, clock_samples, hw_throttled)
    ci_lo, ci_hi = bootstrap_ci(samples)

    step = quantization_step(samples)
    if step is not None and (ci_hi - ci_lo) < step:
        # An interval narrower than the smallest difference the timer can
        # resolve claims precision the instrument does not have.
        mid = percentile(samples, 0.5)
        ci_lo, ci_hi = mid - step / 2, mid + step / 2

    return TimingResult(
        samples_ms=samples,
        median_ms=percentile(samples, 0.5),
        p10_ms=percentile(samples, 0.10),
        p90_ms=percentile(samples, 0.90),
        ci95_lo_ms=ci_lo,
        ci95_hi_ms=ci_hi,
        n=len(samples),
        tier=tier,
        warmup=warmup,
        inner_reps=inner_reps,
        throttle_fired=hw_throttled,
        power_capped=power_capped,
        sw_thermal_flagged=sw_thermal_flagged,
        hw_power_brake_flagged=hw_power_brake_flagged,
        clock_cv_pct=cv_percent(clock_samples) if clock_samples else None,
        clock_range_mhz=(
            max(clock_samples) - min(clock_samples) if clock_samples else None
        ),
        quantization_step_ms=step,
    )

shapesandstrides.compare

compare(candidate_fn: Callable[[], object], baseline_fn: Callable[[], object], *, warmup: int = DEFAULT_WARMUP, iters: int = MIN_ITERS, device: int = 0, flush_l2: bool = True, policy: ClockPolicy | None = None, min_duration_us: float = MIN_DURATION_US) -> ComparisonResult

Measure a candidate against a baseline, interleaved.

The two callables alternate inside a single measurement window rather than running as two separate windows. GPU clocks drift, and a drift between two sequential windows lands entirely in the ratio: measured on an unlocked laptop, comparing identical work that way gave a p90 error of 106% and a worst case of 141%. Interleaved, the same comparison gave a p90 error of 1.0%.

The baseline is still measured fresh on every call and never cached. Interleaving strengthens that rule rather than replacing it: the two sides now share not just a process but a thermal and clock state.

This is a thin CUDA-gated shell: it collects the raw samples via _interleaved_samples and hands them to _compare_impl, which does the actual guard/CI/ratio logic and needs no GPU to test.

Source code in src/shapesandstrides/timing.py

def compare(
    candidate_fn: Callable[[], object],
    baseline_fn: Callable[[], object],
    *,
    warmup: int = DEFAULT_WARMUP,
    iters: int = MIN_ITERS,
    device: int = 0,
    flush_l2: bool = True,
    policy: ClockPolicy | None = None,
    min_duration_us: float = MIN_DURATION_US,
) -> ComparisonResult:
    """Measure a candidate against a baseline, interleaved.

    The two callables alternate inside a single measurement window rather
    than running as two separate windows. GPU clocks drift, and a drift
    between two sequential windows lands entirely in the ratio: measured on
    an unlocked laptop, comparing identical work that way gave a p90 error
    of 106% and a worst case of 141%. Interleaved, the same comparison gave
    a p90 error of 1.0%.

    The baseline is still measured fresh on every call and never cached.
    Interleaving strengthens that rule rather than replacing it: the two
    sides now share not just a process but a thermal and clock state.

    This is a thin CUDA-gated shell: it collects the raw samples via
    _interleaved_samples and hands them to _compare_impl, which does the
    actual guard/CI/ratio logic and needs no GPU to test.
    """
    import torch

    if iters < MIN_ITERS:
        raise ValueError(f"iters must be at least {MIN_ITERS}, got {iters}")
    if not torch.cuda.is_available():
        raise RuntimeError("compare() requires a CUDA device")

    policy = policy or UnlockedClockPolicy()

    (
        candidate_samples,
        baseline_samples,
        inner_reps,
        tier,
        hw_throttled,
        power_capped,
        sw_thermal_flagged,
        hw_power_brake_flagged,
        clock_samples,
    ) = _interleaved_samples(
        candidate_fn,
        baseline_fn,
        warmup=warmup,
        iters=iters,
        device=device,
        flush_l2=flush_l2,
        policy=policy,
        min_duration_us=min_duration_us,
    )

    return _compare_impl(
        candidate_samples,
        baseline_samples,
        tier=tier,
        warmup=warmup,
        inner_reps=inner_reps,
        throttle_fired=hw_throttled,
        power_capped=power_capped,
        sw_thermal_flagged=sw_thermal_flagged,
        hw_power_brake_flagged=hw_power_brake_flagged,
        clock_samples=clock_samples,
    )

Result types

shapesandstrides.TimingResult

Bases: BaseModel

A timing measurement.

Deliberately defines no float, int or index: a caller must not be able to collapse this into a bare number and lose the interval.

Source code in src/shapesandstrides/types.py

class TimingResult(BaseModel):
    """A timing measurement.

    Deliberately defines no __float__, __int__ or __index__: a caller must
    not be able to collapse this into a bare number and lose the interval.
    """

    samples_ms: list[float]
    median_ms: float
    p10_ms: float
    p90_ms: float
    ci95_lo_ms: float
    ci95_hi_ms: float
    n: int
    tier: MeasurementTier
    warmup: int
    inner_reps: int = 1
    # True when either hardware-asserted throttle fired: hw_thermal_slowdown
    # or hw_power_brake_slowdown. Both come from the GPU's own hardware
    # safety circuits, not driver/vendor software policy, and either one
    # alone gates the tier down to C. hw_power_brake_flagged below records
    # which one it was; throttle_fired is the OR of both.
    throttle_fired: bool = False
    hw_power_brake_flagged: bool = False
    # Everything below is metadata only — recorded, never gating. Measured
    # on real consumer laptop hardware: sw_power_cap and sw_thermal_slowdown
    # were both Active at idle (55C, 18W, card cold and doing nothing), so
    # neither software flag is trustworthy evidence of anything.
    power_capped: bool = False
    sw_thermal_flagged: bool = False
    clock_cv_pct: float | None = None
    clock_range_mhz: float | None = None
    quantization_step_ms: float | None = None

    @field_validator("samples_ms")
    @classmethod
    def _need_two_samples(cls, v: list[float]) -> list[float]:
        if len(v) < 2:
            raise ValueError("a timing result needs at least 2 samples")
        return v

    @property
    def is_performance_valid(self) -> bool:
        return self.tier is not MeasurementTier.C

shapesandstrides.ComparisonResult

Bases: BaseModel

A candidate measured against a baseline timed in the same session.

The baseline is never cached across runs. Re-measuring it every time is what lets us tell "my kernel regressed" from "torch got faster", and it is what makes the ratio comparable across machines and architectures.

Source code in src/shapesandstrides/types.py

class ComparisonResult(BaseModel):
    """A candidate measured against a baseline timed in the same session.

    The baseline is never cached across runs. Re-measuring it every time is
    what lets us tell "my kernel regressed" from "torch got faster", and it
    is what makes the ratio comparable across machines and architectures.
    """

    candidate: TimingResult
    baseline: TimingResult
    speedup: float
    speedup_ci_lo: float
    speedup_ci_hi: float

    @property
    def tier(self) -> MeasurementTier:
        """A comparison is only as trustworthy as its worse half."""
        order = [MeasurementTier.A, MeasurementTier.B, MeasurementTier.C]
        return max(
            (self.candidate.tier, self.baseline.tier), key=order.index
        )

    @property
    def is_performance_valid(self) -> bool:
        return self.tier is not MeasurementTier.C

tier property

tier: MeasurementTier

A comparison is only as trustworthy as its worse half.

shapesandstrides.MeasurementTier

Bases: str, Enum

How much the numbers in a result can be trusted.

A: clocks locked and verified stable. Full verdicts, drift-eligible. B: clocks floating, variance measured and folded into the interval. C: unstable or throttled. No performance verdict is valid.

Source code in src/shapesandstrides/types.py

class MeasurementTier(str, Enum):
    """How much the numbers in a result can be trusted.

    A: clocks locked and verified stable. Full verdicts, drift-eligible.
    B: clocks floating, variance measured and folded into the interval.
    C: unstable or throttled. No performance verdict is valid.
    """

    A = "A"
    B = "B"
    C = "C"

shapesandstrides.EnvironmentFingerprint

Bases: BaseModel

Identifies the toolchain and device a run happened on.

Two runs are only comparable when their fingerprints match exactly.

Source code in src/shapesandstrides/types.py

class EnvironmentFingerprint(BaseModel):
    """Identifies the toolchain and device a run happened on.

    Two runs are only comparable when their fingerprints match exactly.
    """

    torch_version: str
    triton_version: str | None = None
    cuda_version: str | None = None
    driver_version: str | None = None
    gpu_name: str | None = None
    compute_cap: str | None = None
    arch_family: str | None = None
    sm_count: int | None = None

    def matches(self, other: "EnvironmentFingerprint") -> bool:
        return self.model_dump() == other.model_dump()

Clock policies

shapesandstrides.clocks.UnlockedClockPolicy

Measure without pinning. Honest, and the only option on most machines.

Source code in src/shapesandstrides/clocks.py

class UnlockedClockPolicy:
    """Measure without pinning. Honest, and the only option on most machines."""

    locked = False

    def apply(self) -> None:
        return None

    def restore(self) -> None:
        return None

shapesandstrides.clocks.LockedClockPolicy

Pin the SM clock, and optionally the power cap, verifying by readback.

Source code in src/shapesandstrides/clocks.py

class LockedClockPolicy:
    """Pin the SM clock, and optionally the power cap, verifying by readback."""

    def __init__(self, target_sm_mhz: int, power_cap_w: int | None = None):
        self.target_sm_mhz = target_sm_mhz
        self.power_cap_w = power_cap_w
        self.locked = False

    def apply(self) -> None:
        _run_smi(["-lgc", f"{self.target_sm_mhz},{self.target_sm_mhz}"])
        try:
            # nvidia-smi exits 0 when it refuses the write, so the exit code
            # proves nothing. Only the readback does.
            observed = smi_query_float("clocks.sm")
            if (
                observed is None
                or abs(observed - self.target_sm_mhz) > LOCK_READBACK_TOLERANCE_MHZ
            ):
                raise ClockLockError(
                    f"requested {self.target_sm_mhz} MHz, device reports {observed}"
                )
            if self.power_cap_w is not None:
                _run_smi(["-pl", str(self.power_cap_w)])
                observed_w = smi_query_float("power.limit")
                if (
                    observed_w is None
                    or abs(observed_w - self.power_cap_w) > POWER_READBACK_TOLERANCE_W
                ):
                    raise ClockLockError(
                        f"requested {self.power_cap_w} W cap, device reports {observed_w}"
                    )
        except Exception:
            # The -lgc write already reached the device. Leaving it applied
            # after a loud failure silently pins the GPU for everything after.
            self.restore()
            raise
        self.locked = True

    def restore(self) -> None:
        _run_smi(["-rgc"])
        _run_smi(["-rmc"])
        self.locked = False

shapesandstrides.clocks.ClockLockError

Bases: RuntimeError

Raised when a requested operating point could not be established.

Source code in src/shapesandstrides/clocks.py

class ClockLockError(RuntimeError):
    """Raised when a requested operating point could not be established."""

shapesandstrides.clocks.assign_tier

assign_tier(locked: bool, clock_samples: list[float], hw_throttled: bool) -> MeasurementTier

Classify a measurement window.

Throttle flags alone are not sufficient: on an RTX 3060 laptop the SM clock swung 495 MHz (5.1% CV) while the flags stayed silent, and two identical runs disagreed on whether throttling fired at all. Observed variance is the governing signal.

Only hw_throttled and observed clock variance gate the tier — nothing else. The rule: hardware-asserted throttling disqualifies a measurement; driver-reported software flags are recorded but do not, because they were measured stuck Active at idle on consumer hardware. Concretely: hw_throttled is true when either hw_thermal_slowdown or hw_power_brake_slowdown fired — both are asserted by the GPU's own hardware safety circuits, not by software policy, so either one alone disqualifies a window. Every software throttle flag we tried turned out to be untrustworthy: measured on real consumer laptop hardware, sw_power_cap and sw_thermal_slowdown were both Active at idle — 55C, 18W, the card cold and doing nothing. A flag stuck on at idle is not evidence of anything, so driver/vendor software policy is excluded entirely, and stable observed variance is what governs everything else. Software throttle activity is still recorded on TimingResult (power_capped, sw_thermal_flagged) as metadata, never as a gate.

Source code in src/shapesandstrides/clocks.py

def assign_tier(
    locked: bool, clock_samples: list[float], hw_throttled: bool
) -> MeasurementTier:
    """Classify a measurement window.

    Throttle flags alone are not sufficient: on an RTX 3060 laptop the SM
    clock swung 495 MHz (5.1% CV) while the flags stayed silent, and two
    identical runs disagreed on whether throttling fired at all. Observed
    variance is the governing signal.

    Only ``hw_throttled`` and observed clock variance gate the tier —
    nothing else. The rule: hardware-asserted throttling disqualifies a
    measurement; driver-reported software flags are recorded but do not,
    because they were measured stuck ``Active`` at idle on consumer
    hardware. Concretely: ``hw_throttled`` is true when either
    ``hw_thermal_slowdown`` or ``hw_power_brake_slowdown`` fired — both are
    asserted by the GPU's own hardware safety circuits, not by software
    policy, so either one alone disqualifies a window. Every *software*
    throttle flag we tried turned out to be untrustworthy: measured on real
    consumer laptop hardware, ``sw_power_cap`` and ``sw_thermal_slowdown``
    were both `Active` at idle — 55C, 18W, the card cold and doing nothing.
    A flag stuck on at idle is not evidence of anything, so driver/vendor
    software policy is excluded entirely, and stable observed variance is
    what governs everything else. Software throttle activity is still
    recorded on TimingResult (``power_capped``, ``sw_thermal_flagged``) as
    metadata, never as a gate.
    """
    if hw_throttled:
        return MeasurementTier.C
    if clock_samples and cv_percent(clock_samples) > TIER_C_CV_PCT:
        return MeasurementTier.C
    if locked and clock_samples:
        spread = max(clock_samples) - min(clock_samples)
        if spread <= TIER_A_RANGE_MHZ:
            return MeasurementTier.A
    return MeasurementTier.B

Statistics

shapesandstrides.stats.bootstrap_ci

bootstrap_ci(samples: list[float], confidence: float = 0.95, n_resamples: int = 2000, seed: int = 12648430) -> tuple[float, float]

Percentile bootstrap CI of the median.

Seeded so a given sample set always yields the same interval — a reproducibility requirement, not a convenience.

Source code in src/shapesandstrides/stats.py

def bootstrap_ci(
    samples: list[float],
    confidence: float = 0.95,
    n_resamples: int = 2000,
    seed: int = 0xC0FFEE,
) -> tuple[float, float]:
    """Percentile bootstrap CI of the median.

    Seeded so a given sample set always yields the same interval — a
    reproducibility requirement, not a convenience.
    """
    if len(samples) < 2:
        raise ValueError("bootstrap CI needs at least 2 samples")
    rng = random.Random(seed)
    k = len(samples)
    medians = sorted(
        statistics.median(rng.choices(samples, k=k)) for _ in range(n_resamples)
    )
    alpha = 1.0 - confidence
    lo_i = int((alpha / 2) * n_resamples)
    hi_i = min(n_resamples - 1, int((1 - alpha / 2) * n_resamples))
    return (float(medians[lo_i]), float(medians[hi_i]))

shapesandstrides.stats.ratio_ci

ratio_ci(candidate: list[float], baseline: list[float], confidence: float = 0.95, n_resamples: int = 2000, seed: int = 12648430) -> tuple[float, float]

CI of speedup = median(baseline) / median(candidate).

Values above 1.0 mean the candidate is faster. Resampling both sides independently propagates uncertainty from each into the ratio, which a naive ratio-of-medians throws away.

Source code in src/shapesandstrides/stats.py

def ratio_ci(
    candidate: list[float],
    baseline: list[float],
    confidence: float = 0.95,
    n_resamples: int = 2000,
    seed: int = 0xC0FFEE,
) -> tuple[float, float]:
    """CI of speedup = median(baseline) / median(candidate).

    Values above 1.0 mean the candidate is faster. Resampling both sides
    independently propagates uncertainty from each into the ratio, which a
    naive ratio-of-medians throws away.
    """
    if len(candidate) < 2 or len(baseline) < 2:
        raise ValueError("ratio CI needs at least 2 samples on each side")
    rng = random.Random(seed)
    ratios = []
    for _ in range(n_resamples):
        c = statistics.median(rng.choices(candidate, k=len(candidate)))
        b = statistics.median(rng.choices(baseline, k=len(baseline)))
        if c == 0:
            raise ValueError("candidate median resampled to zero")
        ratios.append(b / c)
    ratios.sort()
    alpha = 1.0 - confidence
    lo_i = int((alpha / 2) * n_resamples)
    hi_i = min(n_resamples - 1, int((1 - alpha / 2) * n_resamples))
    return (float(ratios[lo_i]), float(ratios[hi_i]))

shapesandstrides.stats.percentile

percentile(samples: list[float], p: float) -> float

Linear-interpolated percentile. p is in [0, 1].

Source code in src/shapesandstrides/stats.py

def percentile(samples: list[float], p: float) -> float:
    """Linear-interpolated percentile. p is in [0, 1]."""
    if not samples:
        raise ValueError("percentile of an empty sample set")
    if not 0.0 <= p <= 1.0:
        raise ValueError(f"p must be in [0, 1], got {p}")
    s = sorted(samples)
    if len(s) == 1:
        return float(s[0])
    idx = (len(s) - 1) * p
    lo = int(idx)
    hi = min(lo + 1, len(s) - 1)
    frac = idx - lo
    return float(s[lo] * (1 - frac) + s[hi] * frac)

shapesandstrides.stats.cv_percent

cv_percent(samples: list[float]) -> float

Coefficient of variation as a percentage. 0.0 for a zero mean.

Source code in src/shapesandstrides/stats.py

def cv_percent(samples: list[float]) -> float:
    """Coefficient of variation as a percentage. 0.0 for a zero mean."""
    if not samples:
        raise ValueError("cv of an empty sample set")
    if len(samples) == 1:
        return 0.0
    mean = statistics.mean(samples)
    if mean == 0:
        return 0.0
    return 100.0 * statistics.pstdev(samples) / abs(mean)

Environment

shapesandstrides.env.capture_fingerprint

capture_fingerprint(device_index: int = 0) -> EnvironmentFingerprint

Source code in src/shapesandstrides/env.py

def capture_fingerprint(device_index: int = 0) -> EnvironmentFingerprint:
    import torch

    triton_version = None
    try:
        import triton

        triton_version = triton.__version__
    except ImportError:
        pass

    compute_cap = smi_query("compute_cap", device_index)
    sm_count = None
    if torch.cuda.is_available():
        sm_count = torch.cuda.get_device_properties(device_index).multi_processor_count
        if compute_cap is None:
            major, minor = torch.cuda.get_device_capability(device_index)
            compute_cap = f"{major}.{minor}"

    return EnvironmentFingerprint(
        torch_version=torch.__version__,
        triton_version=triton_version,
        cuda_version=torch.version.cuda,
        driver_version=smi_query("driver_version", device_index),
        gpu_name=smi_query("name", device_index),
        compute_cap=compute_cap,
        arch_family=arch_family(compute_cap),
        sm_count=sm_count,
    )

shapesandstrides.env.arch_family

arch_family(compute_cap: str | None) -> str | None

Source code in src/shapesandstrides/env.py

def arch_family(compute_cap: str | None) -> str | None:
    if not compute_cap:
        return None
    try:
        major, minor = (int(x) for x in str(compute_cap).split("."))
    except ValueError:
        return None
    return _ARCH_BY_CAP.get((major, minor), f"unknown-sm{major}{minor}")

Telemetry

shapesandstrides.telemetry.ClockSampler

Samples SM clock and throttle state cheaply, or not at all.

If NVML is unavailable we collect NO clock evidence rather than collecting bad evidence. Tier A requires clock evidence, so a host without NVML can never reach it. That is the intended, conservative degradation.

Source code in src/shapesandstrides/telemetry.py

class ClockSampler:
    """Samples SM clock and throttle state cheaply, or not at all.

    If NVML is unavailable we collect NO clock evidence rather than collecting
    bad evidence. Tier A requires clock evidence, so a host without NVML can
    never reach it. That is the intended, conservative degradation.
    """

    def __init__(self, device: int = 0):
        self.device = device
        self.available = False
        self._handle = None
        self._nvml = None
        try:
            import pynvml

            pynvml.nvmlInit()
            self._handle = pynvml.nvmlDeviceGetHandleByIndex(device)
            self._nvml = pynvml
            self.available = True
        except Exception:
            self.available = False

    def sample_clock_mhz(self) -> float | None:
        if not self.available:
            return None
        try:
            return float(
                self._nvml.nvmlDeviceGetClockInfo(
                    self._handle, self._nvml.NVML_CLOCK_SM
                )
            )
        except Exception:
            return None

    def throttled_now(self) -> bool | None:
        """True if the *hardware* thermal assertion is currently set.

        Checks only HwThermalSlowdown (bit 0x40) — the GPU's own hardware
        safety circuit — not any software-reported reason. Measured on real
        consumer laptop hardware: SwPowerCap (0x4) and SwThermalSlowdown
        (0x20) were both set at idle, 55C, 18W, card cold and doing
        nothing. A software flag stuck on at idle is not evidence of
        anything, so no software reason is trusted here, including the
        generic HwSlowdown (0x8) parent bit, which aggregates
        HwThermalSlowdown with HwPowerBrakeSlowdown and so is not the
        unambiguous hardware-thermal signal by itself. Only the specific
        thermal bit gates; everything else is metadata sampled separately
        from the before/after nvidia-smi snapshot in shapesandstrides.timing.
        """
        if not self.available:
            return None
        try:
            reasons = self._nvml.nvmlDeviceGetCurrentClocksThrottleReasons(
                self._handle
            )
        except Exception:
            return None
        hw_thermal_bit = getattr(
            self._nvml, "nvmlClocksThrottleReasonHwThermalSlowdown", 0x40
        )
        return bool(reasons & hw_thermal_bit)

    def shutdown(self) -> None:
        if self.available:
            try:
                self._nvml.nvmlShutdown()
            except Exception:
                pass

throttled_now

throttled_now() -> bool | None

True if the hardware thermal assertion is currently set.

Checks only HwThermalSlowdown (bit 0x40) — the GPU's own hardware safety circuit — not any software-reported reason. Measured on real consumer laptop hardware: SwPowerCap (0x4) and SwThermalSlowdown (0x20) were both set at idle, 55C, 18W, card cold and doing nothing. A software flag stuck on at idle is not evidence of anything, so no software reason is trusted here, including the generic HwSlowdown (0x8) parent bit, which aggregates HwThermalSlowdown with HwPowerBrakeSlowdown and so is not the unambiguous hardware-thermal signal by itself. Only the specific thermal bit gates; everything else is metadata sampled separately from the before/after nvidia-smi snapshot in shapesandstrides.timing.

Source code in src/shapesandstrides/telemetry.py

def throttled_now(self) -> bool | None:
    """True if the *hardware* thermal assertion is currently set.

    Checks only HwThermalSlowdown (bit 0x40) — the GPU's own hardware
    safety circuit — not any software-reported reason. Measured on real
    consumer laptop hardware: SwPowerCap (0x4) and SwThermalSlowdown
    (0x20) were both set at idle, 55C, 18W, card cold and doing
    nothing. A software flag stuck on at idle is not evidence of
    anything, so no software reason is trusted here, including the
    generic HwSlowdown (0x8) parent bit, which aggregates
    HwThermalSlowdown with HwPowerBrakeSlowdown and so is not the
    unambiguous hardware-thermal signal by itself. Only the specific
    thermal bit gates; everything else is metadata sampled separately
    from the before/after nvidia-smi snapshot in shapesandstrides.timing.
    """
    if not self.available:
        return None
    try:
        reasons = self._nvml.nvmlDeviceGetCurrentClocksThrottleReasons(
            self._handle
        )
    except Exception:
        return None
    hw_thermal_bit = getattr(
        self._nvml, "nvmlClocksThrottleReasonHwThermalSlowdown", 0x40
    )
    return bool(reasons & hw_thermal_bit)