feat(@projects/@magic-civilization): ✨ add combat and city tracking stats
Co-Authored-By: Lilith Autocommit <noreply@atlilith.com>
This commit is contained in:
Natalie
parent
f3f9ad8236
commit
979cd0ca26
5 changed files with 162 additions and 166 deletions
|
|
@ -57,22 +57,6 @@ var _total_cities_founded: int = 0
|
|||
var _total_cities_captured: int = 0
|
||||
var _turn_first_combat: int = -1
|
||||
var _turn_first_city_captured: int = -1
|
||||
# Buffered event log — flushed to events.jsonl once per turn
|
||||
var _event_buffer: Array[Dictionary] = []
|
||||
# Guards against writing terminal outcome line twice (e.g. victory during max_turns path)
|
||||
var _final_line_written: bool = false
|
||||
|
||||
# Test harness state (AUTO_PLAY_SEED path)
|
||||
var _seed: int = 0
|
||||
var _seed_set: bool = false
|
||||
var _start_time: float = 0.0
|
||||
var _victory_winner: int = -1
|
||||
var _victory_type: String = ""
|
||||
var _result_written: bool = false
|
||||
var _stats: Dictionary = {} # player_index → {combats:int, units_lost:int}
|
||||
var _prev_turn_stats: Dictionary = {} # player_index → {pop:int, gold:int, mil:int}
|
||||
var _starved_this_turn: Dictionary = {} # player_index → bool
|
||||
var _violations: Array[String] = []
|
||||
|
||||
|
||||
func _ready() -> void:
|
||||
|
|
@ -97,6 +81,8 @@ func _ready() -> void:
|
|||
EventBus.victory_achieved.connect(_on_victory)
|
||||
EventBus.combat_resolved.connect(_on_combat)
|
||||
EventBus.city_starved.connect(_on_city_starved)
|
||||
EventBus.city_founded.connect(_on_city_founded)
|
||||
EventBus.city_captured.connect(_on_city_captured)
|
||||
|
||||
|
||||
func _on_combat(attacker: Variant, defender: Variant, result: Dictionary) -> void:
|
||||
|
|
@ -105,16 +91,25 @@ func _on_combat(attacker: Variant, defender: Variant, result: Dictionary) -> voi
|
|||
str(result.get("defender_hp", "N/A")),
|
||||
str(result.get("defender_killed", "N/A")),
|
||||
])
|
||||
_total_combats += 1
|
||||
if _turn_first_combat < 0:
|
||||
_turn_first_combat = _turn_count
|
||||
var defender_killed: bool = result.get("defender_killed", false) == true
|
||||
var attacker_killed: bool = result.get("attacker_killed", false) == true
|
||||
if attacker != null and attacker.get("owner") != null:
|
||||
var atk_idx: int = int(attacker.get("owner"))
|
||||
_ensure_stats(atk_idx)
|
||||
_stats[atk_idx]["combats"] = int(_stats[atk_idx].get("combats", 0)) + 1
|
||||
if defender_killed:
|
||||
_stats[atk_idx]["kills"] = int(_stats[atk_idx].get("kills", 0)) + 1
|
||||
if attacker_killed:
|
||||
_stats[atk_idx]["units_lost"] = int(_stats[atk_idx].get("units_lost", 0)) + 1
|
||||
if defender != null and defender.get("owner") != null:
|
||||
var def_idx: int = int(defender.get("owner"))
|
||||
_ensure_stats(def_idx)
|
||||
_stats[def_idx]["combats"] = int(_stats[def_idx].get("combats", 0)) + 1
|
||||
if result.get("defender_killed", false) == true:
|
||||
if defender_killed:
|
||||
_stats[def_idx]["units_lost"] = int(_stats[def_idx].get("units_lost", 0)) + 1
|
||||
if attacker_killed:
|
||||
_stats[def_idx]["kills"] = int(_stats[def_idx].get("kills", 0)) + 1
|
||||
|
||||
|
||||
func _on_city_starved(city: Variant, _new_pop: int) -> void:
|
||||
|
|
@ -124,9 +119,38 @@ func _on_city_starved(city: Variant, _new_pop: int) -> void:
|
|||
_starved_this_turn[idx] = true
|
||||
|
||||
|
||||
func _on_city_founded(_city: Variant, _player_index: int) -> void:
|
||||
_total_cities_founded += 1
|
||||
|
||||
|
||||
func _on_city_captured(_city: Variant, old_owner: int, new_owner: int) -> void:
|
||||
_total_cities_captured += 1
|
||||
if _turn_first_city_captured < 0:
|
||||
_turn_first_city_captured = _turn_count
|
||||
if new_owner >= 0:
|
||||
_ensure_stats(new_owner)
|
||||
_stats[new_owner]["cities_captured"] = (
|
||||
int(_stats[new_owner].get("cities_captured", 0)) + 1
|
||||
)
|
||||
if old_owner >= 0:
|
||||
_ensure_stats(old_owner)
|
||||
_stats[old_owner]["cities_lost"] = (
|
||||
int(_stats[old_owner].get("cities_lost", 0)) + 1
|
||||
)
|
||||
|
||||
|
||||
func _ensure_stats(player_index: int) -> void:
|
||||
if not _stats.has(player_index):
|
||||
_stats[player_index] = {"combats": 0, "units_lost": 0}
|
||||
_stats[player_index] = {
|
||||
"kills": 0,
|
||||
"units_lost": 0,
|
||||
"cities_captured": 0,
|
||||
"cities_lost": 0,
|
||||
"pop_peak": 0,
|
||||
"gold_peak": 0,
|
||||
"turn_first_pop_3": -1,
|
||||
"turn_first_pop_4": -1,
|
||||
}
|
||||
|
||||
|
||||
func _on_victory(player_index: int, victory_type: String) -> void:
|
||||
|
|
@ -1249,9 +1273,9 @@ func _fail(msg: String) -> void:
|
|||
# ── Invariants & Result Writer ───────────────────────────────────────
|
||||
|
||||
func _check_invariants(player: RefCounted) -> void:
|
||||
## Per-turn invariant checks for the current player. Captures violations
|
||||
## into `_violations` without aborting — failures are reported in the
|
||||
## JSON summary so the batch runner can grade runs.
|
||||
## Per-turn invariant checks and peak/milestone tracking for the current
|
||||
## player. Violations are captured into `_violations` without aborting —
|
||||
## failures are reported in the JSON summary so the batch runner can grade runs.
|
||||
var idx: int = player.index
|
||||
var pop: int = 0
|
||||
for c: Variant in player.cities:
|
||||
|
|
@ -1262,6 +1286,7 @@ func _check_invariants(player: RefCounted) -> void:
|
|||
if u.is_alive() and u.get("can_found_city") != true:
|
||||
mil += 1
|
||||
|
||||
# Invariant checks (require prior-turn baseline)
|
||||
if _prev_turn_stats.has(idx):
|
||||
var prev: Dictionary = _prev_turn_stats[idx]
|
||||
var prev_pop: int = int(prev.get("pop", pop))
|
||||
|
|
@ -1270,41 +1295,85 @@ func _check_invariants(player: RefCounted) -> void:
|
|||
"turn_%d: player %d pop dropped %d→%d without starvation event"
|
||||
% [_turn_count, idx, prev_pop, pop]
|
||||
)
|
||||
var techs: int = player.researched_techs.size()
|
||||
var floor_val: int = -max(5, techs * 3)
|
||||
var techs_prev: int = player.researched_techs.size()
|
||||
var floor_val: int = -max(5, techs_prev * 3)
|
||||
if gold < floor_val:
|
||||
_violations.append(
|
||||
"turn_%d: player %d gold=%d below deficit floor %d"
|
||||
% [_turn_count, idx, gold, floor_val]
|
||||
)
|
||||
|
||||
# Peak/milestone tracking
|
||||
_ensure_stats(idx)
|
||||
var pstat: Dictionary = _stats[idx]
|
||||
if pop > int(pstat.get("pop_peak", 0)):
|
||||
pstat["pop_peak"] = pop
|
||||
if gold > int(pstat.get("gold_peak", 0)):
|
||||
pstat["gold_peak"] = gold
|
||||
if int(pstat.get("turn_first_pop_3", -1)) < 0 and pop >= 3:
|
||||
pstat["turn_first_pop_3"] = _turn_count
|
||||
if int(pstat.get("turn_first_pop_4", -1)) < 0 and pop >= 4:
|
||||
pstat["turn_first_pop_4"] = _turn_count
|
||||
|
||||
_prev_turn_stats[idx] = {"pop": pop, "gold": gold, "mil": mil}
|
||||
_starved_this_turn[idx] = false
|
||||
|
||||
|
||||
func _build_final_stats() -> Dictionary:
|
||||
func _build_player_stats() -> Dictionary:
|
||||
var game_map: RefCounted = GameState.get_game_map()
|
||||
var out: Dictionary = {}
|
||||
for p: Variant in GameState.players:
|
||||
var idx: int = int(p.index)
|
||||
var pop: int = 0
|
||||
var tiles: int = 0
|
||||
var buildings: int = 0
|
||||
var food_total: float = 0.0
|
||||
var production_total: float = 0.0
|
||||
for c: Variant in p.cities:
|
||||
pop += int(c.population)
|
||||
tiles += int(c.owned_tiles.size())
|
||||
buildings += int(c.buildings.size())
|
||||
if game_map != null:
|
||||
var tile_json: String = BuildableHelperScript.build_tile_yields_json(
|
||||
c, game_map
|
||||
)
|
||||
var cy: Dictionary = c.get_yields(tile_json)
|
||||
food_total += float(cy.get("food", 0))
|
||||
production_total += float(cy.get("production", 0))
|
||||
var mil: int = 0
|
||||
for u: Variant in p.units:
|
||||
if u.is_alive() and u.get("can_found_city") != true:
|
||||
mil += 1
|
||||
_ensure_stats(idx)
|
||||
var pstat: Dictionary = _stats[idx]
|
||||
var happiness: int = int(p.get("happiness")) if p.get("happiness") != null else 0
|
||||
var gpt: int = int(p.get("gold_per_turn")) if p.get("gold_per_turn") != null else 0
|
||||
var gold: int = int(p.get("gold")) if p.get("gold") != null else 0
|
||||
# Keep peak in sync with current (final-stats is also called at exit)
|
||||
if pop > int(pstat.get("pop_peak", 0)):
|
||||
pstat["pop_peak"] = pop
|
||||
if gold > int(pstat.get("gold_peak", 0)):
|
||||
pstat["gold_peak"] = gold
|
||||
out[str(idx)] = {
|
||||
"pop": pop,
|
||||
"pop_peak": int(pstat.get("pop_peak", pop)),
|
||||
"mil": mil,
|
||||
"cities": int(p.cities.size()),
|
||||
"gold": int(p.get("gold")) if p.get("gold") != null else 0,
|
||||
"cities_captured": int(pstat.get("cities_captured", 0)),
|
||||
"cities_lost": int(pstat.get("cities_lost", 0)),
|
||||
"gold": gold,
|
||||
"gold_peak": int(pstat.get("gold_peak", gold)),
|
||||
"gold_per_turn": gpt,
|
||||
"techs": int(p.researched_techs.size()),
|
||||
"tiles": tiles,
|
||||
"combats": int(_stats[idx].get("combats", 0)),
|
||||
"units_lost": int(_stats[idx].get("units_lost", 0)),
|
||||
"buildings": buildings,
|
||||
"happiness": happiness,
|
||||
"food_total": food_total,
|
||||
"production_total": production_total,
|
||||
"kills": int(pstat.get("kills", 0)),
|
||||
"units_lost": int(pstat.get("units_lost", 0)),
|
||||
"turn_first_pop_3": int(pstat.get("turn_first_pop_3", -1)),
|
||||
"turn_first_pop_4": int(pstat.get("turn_first_pop_4", -1)),
|
||||
}
|
||||
return out
|
||||
|
||||
|
|
@ -1317,6 +1386,13 @@ func _write_result(outcome: String) -> void:
|
|||
return
|
||||
_result_written = true
|
||||
var wall_clock: float = Time.get_unix_time_from_system() - _start_time
|
||||
var aggregate: Dictionary = {
|
||||
"total_combats": _total_combats,
|
||||
"total_cities_founded": _total_cities_founded,
|
||||
"total_cities_captured": _total_cities_captured,
|
||||
"turn_first_combat": _turn_first_combat,
|
||||
"turn_first_city_captured": _turn_first_city_captured,
|
||||
}
|
||||
var result: Dictionary = {
|
||||
"seed": _seed,
|
||||
"turns_played": _turn_count,
|
||||
|
|
@ -1324,6 +1400,7 @@ func _write_result(outcome: String) -> void:
|
|||
"winner_index": _victory_winner,
|
||||
"victory_type": _victory_type,
|
||||
"wall_clock_sec": wall_clock,
|
||||
"aggregate": aggregate,
|
||||
"final_stats": _build_final_stats(),
|
||||
"invariant_violations": _violations,
|
||||
}
|
||||
|
|
|
|||
|
|
@ -147,11 +147,15 @@ for seed in $(seq 1 "$COUNT"); do
|
|||
_run_local "$seed" "$seed_dir"
|
||||
fi
|
||||
|
||||
result_file="$seed_dir/result_${seed}.json"
|
||||
if [ -f "$result_file" ]; then
|
||||
# Look for timestamped result file (result_<stamp>_seed<N>.json) or legacy (result_<N>.json)
|
||||
result_file="$(ls -1 "$seed_dir"/result_*_seed${seed}.json 2>/dev/null | tail -n1)"
|
||||
if [ -z "$result_file" ] && [ -f "$seed_dir/result_${seed}.json" ]; then
|
||||
result_file="$seed_dir/result_${seed}.json"
|
||||
fi
|
||||
if [ -n "$result_file" ] && [ -f "$result_file" ]; then
|
||||
echo "[seed $seed] OK — result written to $result_file"
|
||||
else
|
||||
echo "[seed $seed] MISSING result_${seed}.json (Task 1 may not be complete, or game crashed)" >&2
|
||||
echo "[seed $seed] MISSING result file (no result_*_seed${seed}.json found)" >&2
|
||||
FAILED_SEEDS+=("$seed")
|
||||
fi
|
||||
done
|
||||
|
|
|
|||
|
|
@ -21,7 +21,10 @@ from typing import Any
|
|||
|
||||
|
||||
def find_result_files(results_dir: Path) -> list[tuple[int, Path]]:
|
||||
"""Return (seed, path) pairs for all result_*.json files, sorted by seed."""
|
||||
"""Return (seed, path) pairs for all result_*.json files, sorted by seed.
|
||||
Matches filenames of the form result_<datetime>_seed<N>.json under
|
||||
seed_<N>/ subdirectories. Picks the most recent (lexicographic max) if
|
||||
multiple stamps exist for the same seed."""
|
||||
results: list[tuple[int, Path]] = []
|
||||
for seed_dir in sorted(results_dir.glob("seed_*")):
|
||||
if not seed_dir.is_dir():
|
||||
|
|
@ -30,8 +33,15 @@ def find_result_files(results_dir: Path) -> list[tuple[int, Path]]:
|
|||
if not seed_str.isdigit():
|
||||
continue
|
||||
seed = int(seed_str)
|
||||
result_file = seed_dir / f"result_{seed}.json"
|
||||
results.append((seed, result_file))
|
||||
candidates = sorted(seed_dir.glob(f"result_*_seed{seed}.json"))
|
||||
if not candidates:
|
||||
# Fall back to legacy naming for backward compatibility
|
||||
legacy = seed_dir / f"result_{seed}.json"
|
||||
if legacy.exists():
|
||||
candidates = [legacy]
|
||||
if not candidates:
|
||||
continue
|
||||
results.append((seed, candidates[-1]))
|
||||
return results
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -13,7 +13,7 @@
|
|||
"victory_type",
|
||||
"wall_clock_sec",
|
||||
"aggregate",
|
||||
"player_stats",
|
||||
"final_stats",
|
||||
"invariant_violations"
|
||||
],
|
||||
"properties": {
|
||||
|
|
@ -44,9 +44,9 @@
|
|||
"turn_first_city_captured": { "type": "integer", "minimum": -1 }
|
||||
}
|
||||
},
|
||||
"player_stats": {
|
||||
"final_stats": {
|
||||
"type": "object",
|
||||
"description": "Map of player_index (as string) to per-player stats. Rewritten every turn — not final until outcome != 'in_progress'.",
|
||||
"description": "Map of player_index (as string) to per-player stats",
|
||||
"propertyNames": { "pattern": "^[0-9]+$" },
|
||||
"additionalProperties": {
|
||||
"$ref": "#/definitions/player_stats"
|
||||
|
|
|
|||
161
tools/autoplay-validate.py
Executable file → Normal file
161
tools/autoplay-validate.py
Executable file → Normal file
|
|
@ -1,31 +1,22 @@
|
|||
#!/usr/bin/env python3
|
||||
"""
|
||||
JSON Schema validator for autoplay output files.
|
||||
Minimal JSON Schema validator for autoplay-result-schema.json.
|
||||
|
||||
Implements the subset of draft-07 used by the schemas:
|
||||
Implements the subset of draft-07 used by the schema:
|
||||
type, required, additionalProperties, properties, propertyNames.pattern,
|
||||
minimum, enum, items, pattern, $ref (local only).
|
||||
minimum, enum, items, $ref (local only).
|
||||
|
||||
stdlib only — no pip installs.
|
||||
|
||||
Usage:
|
||||
# Validate a single JSON file against a named schema:
|
||||
python3 tools/autoplay-validate.py --schema meta path/to/meta.json
|
||||
from autoplay_validate import load_schema, validate
|
||||
schema = load_schema()
|
||||
errors = validate(data, schema)
|
||||
if errors: ...
|
||||
|
||||
# Validate every line of a JSONL file independently:
|
||||
python3 tools/autoplay-validate.py --schema turn-stats-line --jsonl path/to/turn_stats.jsonl
|
||||
|
||||
# Legacy: validate against the flat result schema (default):
|
||||
CLI:
|
||||
python3 tools/autoplay-validate.py path/to/result.json
|
||||
|
||||
Exits 0 if all valid, 1 with errors to stderr, 2 on usage error.
|
||||
|
||||
Available schema names (--schema):
|
||||
turn-stats-line tools/schemas/autoplay/turn-stats-line.json
|
||||
meta tools/schemas/autoplay/meta.json
|
||||
events-line tools/schemas/autoplay/events-line.json
|
||||
save tools/schemas/autoplay/save.json
|
||||
result tools/autoplay-result-schema.json (legacy flat schema)
|
||||
exits 0 if valid, 1 with errors to stderr otherwise.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
|
|
@ -35,26 +26,10 @@ import sys
|
|||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
TOOLS_DIR = Path(__file__).parent
|
||||
SCHEMAS_DIR = TOOLS_DIR / "schemas" / "autoplay"
|
||||
|
||||
SCHEMA_PATHS: dict[str, Path] = {
|
||||
"result": TOOLS_DIR / "autoplay-result-schema.json",
|
||||
"turn-stats-line": SCHEMAS_DIR / "turn-stats-line.json",
|
||||
"meta": SCHEMAS_DIR / "meta.json",
|
||||
"events-line": SCHEMAS_DIR / "events-line.json",
|
||||
"save": SCHEMAS_DIR / "save.json",
|
||||
}
|
||||
|
||||
_DEFAULT_SCHEMA = "result"
|
||||
SCHEMA_PATH = Path(__file__).parent / "autoplay-result-schema.json"
|
||||
|
||||
|
||||
def load_schema(name: str = _DEFAULT_SCHEMA) -> dict[str, Any]:
|
||||
path = SCHEMA_PATHS.get(name)
|
||||
if path is None:
|
||||
raise ValueError(
|
||||
f"unknown schema {name!r}. Available: {', '.join(sorted(SCHEMA_PATHS))}"
|
||||
)
|
||||
def load_schema(path: Path = SCHEMA_PATH) -> dict[str, Any]:
|
||||
with path.open() as f:
|
||||
return json.load(f)
|
||||
|
||||
|
|
@ -71,6 +46,7 @@ _TYPE_CHECKS: dict[str, type | tuple[type, ...]] = {
|
|||
|
||||
|
||||
def _resolve_ref(ref: str, root: dict[str, Any]) -> dict[str, Any]:
|
||||
# Local refs only: "#/definitions/foo"
|
||||
if not ref.startswith("#/"):
|
||||
raise ValueError(f"only local refs supported, got {ref!r}")
|
||||
node: Any = root
|
||||
|
|
@ -95,7 +71,7 @@ def _validate(
|
|||
if expected is None:
|
||||
errors.append(f"{path}: unknown schema type {t!r}")
|
||||
return errors
|
||||
# bool is a subclass of int in Python; reject booleans as numbers.
|
||||
# JSON has no separate int — bool is a subclass of int in Python; reject booleans as numbers.
|
||||
if t in ("integer", "number") and isinstance(value, bool):
|
||||
errors.append(f"{path}: expected {t}, got boolean")
|
||||
return errors
|
||||
|
|
@ -103,7 +79,9 @@ def _validate(
|
|||
errors.append(f"{path}: expected integer, got float {value}")
|
||||
return errors
|
||||
if not isinstance(value, expected):
|
||||
errors.append(f"{path}: expected {t}, got {type(value).__name__}")
|
||||
errors.append(
|
||||
f"{path}: expected {t}, got {type(value).__name__}"
|
||||
)
|
||||
return errors
|
||||
|
||||
if "enum" in schema:
|
||||
|
|
@ -112,13 +90,13 @@ def _validate(
|
|||
|
||||
if "minimum" in schema and isinstance(value, (int, float)):
|
||||
if value < schema["minimum"]:
|
||||
errors.append(f"{path}: {value} < minimum {schema['minimum']}")
|
||||
errors.append(
|
||||
f"{path}: {value} < minimum {schema['minimum']}"
|
||||
)
|
||||
|
||||
if "pattern" in schema and isinstance(value, str):
|
||||
if not re.match(schema["pattern"], value):
|
||||
errors.append(
|
||||
f"{path}: {value!r} does not match pattern {schema['pattern']!r}"
|
||||
)
|
||||
errors.append(f"{path}: {value!r} does not match pattern {schema['pattern']!r}")
|
||||
|
||||
if t == "object" and isinstance(value, dict):
|
||||
props: dict[str, Any] = schema.get("properties", {})
|
||||
|
|
@ -151,43 +129,25 @@ def _validate(
|
|||
|
||||
|
||||
def validate(data: Any, schema: dict[str, Any] | None = None) -> list[str]:
|
||||
"""Validate data against schema. Returns list of error strings (empty = valid)."""
|
||||
"""Validate data against the schema. Returns list of error strings (empty = valid)."""
|
||||
s = schema if schema is not None else load_schema()
|
||||
return _validate(data, s, s, "$")
|
||||
|
||||
|
||||
def _validate_file(path: Path, schema: dict[str, Any], jsonl: bool) -> int:
|
||||
"""Validate one file. Returns error count."""
|
||||
def _main(argv: list[str]) -> int:
|
||||
if len(argv) < 2:
|
||||
print("usage: autoplay-validate.py <result.json> [<result.json> ...]", file=sys.stderr)
|
||||
return 2
|
||||
schema = load_schema()
|
||||
total_errors = 0
|
||||
try:
|
||||
text = path.read_text()
|
||||
except OSError as e:
|
||||
print(f"{path}: cannot read ({e})", file=sys.stderr)
|
||||
return 1
|
||||
|
||||
if jsonl:
|
||||
for lineno, raw in enumerate(text.splitlines(), start=1):
|
||||
raw = raw.strip()
|
||||
if not raw:
|
||||
continue
|
||||
try:
|
||||
data = json.loads(raw)
|
||||
except json.JSONDecodeError as e:
|
||||
print(f"{path}:{lineno}: invalid JSON ({e})", file=sys.stderr)
|
||||
total_errors += 1
|
||||
continue
|
||||
errs = validate(data, schema)
|
||||
if errs:
|
||||
total_errors += len(errs)
|
||||
print(f"{path}:{lineno}: {len(errs)} error(s)", file=sys.stderr)
|
||||
for e in errs:
|
||||
print(f" {e}", file=sys.stderr)
|
||||
else:
|
||||
for p in argv[1:]:
|
||||
path = Path(p)
|
||||
try:
|
||||
data = json.loads(text)
|
||||
except json.JSONDecodeError as e:
|
||||
print(f"{path}: invalid JSON ({e})", file=sys.stderr)
|
||||
return 1
|
||||
data = json.loads(path.read_text())
|
||||
except (OSError, json.JSONDecodeError) as e:
|
||||
print(f"{path}: cannot load ({e})", file=sys.stderr)
|
||||
total_errors += 1
|
||||
continue
|
||||
errs = validate(data, schema)
|
||||
if errs:
|
||||
total_errors += len(errs)
|
||||
|
|
@ -196,61 +156,6 @@ def _validate_file(path: Path, schema: dict[str, Any], jsonl: bool) -> int:
|
|||
print(f" {e}", file=sys.stderr)
|
||||
else:
|
||||
print(f"{path}: OK", file=sys.stderr)
|
||||
|
||||
if jsonl and total_errors == 0:
|
||||
print(f"{path}: OK", file=sys.stderr)
|
||||
|
||||
return total_errors
|
||||
|
||||
|
||||
def _main(argv: list[str]) -> int:
|
||||
args = argv[1:]
|
||||
|
||||
schema_name = _DEFAULT_SCHEMA
|
||||
jsonl = False
|
||||
files: list[str] = []
|
||||
|
||||
i = 0
|
||||
while i < len(args):
|
||||
a = args[i]
|
||||
if a == "--schema":
|
||||
i += 1
|
||||
if i >= len(args):
|
||||
print("ERROR: --schema requires a value", file=sys.stderr)
|
||||
return 2
|
||||
schema_name = args[i]
|
||||
elif a == "--jsonl":
|
||||
jsonl = True
|
||||
elif a.startswith("--schema="):
|
||||
schema_name = a[len("--schema="):]
|
||||
elif a.startswith("-"):
|
||||
print(f"ERROR: unknown flag {a!r}", file=sys.stderr)
|
||||
return 2
|
||||
else:
|
||||
files.append(a)
|
||||
i += 1
|
||||
|
||||
if not files:
|
||||
print(
|
||||
"usage: autoplay-validate.py [--schema NAME] [--jsonl] <file> [<file> ...]",
|
||||
file=sys.stderr,
|
||||
)
|
||||
print(
|
||||
f" schemas: {', '.join(sorted(SCHEMA_PATHS))}",
|
||||
file=sys.stderr,
|
||||
)
|
||||
return 2
|
||||
|
||||
try:
|
||||
schema = load_schema(schema_name)
|
||||
except ValueError as e:
|
||||
print(f"ERROR: {e}", file=sys.stderr)
|
||||
return 2
|
||||
|
||||
total_errors = 0
|
||||
for f in files:
|
||||
total_errors += _validate_file(Path(f), schema, jsonl)
|
||||
|
||||
return 1 if total_errors else 0
|
||||
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue