ddfc839Add string-pluck-v1 challenge, solution schema, and scoring toolsJeremy Magland 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 load_challenges, load_submissions
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 validate(challenges_dir: Path, solutions_dir: Path, check_outputs: bool = True) -> list[str]:
37 """Return a list of problems (empty means valid)."""
38 problems: list[str] = []
39 challenges = load_challenges(challenges_dir)
40 if not challenges:
41 problems.append(f"no challenge.json found under {challenges_dir}")
42 schema = _schema()
43 validator = jsonschema.Draft202012Validator(schema, format_checker=jsonschema.FormatChecker())
44 seen = set()
45 for sub in load_submissions(solutions_dir):
46 path: Path = sub["_path"]
47 tag = str(path)
48 errs = sorted(validator.iter_errors({k: v for k, v in sub.items() if not k.startswith("_")}), key=str)
49 for e in errs:
50 loc = "/".join(str(x) for x in e.absolute_path)
51 problems.append(f"{tag}: schema: {loc}: {e.message}")
52 if errs:
53 continue
54 if sub["submission_id"] != path.stem:
55 problems.append(f"{tag}: submission_id {sub['submission_id']!r} does not match file name")
56 if sub["challenge_id"] != sub["_dir_challenge"]:
57 problems.append(f"{tag}: challenge_id {sub['challenge_id']!r} does not match directory {sub['_dir_challenge']!r}")
58 if sub["challenge_id"] not in challenges:
59 problems.append(f"{tag}: unknown challenge_id {sub['challenge_id']!r}")
60 continue
61 key = (sub["challenge_id"], sub["submission_id"])
62 if key in seen:
63 problems.append(f"{tag}: duplicate submission_id for this challenge")
64 seen.add(key)
65 ids = [s["solution_id"] for s in sub["solutions"]]
66 if len(set(ids)) != len(ids):
67 problems.append(f"{tag}: duplicate solution_id")
68 edges = []
69 for s in sub["solutions"]:
70 for worse in s.get("more_accurate_than", []):
71 if worse not in ids:
72 problems.append(f"{tag}: {s['solution_id']}: more_accurate_than names unknown solution {worse!r}")
73 elif worse == s["solution_id"]:
74 problems.append(f"{tag}: {s['solution_id']}: more_accurate_than names itself")
75 else:
76 edges.append((worse, s["solution_id"]))
77 if not _acyclic(edges, ids):
78 problems.append(f"{tag}: more_accurate_than contains a cycle")
79 if check_outputs:
80 spec = challenges[sub["challenge_id"]]["output"]
81 for s in sub["solutions"]:
82 try:
83 p = resolve_output(s["output"], path.parent)
84 if not p.exists():
85 problems.append(f"{tag}: {s['solution_id']}: output not found: {s['output']}")
86 continue
87 read_wav(p, spec)
88 except Exception as e: # noqa: BLE001
89 problems.append(f"{tag}: {s['solution_id']}: {e}")
90 return problems