diff --git a/experiments/perception/worker/streaming_profile_stage1/ddrnet-logits-diagnostic-config.pbtxt b/experiments/perception/worker/streaming_profile_stage1/ddrnet-logits-diagnostic-config.pbtxt new file mode 100644 index 0000000..be3bf30 --- /dev/null +++ b/experiments/perception/worker/streaming_profile_stage1/ddrnet-logits-diagnostic-config.pbtxt @@ -0,0 +1,7 @@ +# Diagnostic only: never use 64 MiB logits as the streaming profile output. +name: "ddrnet_goose_fp32_logits" +platform: "tensorrt_plan" +max_batch_size: 0 +input [{ name: "input" data_type: TYPE_FP32 dims: [1,3,512,512] }] +output [{ name: "logits" data_type: TYPE_FP32 dims: [1,64,512,512] }] +instance_group [{ count: 1 kind: KIND_GPU gpus: [0] }] diff --git a/experiments/perception/worker/streaming_profile_stage1/diagnose_ddrnet_costmap.py b/experiments/perception/worker/streaming_profile_stage1/diagnose_ddrnet_costmap.py new file mode 100644 index 0000000..ca140e6 --- /dev/null +++ b/experiments/perception/worker/streaming_profile_stage1/diagnose_ddrnet_costmap.py @@ -0,0 +1,128 @@ +"""Offline single-cell attribution using the unchanged production-pilot voting. + +No detector, model GPU calls or obstacle-history simulation. Replays only causal +raw point/pose increments to attribute material voting, not to qualify policy. +""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import signal +import subprocess +import time +from collections import deque +from datetime import datetime, timezone +from pathlib import Path +from types import SimpleNamespace + +import numpy as np +from pilot_graph import JointGraph +from pilot_sensor_binding import bind_sensors +from pilot_source import SensorArchive, camera_events, merged_events + + +def source_bundle(sequence): + archive = SensorArchive(Path("/sensor-source.npz")) + zero = next(camera_events("/camera-index.jsonl", 1)).time_ns - 500_000_000 + rolling, fresh, pose, previous = deque(), [], None, None + try: + for event in merged_events(archive, "/camera-index.jsonl", sequence + 1): + if event.time_ns < zero: + continue + while rolling and event.time_ns - rolling[0].time_ns > 1_000_000_000: + rolling.popleft() + if event.channel == "pose": + pose = event + elif event.channel == "points": + rolling.append(event) + fresh.append(event) + if sum(len(e.value[0]) for e in rolling) > 64000: + raise ValueError("diagnostic rolling history exceeds pilot bound") + else: + binding = bind_sensors(event.time_ns, pose, fresh, previous_camera_time_ns=previous) + if event.sequence == sequence: + return { + "available": binding.available, + "time_ns": event.time_ns, + "sensor_binding": binding.document(), + "pose": pose.value if pose else None, + "rolling_points": np.concatenate([e.value[0] for e in rolling]), + "rolling_times": np.concatenate( + [np.full(len(e.value[0]), e.time_ns, np.int64) for e in rolling] + ), + } + previous, fresh = event.time_ns, [] + raise RuntimeError("diagnostic source frame unavailable") + finally: + archive.close() + + +def main(): + parser = argparse.ArgumentParser() + parser.add_argument("--output", type=Path, required=True) + parser.add_argument("--masks", type=Path, required=True) + parser.add_argument("--sequence", type=int, choices=range(128), default=115) + parser.add_argument("--cell", type=int, default=1900) + args = parser.parse_args() + signal.alarm(45) + args.output.mkdir(parents=True, exist_ok=False) + started = time.monotonic_ns() + started_utc = datetime.now(timezone.utc).isoformat() + bundle = source_bundle(args.sequence) + with (args.output / "tgs.log").open("wb") as log: + tgs = subprocess.Popen( + ["/usr/local/bin/pilot-tgs"], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=log + ) + try: + graph = JointGraph( + "/code", + tgs, + mask_path="/valid-fov.png", + mapping_path="/goose.csv", + calibration_pack="/calibration.npz", + ) + results = {} + with np.load(args.masks, allow_pickle=False) as masks: + for name in masks.files: + mask = masks[name] + if mask.shape != (512, 512) or mask.dtype != np.uint8: + raise ValueError("diagnostic mask contract changed") + cells, actions, material, evidence = graph.costmap( + bundle, + mask, + SimpleNamespace(occupied=(), unknown=()), + diagnostic_cells=(args.cell,), + ) + results[name] = { + "mask_sha256": hashlib.sha256(mask.tobytes()).hexdigest(), + "cell_state": int(cells[args.cell]), + "material": graph.material_names[material[args.cell]], + "action_before_obstacle_history": int(actions[args.cell]), + **evidence["diagnostic_cells"][str(args.cell)], + } + report = { + "schema_version": "missioncore.ddrnet-costmap-attribution/v1", + "source_sequence": args.sequence, + "cell_index": args.cell, + "grid_cell": graph.grid[args.cell].tolist(), + "started_utc": started_utc, + "started_monotonic_ns": started, + "elapsed_ms": (time.monotonic_ns() - started) / 1e6, + "source_paced": False, + "qualified": False, + "actuation_enabled": False, + "obstacle_history_replayed": False, + "sensor_binding": bundle["sensor_binding"], + "results": results, + } + (args.output / "report.json").write_text(json.dumps(report, indent=2) + "\n") + print(json.dumps(report), flush=True) + finally: + tgs.stdin.close() + tgs.wait(timeout=3) + + +if __name__ == "__main__": + main() diff --git a/experiments/perception/worker/streaming_profile_stage1/diagnose_ddrnet_numeric.py b/experiments/perception/worker/streaming_profile_stage1/diagnose_ddrnet_numeric.py new file mode 100644 index 0000000..23408bb --- /dev/null +++ b/experiments/perception/worker/streaming_profile_stage1/diagnose_ddrnet_numeric.py @@ -0,0 +1,256 @@ +"""Single-frame numerical diagnosis, never a production inference path. + +The 64 MiB logits response is confined to this bounded offline diagnostic. +Reference, alternative arithmetic and TensorRT calls execute sequentially. +""" + +from __future__ import annotations + +import argparse +import hashlib +import http.client +import importlib.util +import json +import signal +import subprocess +import time +from datetime import datetime, timezone +from pathlib import Path + +import numpy as np + +LOGITS_SHAPE = (1, 64, 512, 512) +LOGITS_BYTES = int(np.prod(LOGITS_SHAPE)) * 4 + + +def read_logits(tensor): + binary = tensor.tobytes() + header = json.dumps( + { + "inputs": [ + { + "name": "input", + "shape": [1, 3, 512, 512], + "datatype": "FP32", + "parameters": {"binary_data_size": len(binary)}, + } + ], + "outputs": [{"name": "logits", "parameters": {"binary_data": True}}], + } + ).encode() + connection = http.client.HTTPConnection("127.0.0.1", 8000, timeout=30) + try: + connection.request( + "POST", + "/v2/models/ddrnet_goose_fp32_logits/infer", + body=header + binary, + headers={ + "Inference-Header-Content-Length": str(len(header)), + "Content-Type": "application/octet-stream", + }, + ) + response = connection.getresponse() + payload = response.read(LOGITS_BYTES + 65537) + length = int(response.getheader("Inference-Header-Content-Length") or "0") + if response.status != 200 or not 0 < length <= 65536: + raise RuntimeError("invalid diagnostic logits response") + output = json.loads(payload[:length])["outputs"] + if ( + len(payload) != length + LOGITS_BYTES + or len(output) != 1 + or not ( + output[0]["name"] == "logits" + and output[0]["datatype"] == "FP32" + and output[0]["shape"] == list(LOGITS_SHAPE) + and output[0]["parameters"]["binary_data_size"] == LOGITS_BYTES + ) + ): + raise RuntimeError("diagnostic logits contract changed") + return np.frombuffer(payload, dtype=" len(selected), + "reference_details": score_details(*reference, selected), + "candidate_logit_engine_details": score_details(*candidate, selected), + "elapsed_ms": (time.monotonic_ns() - started) / 1e6, + } + np.savez_compressed( + args.output / "masks.npz", + reference=reference[2], + no_tf32=no_tf32[2], + trt_mask=candidate_mask, + trt_logits_torch_postprocess=candidate[2], + ) + (args.output / "report.json").write_text( + json.dumps(report, indent=2, allow_nan=False) + "\n" + ) + print( + json.dumps({k: v for k, v in report.items() if not k.endswith("details")}), flush=True + ) + finally: + if backend is not None: + backend.close() + if triton.poll() is None: + triton.terminate() + triton.wait(timeout=5) + log.close() + + +if __name__ == "__main__": + main() diff --git a/experiments/perception/worker/streaming_profile_stage1/diagnose_ddrnet_pacing.py b/experiments/perception/worker/streaming_profile_stage1/diagnose_ddrnet_pacing.py new file mode 100644 index 0000000..61a9691 --- /dev/null +++ b/experiments/perception/worker/streaming_profile_stage1/diagnose_ddrnet_pacing.py @@ -0,0 +1,177 @@ +"""Bounded, sequential trtexec pause/DMA diagnostic; not a source-paced gate.""" + +from __future__ import annotations + +import argparse +import hashlib +import http.client +import json +import signal +import subprocess +import threading +import time +from datetime import datetime, timezone +from pathlib import Path + +from pilot_telemetry import NvmlSampler + + +def triton_run(args, payload): + """Same fixed tensor via loopback HTTP; only one model/call in flight.""" + import numpy as np + from pilot_ddrnet_triton import MODEL_ID, TritonDdrnetMaskHttpInferenceBackend + from pilot_triton_stats import read_stats, stats_delta + + command = [ + "tritonserver", + "--model-repository=/models", + "--model-control-mode=explicit", + f"--load-model={MODEL_ID}", + "--allow-grpc=false", + "--allow-metrics=false", + "--http-address=127.0.0.1", + "--log-verbose=1", + ] + tensor = np.frombuffer(payload, dtype="= deadline: + raise RuntimeError("diagnostic Triton not ready within bound") + connection = http.client.HTTPConnection("127.0.0.1", 8000, timeout=1) + try: + connection.request("GET", f"/v2/models/{MODEL_ID}/ready") + if connection.getresponse().status == 200: + break + except OSError: + pass + finally: + connection.close() + time.sleep(0.05) + for _ in range(8): + backend.infer(np.zeros_like(tensor)) + before = read_stats((MODEL_ID,)) + for index in range(32): + if index: + time.sleep(args.idle_ms / 1000) + started = time.monotonic_ns() + mask = backend.infer(tensor) + finished = time.monotonic_ns() + rows.append( + { + "index": index, + "started_monotonic_ns": started, + "finished_monotonic_ns": finished, + "wall_ms": (finished - started) / 1e6, + "mask_sha256": hashlib.sha256(mask.tobytes()).hexdigest(), + **backend.last_timing_ms, + } + ) + stats = stats_delta(before, read_stats((MODEL_ID,))) + return {"returncode": 0, "command": command, "rows": rows, "triton_stats": stats} + finally: + backend.close() + if process.poll() is None: + process.terminate() + process.wait(timeout=5) + + +def main(): + parser = argparse.ArgumentParser() + parser.add_argument("--output", type=Path, required=True) + parser.add_argument("--idle-ms", type=int, choices=(0, 50, 100), default=0) + parser.add_argument("--include-transfers", action="store_true") + parser.add_argument("--blocking-sync", action="store_true") + parser.add_argument("--runtime", choices=("trtexec", "triton"), default="trtexec") + args = parser.parse_args() + if args.runtime == "triton" and (args.blocking_sync or args.include_transfers): + parser.error("Triton transfer/sync settings come from its model config") + signal.alarm(55) + args.output.mkdir(parents=True, exist_ok=False) + payload = Path("/input.f32").read_bytes() + if len(payload) != 1 * 3 * 512 * 512 * 4: + raise ValueError("diagnostic input contract changed") + command = [ + "/usr/bin/trtexec", + "--loadEngine=/model.plan", + "--loadInputs=input:/input.f32", + "--warmUp=500", + "--duration=0", + "--iterations=32", + "--avgRuns=32", + "--infStreams=1", + "--exposeDMA", + f"--idleTime={args.idle_ms}", + f"--exportTimes={args.output / 'times.json'}", + ] + if args.include_transfers: + command.append("--includeDataTransfers") + if args.blocking_sync: + command.append("--noSpinWait") + started = time.monotonic_ns() + started_utc = datetime.now(timezone.utc).isoformat() + sampler, stopped = NvmlSampler(), threading.Event() + samples, errors = [], [] + + def sample(): + try: + for _ in range(1000): + samples.append( + { + "monotonic_ns": time.monotonic_ns(), + **sampler.sample(), + **sampler.sample_device_state(), + } + ) + if stopped.wait(0.05): + return + errors.append("diagnostic telemetry sample bound reached") + except Exception as error: + errors.append(str(error)) + + thread = threading.Thread(target=sample, daemon=True) + thread.start() + try: + if args.runtime == "triton": + result = triton_run(args, payload) + else: + with (args.output / "trtexec.log").open("wb") as log: + completed = subprocess.run( + command, stdout=log, stderr=subprocess.STDOUT, timeout=45 + ) + result = {"command": command, "returncode": completed.returncode} + finally: + stopped.set() + thread.join(timeout=2) + if thread.is_alive(): + raise RuntimeError("diagnostic telemetry did not stop") + sampler.close() + report = { + "schema_version": "missioncore.ddrnet-pacing-diagnostic/v1", + "started_utc": started_utc, + "started_monotonic_ns": started, + "elapsed_ms": (time.monotonic_ns() - started) / 1e6, + "repeated_same_frame": True, + "source_paced": False, + "qualified": False, + "actuation_enabled": False, + "input_sha256": hashlib.sha256(payload).hexdigest(), + "idle_ms": args.idle_ms, + "transfers_included": True if args.runtime == "triton" else args.include_transfers, + "blocking_sync": None if args.runtime == "triton" else args.blocking_sync, + "runtime": args.runtime, + **result, + "telemetry_errors": errors, + "samples": samples, + } + (args.output / "report.json").write_text(json.dumps(report, indent=2) + "\n") + print(json.dumps({k: v for k, v in report.items() if k not in ("samples", "rows")}), flush=True) + return result["returncode"] + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/experiments/perception/worker/streaming_profile_stage1/pilot_graph.py b/experiments/perception/worker/streaming_profile_stage1/pilot_graph.py index ae2df27..2d0d954 100644 --- a/experiments/perception/worker/streaming_profile_stage1/pilot_graph.py +++ b/experiments/perception/worker/streaming_profile_stage1/pilot_graph.py @@ -167,6 +167,17 @@ def current_body(frame, surface): ) +def select_material(votes): + """A tied plurality is unknown, never an allowance by class ordering.""" + if votes.ndim != 2 or votes.shape[1] < 1 or np.any(votes < 0): + raise ValueError("invalid material votes") + maximum = votes.max(axis=1) + unique = (votes == maximum[:, None]).sum(axis=1) == 1 + winner = votes.argmax(axis=1).astype(np.int32) + winner[(maximum == 0) | ~unique] = 0 # material index 0 is always unknown + return winner + + class JointGraph: def __init__(self, code_root, tgs, *, mask_path, mapping_path, calibration_pack): root = Path(code_root) @@ -372,7 +383,11 @@ class JointGraph: "actuation_allowed": False, } - def costmap(self, bundle, mask, obstacle_map): + def costmap(self, bundle, mask, obstacle_map, *, diagnostic_cells=()): + if len(diagnostic_cells) > 8 or any( + type(i) is not int or not 0 <= i < len(self.grid) for i in diagnostic_cells + ): + raise ValueError("costmap diagnostic cell bound exceeded") count = len(self.grid) material = np.zeros(count, np.int32) if not bundle["available"]: @@ -424,7 +439,24 @@ class JointGraph: cell_ids = ids[point_ids] good = (cell_ids >= 0) & (states[point_ids] == 1) np.add.at(votes, (cell_ids[good], self.material_lut[labels[good]]), 1) - material = votes.argmax(axis=1).astype(np.int32) + material = select_material(votes) + diagnostics = {} + for index in diagnostic_cells: + matching = np.flatnonzero(good & (cell_ids == index)) + diagnostics[str(index)] = { + "votes": dict(zip(self.material_names, votes[index].tolist())), + "projected_ground_points": len(matching), + "points_truncated": len(matching) > 128, + "points": [ + { + "mask_y": int(uv[j, 1] * 512 / 600), + "mask_x": int((uv[j, 0] - 100) * 512 / 600), + "label": int(labels[j]), + "material": self.material_names[self.material_lut[labels[j]]], + } + for j in matching[:128] + ], + } # Current and retained geometry can only add a prohibition, never clear TGS. for obstacle in (*obstacle_map.occupied, *obstacle_map.unknown): for cell in obstacle.cells: @@ -447,5 +479,6 @@ class JointGraph: "rejected": int(np.count_nonzero(states == 3)), "stale_cells": int(np.count_nonzero(stale)), "state_counts": dict(Counter(int(x) for x in cells)), + **({"diagnostic_cells": diagnostics} if diagnostics else {}), }, ) diff --git a/experiments/perception/worker/streaming_profile_stage1/pilot_telemetry.py b/experiments/perception/worker/streaming_profile_stage1/pilot_telemetry.py index c3acec5..92cde5a 100644 --- a/experiments/perception/worker/streaming_profile_stage1/pilot_telemetry.py +++ b/experiments/perception/worker/streaming_profile_stage1/pilot_telemetry.py @@ -81,6 +81,38 @@ class NvmlSampler: "gpu_utilization": int(utilization.gpu), } + def sample_device_state(self): + """Optional read-only diagnostic; unsupported fields are not zeroes. + + Kept out of the default streaming sampler to avoid changing its cost. + NVML clock enums: SM=1, memory=2. Power usage is in milliwatts. + """ + if self.closed: + raise RuntimeError("NVML session is closed") + result, errors = {}, {} + for name, symbol, extra in ( + ("sm_clock_mhz", "nvmlDeviceGetClockInfo", (1,)), + ("memory_clock_mhz", "nvmlDeviceGetClockInfo", (2,)), + ("pstate", "nvmlDeviceGetPerformanceState", ()), + ("power_mw", "nvmlDeviceGetPowerUsage", ()), + ): + function = getattr(self.library, symbol, None) + if function is None: + result[name], errors[name] = None, "symbol-unavailable" + continue + function.argtypes = [ + ctypes.c_void_p, + *([ctypes.c_uint] * len(extra)), + ctypes.POINTER(ctypes.c_uint), + ] + function.restype = ctypes.c_int + value = ctypes.c_uint() + code = function(self.handle, *extra, ctypes.byref(value)) + result[name] = int(value.value) if code == 0 else None + if code: + errors[name] = int(code) + return {**result, "device_state_errors": errors} + def close(self): if self.closed: return diff --git a/experiments/perception/worker/streaming_profile_stage1/run_joint_pilot.py b/experiments/perception/worker/streaming_profile_stage1/run_joint_pilot.py index 8c74d8a..d49d5d6 100644 --- a/experiments/perception/worker/streaming_profile_stage1/run_joint_pilot.py +++ b/experiments/perception/worker/streaming_profile_stage1/run_joint_pilot.py @@ -180,7 +180,7 @@ def produce(args, decoder, mailbox, stop, report): mailbox.finish(mailbox.error) -def telemetry(stop, samples): +def telemetry(stop, samples, device_state=False): sampler = None try: sampler = NvmlSampler() @@ -188,6 +188,8 @@ def telemetry(stop, samples): sample = {"monotonic_ns": time.monotonic_ns()} try: sample.update(sampler.sample()) + if device_state: + sample.update(sampler.sample_device_state()) sample["cgroup_memory_mib"] = ( int(Path("/sys/fs/cgroup/memory.current").read_text()) / 1048576 ) @@ -235,6 +237,7 @@ def run(args): output.mkdir(parents=True, exist_ok=False) started = time.monotonic_ns() report = { + "material_vote_rule": "unique-plurality-or-unknown/v1", "schema_version": "missioncore.stage1-joint-pilot/v1", "run_id": args.run_id, "started_utc": datetime.now(UTC).isoformat(), @@ -257,6 +260,8 @@ def run(args): "ddrnet_preprocess": args.ddrnet_preprocess, "schedule": args.schedule, "telemetry_mode": args.telemetry_mode, + "telemetry_device_state": args.telemetry_device_state, + "triton_verbose": args.triton_verbose, "ddrnet_min_source_interval_ms": args.ddrnet_min_source_interval_ms, } children, logs, results, samples = [], [], [], [] @@ -298,6 +303,7 @@ def run(args): "--http-address=127.0.0.1", "--pinned-memory-pool-byte-size=16777216", "--cuda-memory-pool-byte-size=0:16777216", + *(["--log-verbose=1"] if args.triton_verbose else []), ] triton = child( "triton", @@ -446,7 +452,9 @@ def run(args): return begin, ddr_done, gpu_done, ddr_result, mask, proposals, ddr_layer if args.telemetry_mode == "nvml": - monitor = threading.Thread(target=telemetry, args=(stop, samples), daemon=True) + monitor = threading.Thread( + target=telemetry, args=(stop, samples, args.telemetry_device_state), daemon=True + ) monitor.start() producer = threading.Thread( target=produce, args=(args, decoder, mailbox, stop, source_report), daemon=True @@ -753,10 +761,14 @@ if __name__ == "__main__": ) parser.add_argument("--schedule", choices=("serial", "overlap-cpu"), default="serial") parser.add_argument("--telemetry-mode", choices=("nvml", "none"), default="nvml") + parser.add_argument("--telemetry-device-state", action="store_true") + parser.add_argument("--triton-verbose", action="store_true") parser.add_argument("--ddrnet-min-source-interval-ms", type=float, default=0.0) args = parser.parse_args() if not 1 <= args.frames <= 256: parser.error("pilot window must be 1..256 frames") if not 0 <= args.ddrnet_min_source_interval_ms <= 250: parser.error("DDRNet source interval must be 0..250 ms") + if args.telemetry_device_state and args.telemetry_mode != "nvml": + parser.error("device-state telemetry requires NVML") raise SystemExit(run(args)) diff --git a/tests/test_perception_joint_pilot.py b/tests/test_perception_joint_pilot.py index a049cc0..cad88e7 100644 --- a/tests/test_perception_joint_pilot.py +++ b/tests/test_perception_joint_pilot.py @@ -135,10 +135,16 @@ def test_triton_statistics_exclude_warmup_and_reject_counter_reset(pilot): def test_ddrnet_triton_transport_is_bounded_and_validates_output_identity(pilot): module = pilot("pilot_ddrnet_triton") mask = np.zeros((512, 512), np.uint8) - descriptor = {"outputs": [{ - "name": "mask", "datatype": "UINT8", "shape": [1, 512, 512], - "parameters": {"binary_data_size": mask.nbytes}, - }]} + descriptor = { + "outputs": [ + { + "name": "mask", + "datatype": "UINT8", + "shape": [1, 512, 512], + "parameters": {"binary_data_size": mask.nbytes}, + } + ] + } header = json.dumps(descriptor).encode() class Response: @@ -214,6 +220,61 @@ def test_nvml_sampler_reuses_one_session_without_spawning_processes(pilot): assert library.nvmlDeviceGetUtilizationRates.calls == 2 +def test_material_vote_tie_never_allows_by_class_order(pilot): + choose = pilot("pilot_graph").select_material + # unknown=0, hard_surface=1, bare_soil=2, grass=3 + votes = np.array( + [[0, 0, 0, 2], [0, 1, 0, 1], [0, 2, 0, 1], [0, 0, 0, 0], [1, 1, 0, 0], [0, 0, 2, 2]] + ) + np.testing.assert_array_equal(choose(votes), [3, 0, 1, 0, 0, 0]) + with pytest.raises(ValueError, match="invalid material"): + choose(np.array([[0, -1, 0]])) + + +def test_numeric_diagnostic_counts_transitions_and_score_ties(pilot): + module = pilot("diagnose_ddrnet_numeric") + reference, candidate = np.array([[3, 3], [1, 0]]), np.array([[1, 3], [0, 0]]) + assert module.compare_masks(reference, reference)["different_pixels"] == 0 + assert module.compare_masks(reference, candidate) == { + "different_pixels": 2, + "transitions": [ + {"reference": 1, "candidate": 0, "pixels": 1}, + {"reference": 3, "candidate": 1, "pixels": 1}, + ], + } + logits = np.array([[[[100.0]], [[101.0]]]], dtype=np.float32) + details = module.score_details(logits, np.ones_like(logits), np.array([[0]]), [(0, 0)]) + assert details[0]["selected"] == 0 and details[0]["logit_margin"] == 1 + assert details[0]["max_score_ties"] == [0, 1] + + +def test_optional_nvml_device_state_distinguishes_unavailable_from_zero(pilot): + class Function: + def __init__(self, code, value): + self.code, self.value = code, value + + def __call__(self, *args): + args[-1]._obj.value = self.value + return self.code + + module = pilot("pilot_telemetry") + sampler = object.__new__(module.NvmlSampler) + sampler.closed, sampler.handle = False, None + sampler.library = SimpleNamespace( + nvmlDeviceGetClockInfo=Function(0, 2685), + nvmlDeviceGetPerformanceState=Function(0, 0), + nvmlDeviceGetPowerUsage=Function(3, 0), + ) + state = sampler.sample_device_state() + assert state["sm_clock_mhz"] == 2685 and state["pstate"] == 0 + assert state["power_mw"] is None and state["device_state_errors"] == {"power_mw": 3} + del sampler.library.nvmlDeviceGetClockInfo + assert sampler.sample_device_state()["sm_clock_mhz"] is None + sampler.closed = True + with pytest.raises(RuntimeError, match="closed"): + sampler.sample_device_state() + + def test_body_history_is_bounded_and_never_reads_future(pilot): store = pilot("pilot_graph").CurrentStore(None) for index in range(100):