Spaces:
Runtime error
Runtime error
File size: 3,856 Bytes
228ed67 | 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 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 | #!/usr/bin/env python3
"""Offline calibration harness for PopulationMutationPolicy."""
from __future__ import annotations
import argparse
import asyncio
import json
from pathlib import Path
from typing import Any
import yaml
from open_range.builder.mutation_policy import (
PopulationMutationPolicy,
load_mutation_policy_settings,
)
from open_range.builder.snapshot_store import SnapshotStore
from open_range.protocols import BuildContext
def _load_object(path: str | None) -> dict[str, Any]:
if not path:
return {}
payload = Path(path).read_text(encoding="utf-8")
suffix = Path(path).suffix.lower()
if suffix in {".yaml", ".yml"}:
data = yaml.safe_load(payload) or {}
else:
data = json.loads(payload)
if not isinstance(data, dict):
raise ValueError(f"expected an object in {path}")
return data
def _parse_settings_arg(value: str) -> tuple[str, Path]:
if "=" in value:
label, raw_path = value.split("=", 1)
return label.strip(), Path(raw_path).resolve()
path = Path(value).resolve()
return path.stem, path
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(
description=(
"Compare parent-selection scores across one or more "
"PopulationMutationPolicy settings files."
)
)
parser.add_argument(
"--store-dir",
default="snapshots",
help="Snapshot store directory containing <snapshot_id>/spec.json entries.",
)
parser.add_argument(
"--stats",
help=(
"Optional JSON/YAML file mapping snapshot_id to runtime stats such as "
"plays, plays_recent, red_solve_rate, and blue_detect_rate."
),
)
parser.add_argument(
"--context",
help="Optional JSON/YAML file describing the BuildContext to score against.",
)
parser.add_argument(
"--settings",
action="append",
default=[],
help=(
"Optional policy settings file to compare. Repeatable. Accepts "
"'label=path' or just 'path'."
),
)
parser.add_argument(
"--limit",
type=int,
default=5,
help="How many top-ranked parents to include per policy.",
)
args = parser.parse_args(argv)
entries = asyncio.run(SnapshotStore(args.store_dir).list_entries())
if not entries:
raise SystemExit(f"No stored snapshots found under {args.store_dir}")
context = BuildContext.model_validate(_load_object(args.context))
snapshot_stats = _load_object(args.stats)
policies: list[tuple[str, PopulationMutationPolicy]] = [
("default", PopulationMutationPolicy()),
]
for item in args.settings:
label, path = _parse_settings_arg(item)
policies.append(
(label, PopulationMutationPolicy(settings=load_mutation_policy_settings(path)))
)
report = {
"store_dir": str(Path(args.store_dir).resolve()),
"snapshot_count": len(entries),
"context": context.model_dump(mode="json"),
"policies": [],
}
for label, policy in policies:
ranked = sorted(
policy.score_parents(
entries,
context=context,
snapshot_stats=snapshot_stats,
),
key=lambda score: score.total,
reverse=True,
)[: max(args.limit, 1)]
report["policies"].append(
{
"label": label,
"profile_name": policy.name,
"settings": policy.settings_dict(),
"top_parents": [score.log_payload() for score in ranked],
}
)
print(json.dumps(report, indent=2, sort_keys=True))
return 0
if __name__ == "__main__":
raise SystemExit(main())
|