1"""Validation of submission files against the schema and the challenge output spec."""
2from __future__ import annotations
4import json
5from importlib import resources
6from pathlib import Path
8import jsonschema
10from .loading import Source, load_challenges, load_submissions, public_fields
11from .wav import read_wav, resolve_output
14def _schema() -> dict:
15 with resources.files("challenge_tools").joinpath("solution.schema.json").open() as f:
16 return json.load(f)
19def _acyclic(edges: list[tuple[str, str]], nodes: list[str]) -> bool:
20 adj = {n: [] for n in nodes}
21 for a, b in edges:
22 adj[a].append(b)
23 state = {n: 0 for n in nodes}
25 def visit(n):
26 state[n] = 1
27 for m in adj[n]:
28 if state[m] == 1 or (state[m] == 0 and not visit(m)):
29 return False
30 state[n] = 2
31 return True
33 return all(state[n] == 2 or visit(n) for n in nodes)
36def check_fields_file(path: Path, challenge: dict) -> list[str]:
37 """Open an optional HDF5 field export and check the datasets against the challenge's grid."""
38 import h5py
40 problems = []
41 spec = challenge.get("fields")
42 with h5py.File(path, "r") as f:
43 for name in ("x", "t", "u"):
44 if name not in f:
45 problems.append(f"fields file lacks dataset {name!r}")
46 if problems:
47 return problems
48 nx, nt = f["x"].shape[0], f["t"].shape[0]
49 if f["u"].shape != (nt, nx):
50 problems.append(f"fields dataset u has shape {f['u'].shape}, expected {(nt, nx)}")
51 if spec:
52 if nx != spec["nx"] or nt != spec["nt"]:
53 problems.append(f"fields grid is {nx} x {nt}, challenge recommends nx={spec['nx']} nt={spec['nt']}")
54 return problems
57def validate(challenges_dir: Path, sources: list[Source], check_outputs: bool = True) -> list[str]:
58 """Return a list of problems (empty means valid)."""
59 problems: list[str] = []
60 challenges = load_challenges(challenges_dir)
61 if not challenges:
62 problems.append(f"no challenge.json found under {challenges_dir}")
63 schema = _schema()
64 validator = jsonschema.Draft202012Validator(schema, format_checker=jsonschema.FormatChecker())
65 seen = set()
66 for sub in load_submissions(sources):
67 path: Path = sub["_path"]
68 tag = str(path)
69 errs = sorted(validator.iter_errors(public_fields(sub)), key=str)
70 for e in errs:
71 loc = "/".join(str(x) for x in e.absolute_path)
72 problems.append(f"{tag}: schema: {loc}: {e.message}")
73 if errs:
74 continue
75 if sub["submission_id"] != path.stem:
76 problems.append(f"{tag}: submission_id {sub['submission_id']!r} does not match file name")
77 if sub["challenge_id"] != sub["_dir_challenge"]:
78 problems.append(f"{tag}: challenge_id {sub['challenge_id']!r} does not match directory {sub['_dir_challenge']!r}")
79 if sub["challenge_id"] not in challenges:
80 problems.append(f"{tag}: unknown challenge_id {sub['challenge_id']!r}")
81 continue
82 key = (sub["challenge_id"], sub["_key"])
83 if key in seen:
84 problems.append(f"{tag}: duplicate submission_id for this challenge")
85 seen.add(key)
86 ids = [s["solution_id"] for s in sub["solutions"]]
87 if len(set(ids)) != len(ids):
88 problems.append(f"{tag}: duplicate solution_id")
89 edges = []
90 for s in sub["solutions"]:
91 for worse in s.get("more_accurate_than", []):
92 if worse not in ids:
93 problems.append(f"{tag}: {s['solution_id']}: more_accurate_than names unknown solution {worse!r}")
94 elif worse == s["solution_id"]:
95 problems.append(f"{tag}: {s['solution_id']}: more_accurate_than names itself")
96 else:
97 edges.append((worse, s["solution_id"]))
98 if not _acyclic(edges, ids):
99 problems.append(f"{tag}: more_accurate_than contains a cycle")
100 if check_outputs:
101 challenge = challenges[sub["challenge_id"]]
102 for s in sub["solutions"]:
103 try:
104 p = resolve_output(s["output"], path.parent)
105 if not p.exists():
106 problems.append(f"{tag}: {s['solution_id']}: output not found: {s['output']}")
107 continue
108 read_wav(p, challenge["output"])
109 except Exception as e: # noqa: BLE001
110 problems.append(f"{tag}: {s['solution_id']}: {e}")
111 if "fields" in s:
112 try:
113 fp = resolve_output(s["fields"]["path"], path.parent)
114 if not fp.exists():
115 problems.append(f"{tag}: {s['solution_id']}: fields file not found: {s['fields']['path']}")
116 else:
117 for pr in check_fields_file(fp, challenge):
118 problems.append(f"{tag}: {s['solution_id']}: {pr}")
119 except Exception as e: # noqa: BLE001
120 problems.append(f"{tag}: {s['solution_id']}: fields: {e}")
121 return problems