Spaces:
Paused
Paused
File size: 3,999 Bytes
800f9af | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 | """Read a complete schema 2 benchmark snapshot and verify its recorded hashes."""
import hashlib
import json
from pathlib import Path
class BenchmarkData:
def __init__(self, root):
self.root = Path(root).resolve()
manifest_bytes = self._path("manifest.json").read_bytes()
self.manifest = json.loads(manifest_bytes)
if (self.manifest.get("schema_version") != 2 or
self.manifest.get("status") != "complete"):
raise ValueError("Нужен complete manifest schema 2")
source_files = self.manifest.get("source_files")
if not isinstance(source_files, dict) or not source_files:
raise ValueError("Manifest не содержит source_files")
source_digest = hashlib.sha256(json.dumps(
source_files, sort_keys=True, separators=(",", ":")).encode()).hexdigest()
if source_digest != self.manifest.get("source_sha256"):
raise ValueError("Не совпадает source_sha256 в manifest")
artifacts = self.manifest.get("artifacts")
if not isinstance(artifacts, dict) or "summary.json" not in artifacts:
raise ValueError("Manifest не содержит summary.json")
self._raw = {"manifest.json": manifest_bytes.decode("utf-8")}
for name, expected in artifacts.items():
if name == "manifest.json":
raise ValueError("Manifest не может включать собственный хэш")
contents = self._path(name).read_bytes()
if hashlib.sha256(contents).hexdigest() != expected:
raise ValueError(f"Не совпадает SHA256: {name}")
self._raw[name] = contents.decode("utf-8")
self.summary = json.loads(self._raw["summary.json"])
if self.summary.get("schema_version") != 2:
raise ValueError("Нужен summary schema 2")
results = self.summary.get("results")
if not isinstance(results, list) or not results:
raise ValueError("Summary не содержит результатов")
self._results = {}
for row in results:
if row.get("metrics_schema_version") != 2:
raise ValueError("Нужны метрики schema 2")
key = (row["bench_id"], row["controller"])
if key in self._results:
raise ValueError("Повтор сценария/контроллера в summary")
if (row["events_file"] not in artifacts or
row["model_file"] not in artifacts or
row["model_sha256"] != artifacts[row["model_file"]]):
raise ValueError("Файлы прогона не совпадают с manifest")
self._results[key] = row
def _path(self, name):
if (not isinstance(name, str) or not name or
Path(name).name != name or "\\" in name or
Path(name).suffix not in (".json", ".jsonl", ".xml")):
raise ValueError("Недопустимое имя файла")
path = (self.root / name).resolve()
if path.parent != self.root or not path.is_file():
raise ValueError(f"Файл отсутствует или находится вне schema2: {name}")
return path
@property
def raw_files(self):
return sorted(self._raw)
def read_raw(self, name):
if not isinstance(name, str) or name not in self._raw:
raise ValueError("Недоступный файл: выберите файл из списка schema 2")
return self._raw[name]
def result(self, scenario, controller):
return self._results.get((scenario, controller))
def events(self, scenario, controller):
row = self.result(scenario, controller)
if row is None:
return []
return [json.loads(line) for line in self.read_raw(row["events_file"]).splitlines()
if line.strip()]
|