Fabio
c36510c21e
Some checks failed
Linux Build / Linux cpu (push) Has been cancelled
Linux Build / Linux cpu-clang (push) Has been cancelled
Linux Build / Linux vulkan (push) Has been cancelled
mac-build / macOS CPU (push) Has been cancelled
Nix Build / Nix (cpu) (push) Has been cancelled
Nix Build / Nix (vulkan) (push) Has been cancelled
windows-build / Windows CPU (push) Has been cancelled
307 lines
12 KiB
Python
307 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
"""Check sync between runtime loaders, model_specs packages, and model_manager_v2.
|
|
|
|
Schema correctness is owned by the typed model-spec validator. This script only
|
|
checks cross-system drift:
|
|
|
|
- registered loader families vs model_specs families
|
|
- model_specs packages vs model_manager_v2 package output
|
|
See docs/maintainers/loader_and_catalog.md.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import re
|
|
import sys
|
|
import unittest
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parents[1]
|
|
CMAKE_PATH = REPO_ROOT / "CMakeLists.txt"
|
|
REGISTRY_PATH = REPO_ROOT / "src" / "framework" / "runtime" / "registry.cpp"
|
|
SPECS_DIR = REPO_ROOT / "model_specs"
|
|
|
|
_LOADER_CALL_RE = re.compile(r"\bmake_([a-z0-9_]+)_loader(?:\s*\(\s*\))?")
|
|
|
|
BUNDLED_LOADERS_WITHOUT_SPEC = {
|
|
"earshot_vad",
|
|
"marblenet_vad",
|
|
"silero_vad",
|
|
}
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class SpecPackage:
|
|
family: str
|
|
id: str
|
|
format: str
|
|
target_directory: str
|
|
default: bool
|
|
|
|
|
|
def rel(path: Path) -> str:
|
|
try:
|
|
return str(path.relative_to(REPO_ROOT))
|
|
except ValueError:
|
|
return str(path)
|
|
|
|
|
|
def parse_loader_declarations(text: str, comment_prefix: str) -> tuple[set[str], set[str]]:
|
|
active: set[str] = set()
|
|
commented: set[str] = set()
|
|
for raw_line in text.splitlines():
|
|
line = raw_line.strip()
|
|
match = _LOADER_CALL_RE.search(line)
|
|
if not match:
|
|
continue
|
|
family = match.group(1)
|
|
if line.startswith(comment_prefix):
|
|
commented.add(family)
|
|
else:
|
|
active.add(family)
|
|
return active, commented
|
|
|
|
|
|
def parse_declared_loaders(cmake_text: str, registry_text: str) -> tuple[set[str], set[str]]:
|
|
cmake_active, cmake_commented = parse_loader_declarations(cmake_text, "#")
|
|
registry_active, registry_commented = parse_loader_declarations(registry_text, "//")
|
|
return cmake_active | registry_active, cmake_commented | registry_commented
|
|
|
|
|
|
def loader_families_from_json(payload: Any) -> set[str]:
|
|
if not isinstance(payload, list):
|
|
raise ValueError("loader JSON must be a list")
|
|
families: set[str] = set()
|
|
for index, row in enumerate(payload):
|
|
if not isinstance(row, dict) or not isinstance(row.get("family"), str):
|
|
raise ValueError(f"loader JSON row {index} must contain string family")
|
|
families.add(row["family"])
|
|
return families
|
|
|
|
|
|
def load_json(path: Path) -> Any:
|
|
with path.open("r", encoding="utf-8") as handle:
|
|
return json.load(handle)
|
|
|
|
|
|
def load_spec_packages(specs_dir: Path) -> tuple[dict[str, dict[str, Any]], dict[str, SpecPackage], list[str]]:
|
|
specs_by_family: dict[str, dict[str, Any]] = {}
|
|
packages_by_id: dict[str, SpecPackage] = {}
|
|
errors: list[str] = []
|
|
for path in sorted(specs_dir.glob("*.json")):
|
|
try:
|
|
spec = load_json(path)
|
|
except json.JSONDecodeError as exc:
|
|
errors.append(f"{rel(path)}: invalid JSON: {exc}")
|
|
continue
|
|
if not isinstance(spec, dict):
|
|
errors.append(f"{rel(path)}: top-level JSON must be an object")
|
|
continue
|
|
family = spec.get("family")
|
|
if not isinstance(family, str) or not family:
|
|
errors.append(f"{rel(path)}: missing family")
|
|
continue
|
|
if family != path.stem:
|
|
errors.append(f"{rel(path)}: family '{family}' does not match filename stem '{path.stem}'")
|
|
if family in specs_by_family:
|
|
errors.append(f"{rel(path)}: duplicate spec family '{family}'")
|
|
specs_by_family[family] = spec
|
|
|
|
packages = spec.get("packages", [])
|
|
if packages is None:
|
|
packages = []
|
|
if not isinstance(packages, list):
|
|
errors.append(f"{rel(path)}: packages must be a list for model_manager_v2")
|
|
continue
|
|
for index, package in enumerate(packages):
|
|
if not isinstance(package, dict):
|
|
errors.append(f"{rel(path)}: packages[{index}] must be an object for model_manager_v2")
|
|
continue
|
|
package_id = package.get("id")
|
|
if not isinstance(package_id, str) or not package_id:
|
|
errors.append(f"{rel(path)}: packages[{index}] missing id")
|
|
continue
|
|
if package_id in packages_by_id:
|
|
errors.append(
|
|
f"{rel(path)}: duplicate package id '{package_id}' already declared by "
|
|
f"model_specs/{packages_by_id[package_id].family}.json"
|
|
)
|
|
continue
|
|
packages_by_id[package_id] = SpecPackage(
|
|
family=family,
|
|
id=package_id,
|
|
format=str(package.get("format") or ""),
|
|
target_directory=str(package.get("target_directory") or ""),
|
|
default=package.get("default") is True,
|
|
)
|
|
return specs_by_family, packages_by_id, errors
|
|
|
|
|
|
def load_manager_packages(specs_dir: Path) -> tuple[dict[str, Any], list[str]]:
|
|
sys.path.insert(0, str(REPO_ROOT / "tools"))
|
|
import model_manager_v2 # noqa: E402
|
|
|
|
errors: list[str] = []
|
|
try:
|
|
records = model_manager_v2.flatten_packages(model_manager_v2.load_specs(specs_dir))
|
|
except Exception as exc:
|
|
return {}, [f"model_manager_v2 failed to load {rel(specs_dir)}: {exc}"]
|
|
packages: dict[str, Any] = {}
|
|
for record in records:
|
|
if record.id in packages:
|
|
errors.append(f"model_manager_v2 produced duplicate package id '{record.id}'")
|
|
packages[record.id] = record
|
|
return packages, errors
|
|
|
|
|
|
def check_loader_spec_sync(active_loaders: set[str], specs_by_family: dict[str, dict[str, Any]]) -> list[str]:
|
|
errors: list[str] = []
|
|
spec_families = set(specs_by_family)
|
|
for family in sorted(spec_families - active_loaders):
|
|
errors.append(f"model_specs/{family}.json has no registered loader family")
|
|
for family in sorted(active_loaders - spec_families - BUNDLED_LOADERS_WITHOUT_SPEC):
|
|
errors.append(f"registered loader '{family}' has no model_specs/{family}.json")
|
|
return errors
|
|
|
|
|
|
def check_manager_sync(spec_packages: dict[str, SpecPackage], manager_packages: dict[str, Any]) -> tuple[list[str], list[str]]:
|
|
errors: list[str] = []
|
|
warnings: list[str] = []
|
|
spec_ids = set(spec_packages)
|
|
manager_ids = set(manager_packages)
|
|
for package_id in sorted(spec_ids - manager_ids):
|
|
errors.append(f"model_specs package '{package_id}' is missing from model_manager_v2 output")
|
|
for package_id in sorted(manager_ids - spec_ids):
|
|
errors.append(f"model_manager_v2 package '{package_id}' is not declared in model_specs")
|
|
|
|
defaults_by_family: dict[str, list[str]] = {}
|
|
for package in spec_packages.values():
|
|
if package.default:
|
|
defaults_by_family.setdefault(package.family, []).append(package.id)
|
|
if package.format != "gguf":
|
|
warnings.append(f"model_specs/{package.family}.json package '{package.id}' is format={package.format}")
|
|
|
|
for family in sorted({package.family for package in spec_packages.values()}):
|
|
defaults = defaults_by_family.get(family, [])
|
|
if len(defaults) != 1:
|
|
errors.append(f"model_specs/{family}.json must expose exactly one default package for v2 family installs")
|
|
continue
|
|
default_id = defaults[0]
|
|
if spec_packages[default_id].format != "gguf":
|
|
errors.append(f"model_specs/{family}.json default package '{default_id}' must be GGUF")
|
|
|
|
for package_id, spec_package in spec_packages.items():
|
|
manager_package = manager_packages.get(package_id)
|
|
if manager_package is None:
|
|
continue
|
|
if manager_package.family != spec_package.family:
|
|
errors.append(
|
|
f"model_manager_v2 package '{package_id}' family '{manager_package.family}' "
|
|
f"does not match model_specs family '{spec_package.family}'"
|
|
)
|
|
if manager_package.target_directory != spec_package.target_directory:
|
|
errors.append(
|
|
f"model_manager_v2 package '{package_id}' target_directory '{manager_package.target_directory}' "
|
|
f"does not match model_specs target_directory '{spec_package.target_directory}'"
|
|
)
|
|
return errors, warnings
|
|
|
|
|
|
class _SyncCheckSelfTests(unittest.TestCase):
|
|
def test_parse_loader_declarations(self) -> None:
|
|
text = """
|
|
# engine::models::old::make_old_loader
|
|
engine::models::new_family::make_new_family_loader
|
|
engine::models::other::make_other_loader # active trailing comment
|
|
"""
|
|
active, commented = parse_loader_declarations(text, "#")
|
|
self.assertEqual(active, {"new_family", "other"})
|
|
self.assertEqual(commented, {"old"})
|
|
|
|
def test_loader_json_family_parse(self) -> None:
|
|
families = loader_families_from_json([{"family": "a"}, {"family": "b"}])
|
|
self.assertEqual(families, {"a", "b"})
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--cmake", type=Path, default=CMAKE_PATH, help="Path to top-level CMakeLists.txt")
|
|
parser.add_argument("--registry", type=Path, default=REGISTRY_PATH, help="Path to registry.cpp")
|
|
parser.add_argument("--specs-dir", type=Path, default=SPECS_DIR, help="Directory containing model spec JSON files")
|
|
parser.add_argument(
|
|
"--loader-json",
|
|
type=Path,
|
|
default=None,
|
|
help="Optional audiocpp_cli --list-loaders --json output. Use '-' to read stdin.",
|
|
)
|
|
parser.add_argument("--self-test", action="store_true", help="Run built-in unit tests and exit")
|
|
args = parser.parse_args()
|
|
|
|
if args.self_test:
|
|
suite = unittest.defaultTestLoader.loadTestsFromTestCase(_SyncCheckSelfTests)
|
|
result = unittest.TextTestRunner(verbosity=2).run(suite)
|
|
return 0 if result.wasSuccessful() else 1
|
|
|
|
errors: list[str] = []
|
|
warnings: list[str] = []
|
|
if args.loader_json is not None:
|
|
try:
|
|
loader_text = sys.stdin.read() if str(args.loader_json) == "-" else args.loader_json.read_text(encoding="utf-8")
|
|
active_loaders = loader_families_from_json(json.loads(loader_text))
|
|
commented_loaders: set[str] = set()
|
|
except Exception as exc:
|
|
print(f"error: failed to read loader JSON: {exc}", file=sys.stderr)
|
|
return 2
|
|
else:
|
|
if not args.cmake.is_file():
|
|
print(f"error: CMakeLists.txt not found: {args.cmake}", file=sys.stderr)
|
|
return 2
|
|
if not args.registry.is_file():
|
|
print(f"error: registry not found: {args.registry}", file=sys.stderr)
|
|
return 2
|
|
active_loaders, commented_loaders = parse_declared_loaders(
|
|
args.cmake.read_text(encoding="utf-8"),
|
|
args.registry.read_text(encoding="utf-8"),
|
|
)
|
|
if not active_loaders:
|
|
print("error: no active loaders found", file=sys.stderr)
|
|
return 2
|
|
|
|
specs_by_family, spec_packages, spec_errors = load_spec_packages(args.specs_dir)
|
|
manager_packages, manager_errors = load_manager_packages(args.specs_dir)
|
|
errors.extend(spec_errors)
|
|
errors.extend(manager_errors)
|
|
errors.extend(check_loader_spec_sync(active_loaders, specs_by_family))
|
|
manager_sync_errors, manager_sync_warnings = check_manager_sync(spec_packages, manager_packages)
|
|
errors.extend(manager_sync_errors)
|
|
warnings.extend(manager_sync_warnings)
|
|
|
|
print(
|
|
f"active_loaders={len(active_loaders)} commented_loaders={len(commented_loaders)} "
|
|
f"specs={len(specs_by_family)} packages={len(spec_packages)} "
|
|
f"manager_packages={len(manager_packages)}"
|
|
)
|
|
for warning in warnings:
|
|
print(f"warning: {warning}")
|
|
if errors:
|
|
print("loader/spec sync failed:", file=sys.stderr)
|
|
for error in errors:
|
|
print(f" - {error}", file=sys.stderr)
|
|
print(
|
|
"\nFix: keep model_specs/*.json, model_manager_v2.py, registered loaders, "
|
|
"and published default GGUF packages aligned. Schema-level validation "
|
|
"belongs to the typed model-spec validator, and WebUI placement is checked separately.",
|
|
file=sys.stderr,
|
|
)
|
|
return 1
|
|
|
|
print("ok: runtime loaders, model_specs, and model_manager_v2 are in sync")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|