kempnerpulse.compute

KempnerPulse Layer 3 — Compute.

Pure-functional domain logic over CanonicalRecord objects: the Real Utilization composite, the workload classification cascade, and health. Outputs are ComputedRecord objects for the Present layer. No I/O, no source vocabulary.

compute_record (single sample) and compute_tick (one tick, threading the previous record per GPU) are the entry points; the building blocks (real_util, classify, health, presets) are exported for direct use and testing.

class kempnerpulse.compute.ComputedRecord

Bases: object

One fully-computed per-GPU sample: canonical metrics + derived signals.

record: CanonicalRecord

gpu_index: int

gpu_uuid: str

model_name: str | None

preset_name: str

weights: Tuple[float, float, float, float]

workload_class: WorkloadClass

bottleneck: BottleneckCategory

health_style: str

memory_total_mebibytes: float | None = None

memory_used_fraction: float | None = None

pcie_replay_rate_per_second: float | None = None

property gpu_id: str

property status_line: str

__init__(record, gpu_index, gpu_uuid, model_name, real_util, preset_name, weights, workload_class, bottleneck, health, health_style, memory_total_mebibytes=None, memory_used_fraction=None, pcie_replay_rate_per_second=None)

Parameters:

• record (CanonicalRecord)

• gpu_index (int)

• gpu_uuid (str)

• model_name (str | None)

• real_util (float)

• preset_name (str)

• weights (Tuple[float, float, float, float])

• workload_class (WorkloadClass)

• bottleneck (BottleneckCategory)

• health (str)

• health_style (str)

• memory_total_mebibytes (float | None)

• memory_used_fraction (float | None)

• pcie_replay_rate_per_second (float | None)

Return type:

None

class kempnerpulse.compute.WorkloadClass

Bases: Enum

The twelve mutually-exclusive workload categories.

The value is the exact human-readable status label. bottleneck gives the coarse rollup; label is the display string.

IDLE = 'idle'

TENSOR_HEAVY_COMPUTE = 'tensor-heavy compute'

TENSOR_COMPUTE = 'tensor compute'

FP64_HPC_COMPUTE = 'FP64 / HPC compute'

IO_OR_DATA_LOADING = 'I/O or data-loading'

MEMORY_BOUND = 'memory-bound'

COMPUTE_HEAVY = 'compute-heavy'

COMPUTE_ACTIVE = 'compute-active'

MEMORY_ACTIVE = 'memory-active'

BUSY_LOW_SM_USE = 'busy, low SM use'

LOW_UTILIZATION = 'low utilization'

MIXED_OR_MODERATE = 'mixed / moderate'

property label: str

property bottleneck: BottleneckCategory

class kempnerpulse.compute.BottleneckCategory

Bases: Enum

The coarse five-way rollup used for summary colouring.

IDLE = 'idle'

COMPUTE = 'compute'

IO = 'io'

MEMORY = 'memory'

MIXED = 'mixed'

kempnerpulse.compute.compute_record(record, *, prev=None, weights=(0.35, 0.35, 0.2, 0.1), preset_name=None, model_name=None)

Compute every derived signal for one canonical record.

prev is the same GPU’s previous record (used only for the replay-rate difference). weights is the composite weight tuple; preset_name is resolved from the weights when not supplied. model_name rides through to the result and selects the per-model temperature warning.

Parameters:

• record (CanonicalRecord)

• prev (CanonicalRecord | None)

• weights (Tuple[float, float, float, float])

• preset_name (str | None)

• model_name (str | None)

Return type:

ComputedRecord

kempnerpulse.compute.compute_tick(records, prev_by_index=None, **opts)

Compute one tick’s records, threading the previous record per GPU.

prev_by_index maps entity_gpu_index to that GPU’s previous canonical record; it is updated in place so a caller can reuse the same dict across ticks. Any prev in opts is ignored in favour of the per-GPU lookup.

Parameters:

prev_by_index (Dict[int, CanonicalRecord] | None)

Return type:

List[ComputedRecord]

kempnerpulse.compute.graphics_engine_percent(record)

Graphics/compute-engine activity as a percent, with the NVML fallback.

Prefers the graphics/compute engine active fraction; when that reading is absent, falls back to the NVML busy-time fraction (the same time-fraction signal nvidia-smi reports). Both missing → 0.0.

Parameters:

record (CanonicalRecord)

Return type:

float

kempnerpulse.compute.warning_temperature_for_model(model_name)

The GPU-die warning temperature for a model name (default when unknown).

Parameters:

model_name (str | None)

Return type:

float

kempnerpulse.compute.resolve_preset(name)

Return the weight tuple for a preset name.

Raises KeyError (with the known names) if the name is not a built-in preset.

Parameters:

name (str)

Return type:

Tuple[float, float, float, float]

kempnerpulse.compute.preset_name_for_weights(weights)

Name a weight tuple: "ai" / "hpc" / "mem", else "custom".

A tuple matches a built-in preset only when every weight is within a small floating-point tolerance of that preset’s weights.

Parameters:

weights (Tuple[float, float, float, float])

Return type:

str

Modules

classify(record, real_util)	Classify one record into a WorkloadClass (first match wins).
health(record, *, ...)	Return (health_label, rich_style) for one record.
pipeline	The Compute layer entry point — CanonicalRecord to ComputedRecord.
presets	Real Utilization weight presets.
real_util(record, weights)	Weighted Real Utilization composite for one record, clamped to [0,100].
result	Compute-layer output types — the Compute → Present contract.
thresholds	Numeric cutoffs for the Real Utilization composite and the classification.