SILVER/NODEDC_MISSION_CORE
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

feat(laboratory): publish M48S load envelope

Browse Source
This commit is contained in:
DCCONSTRUCTIONS
2026-08-25 20:36:26 +03:00
parent e21332b9e1
commit bab0bb2df2
8 changed files with 624 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/k1link/web/m48s_fixed_class_detector_lab_api.py
+75
View File
@@ -331,6 +331,16 @@ def _load_result_uncached(candidate: Path) -> dict[str, Any]:
or decision.get("integrated_world_state_gate_evaluated") is not integrated
or (integrated and decision.get("integrated_world_state_gate_passed") is not True)
or (integrated and decision.get("detector_replacement_authorized") is not False)
or (
metrics.get("load_envelope") is not None
and (
decision.get("load_envelope_evaluated") is not True
or decision.get("production_rate_repeatability_passed") is not False
or decision.get("reserve_12_fps_passed") is not True
or decision.get("limit_15_fps_passed") is not True
or decision.get("load_envelope_accepted") is not False
)
)
or decision.get("production_accepted") is not False
or not isinstance(method, dict)
or method.get("schema_version") != "missioncore.laboratory-method/v1"
@@ -341,6 +351,10 @@ def _load_result_uncached(candidate: Path) -> dict[str, Any]:
metrics.get("runtime_hardening") is not None
and not _valid_runtime_hardening(metrics["runtime_hardening"])
)
or (
metrics.get("load_envelope") is not None
and not _valid_load_envelope(metrics["load_envelope"])
)
or not isinstance(manifest.get("limitations"), list)
or not isinstance(catalog_descriptor, dict)
or catalog_descriptor.get("path") != "catalog.json"
@@ -420,6 +434,67 @@ def _valid_runtime_hardening(value: object) -> bool:
)
def _valid_load_envelope(value: object) -> bool:
if not isinstance(value, dict) or value.get("schema_version") != (
"missioncore.m48s-load-envelope-comparison/v1"
):
return False
if (
value.get("production_rate_repeatability_passed") is not False
or value.get("reserve_12_fps_passed") is not True
or value.get("limit_15_fps_passed") is not True
or value.get("load_envelope_accepted") is not False
or value.get("compute_capacity_at_least_fps") != 15.0
or value.get("bottleneck_interpretation")
!= "rare-source-decode-or-scheduling-tail-not-steady-gpu-saturation"
):
return False
scenarios = value.get("scenarios")
scenario_ids = [item.get("id") for item in scenarios if isinstance(item, dict)] if (
isinstance(scenarios, list)
) else []
if scenario_ids != ["production-10fps", "reserve-12fps", "limit-15fps"]:
return False
expected_targets = (False, True, True)
required_numeric = {
"requested_source_rate_hz",
"source_frames_admitted",
"delivered_world_states",
"superseded_frames",
"delivery_ratio",
"effective_world_state_fps",
"world_state_completion_age_p95_ms",
"world_state_completion_age_p99_ms",
"world_state_completion_age_maximum_ms",
"decode_p95_ms",
"decode_maximum_ms",
"detector_p95_ms",
"detector_maximum_ms",
"gpu_utilization_mean_percent",
"gpu_utilization_maximum_percent",
"gpu_memory_maximum_mib",
"process_peak_rss_mib",
"queue_capacity",
"additional_inference_passes",
}
for scenario, target in zip(scenarios, expected_targets, strict=True):
if (
not isinstance(scenario, dict)
or scenario.get("integrity_gate_passed") is not True
or scenario.get("operating_target_gate_passed") is not target
or scenario.get("additional_inference_passes") != 0
or scenario.get("queue_capacity") != 2
or not isinstance(scenario.get("queue_high_watermarks"), dict)
or any(value != 2 for value in scenario["queue_high_watermarks"].values())
or not isinstance(scenario.get("thresholds"), dict)
or any(not _nonnegative_number(scenario.get(key)) for key in required_numeric)
or not isinstance(scenario.get("frame_evidence_sha256"), str)
or SHA256.fullmatch(scenario["frame_evidence_sha256"]) is None
):
return False
return True
def _nonnegative_number(value: object) -> bool:
return isinstance(value, (int, float)) and not isinstance(value, bool) and float(value) >= 0.0