230 lines
10 KiB
Python
230 lines
10 KiB
Python
#!/usr/bin/env python3
|
|
"""Validate the tiny v0.1 utterance handoff fixtures.
|
|
|
|
This is a contract smoke, not a general schema runtime. It checks the current
|
|
training entrypoint and the fixed vector widths that AquaSynth is allowed to
|
|
depend on for `weksa.utterance_embedding_handoff.v0.1`.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
DEFAULT_BATCH = ROOT / "examples" / "speech-training" / "tiny-panphon-v0.1" / "batch.json"
|
|
|
|
PACKET_SCHEMA_VERSION = "weksa.utterance_embedding_handoff.v0.1"
|
|
BATCH_SCHEMA_VERSION = "weksa.utterance_handoff_batch.v0.1"
|
|
|
|
SPEECH_TEXT_DIMENSIONALITY = 1024
|
|
PANPHON_FEATURE_DIMENSIONALITY = 22
|
|
PROSODY_DIMENSIONALITY = 32
|
|
CHARACTER_STATE_DIMENSIONALITY = 64
|
|
PHONETIC_REALIZATION_DIMENSIONALITY = 256
|
|
UTTERANCE_EMBEDDING_DIMENSIONALITY = 64
|
|
|
|
|
|
class ValidationError(ValueError):
|
|
"""Raised when a fixture violates the v0.1 handoff contract."""
|
|
|
|
|
|
def load_json(path: Path) -> Any:
|
|
try:
|
|
return json.loads(path.read_text(encoding="utf-8"))
|
|
except FileNotFoundError as exc:
|
|
raise ValidationError(f"Missing JSON file: {path}") from exc
|
|
except json.JSONDecodeError as exc:
|
|
raise ValidationError(f"Invalid JSON in {path}: {exc}") from exc
|
|
|
|
|
|
def require(condition: bool, message: str) -> None:
|
|
if not condition:
|
|
raise ValidationError(message)
|
|
|
|
|
|
def require_object(value: Any, path: str) -> dict[str, Any]:
|
|
require(isinstance(value, dict), f"{path} must be an object")
|
|
return value
|
|
|
|
|
|
def require_list(value: Any, path: str) -> list[Any]:
|
|
require(isinstance(value, list), f"{path} must be a list")
|
|
return value
|
|
|
|
|
|
def require_dimensional_vector(
|
|
node: dict[str, Any],
|
|
*,
|
|
path: str,
|
|
dimensionality: int,
|
|
allow_ref: bool,
|
|
packet_dir: Path,
|
|
require_zero: bool = False,
|
|
) -> None:
|
|
require(node.get("dimensionality") == dimensionality, f"{path}.dimensionality must be {dimensionality}")
|
|
|
|
values = node.get("values")
|
|
values_ref = node.get("values_ref")
|
|
|
|
if values is not None:
|
|
vector = require_list(values, f"{path}.values")
|
|
require(len(vector) == dimensionality, f"{path}.values must contain {dimensionality} values")
|
|
require(all(isinstance(value, (int, float)) for value in vector), f"{path}.values must be numeric")
|
|
if require_zero:
|
|
require(all(value == 0.0 for value in vector), f"{path}.values must stay zeroed for pure IPA fixtures")
|
|
return
|
|
|
|
if allow_ref and values_ref:
|
|
ref_path = (packet_dir / values_ref).resolve()
|
|
ref = require_object(load_json(ref_path), str(ref_path))
|
|
require(ref.get("dimensionality") == dimensionality, f"{ref_path}.dimensionality must be {dimensionality}")
|
|
ref_values = require_list(ref.get("values"), f"{ref_path}.values")
|
|
require(len(ref_values) == dimensionality, f"{ref_path}.values must contain {dimensionality} values")
|
|
require(all(isinstance(value, (int, float)) for value in ref_values), f"{ref_path}.values must be numeric")
|
|
if require_zero:
|
|
require(all(value == 0.0 for value in ref_values), f"{ref_path}.values must stay zeroed for pure IPA fixtures")
|
|
return
|
|
|
|
raise ValidationError(f"{path} must contain values or values_ref")
|
|
|
|
|
|
def validate_pending_aquasynth_vector(node: dict[str, Any], *, path: str, dimensionality: int) -> None:
|
|
require(node.get("owner") == "AquaSynth", f"{path}.owner must be AquaSynth")
|
|
require(node.get("dimensionality") == dimensionality, f"{path}.dimensionality must be {dimensionality}")
|
|
values_ref = node.get("values_ref")
|
|
status = node.get("status")
|
|
require(values_ref is None, f"{path}.values_ref must stay null until AquaSynth supplies learned output")
|
|
require(status == "pending_aquasynth_training_or_encoding", f"{path}.status must mark pending AquaSynth output")
|
|
|
|
|
|
def validate_panphon_sequence(node: dict[str, Any], *, path: str) -> None:
|
|
require(node.get("ipa"), f"{path}.ipa must not be empty")
|
|
feature_order = require_list(node.get("feature_order"), f"{path}.feature_order")
|
|
require(
|
|
len(feature_order) == PANPHON_FEATURE_DIMENSIONALITY,
|
|
f"{path}.feature_order must contain {PANPHON_FEATURE_DIMENSIONALITY} features",
|
|
)
|
|
|
|
frames = require_list(node.get("frames"), f"{path}.frames")
|
|
require(frames, f"{path}.frames must not be empty")
|
|
|
|
for index, raw_frame in enumerate(frames):
|
|
frame = require_object(raw_frame, f"{path}.frames[{index}]")
|
|
require(frame.get("index") == index, f"{path}.frames[{index}].index must match frame position")
|
|
require(frame.get("phone"), f"{path}.frames[{index}].phone must not be empty")
|
|
|
|
features = require_list(frame.get("features"), f"{path}.frames[{index}].features")
|
|
require(
|
|
len(features) == PANPHON_FEATURE_DIMENSIONALITY,
|
|
f"{path}.frames[{index}].features must contain {PANPHON_FEATURE_DIMENSIONALITY} values",
|
|
)
|
|
require(
|
|
all(value in (-1.0, 0.0, 1.0) for value in features),
|
|
f"{path}.frames[{index}].features must use ternary PanPhon values",
|
|
)
|
|
|
|
markers = require_object(frame.get("markers"), f"{path}.frames[{index}].markers")
|
|
for marker in ["stress", "length", "tone", "boundary", "duration_ms"]:
|
|
require(isinstance(markers.get(marker), (int, float)), f"{path}.frames[{index}].markers.{marker} must be numeric")
|
|
|
|
require(
|
|
require_object(frames[-1], f"{path}.frames[-1]").get("markers", {}).get("boundary") == 1.0,
|
|
f"{path}.frames[-1].markers.boundary must close the sequence",
|
|
)
|
|
|
|
|
|
def validate_packet(packet_path: Path, expected_batch_id: str) -> None:
|
|
packet = require_object(load_json(packet_path), str(packet_path))
|
|
packet_dir = packet_path.parent
|
|
|
|
require(packet.get("schema_version") == PACKET_SCHEMA_VERSION, f"{packet_path}.schema_version must be {PACKET_SCHEMA_VERSION}")
|
|
require(str(packet.get("packet_id", "")).startswith(f"{expected_batch_id}/"), f"{packet_path}.packet_id must belong to {expected_batch_id}")
|
|
require(packet.get("source_output_ref"), f"{packet_path}.source_output_ref must not be empty")
|
|
require(packet.get("speaker_agent_id"), f"{packet_path}.speaker_agent_id must not be empty")
|
|
|
|
text = require_object(packet.get("text"), f"{packet_path}.text")
|
|
require(text.get("spoken_text") == text.get("normalized_text"), f"{packet_path}.text normalized_text must match pure IPA spoken_text")
|
|
|
|
inputs = require_object(packet.get("inputs"), f"{packet_path}.inputs")
|
|
speech_text_embedding = require_object(inputs.get("speech_text_embedding"), f"{packet_path}.inputs.speech_text_embedding")
|
|
require(speech_text_embedding.get("model_id") == "bge-m3:latest", f"{packet_path}.inputs.speech_text_embedding.model_id must be bge-m3:latest")
|
|
require_dimensional_vector(
|
|
speech_text_embedding,
|
|
path=f"{packet_path}.inputs.speech_text_embedding",
|
|
dimensionality=SPEECH_TEXT_DIMENSIONALITY,
|
|
allow_ref=True,
|
|
packet_dir=packet_dir,
|
|
require_zero=True,
|
|
)
|
|
|
|
validate_panphon_sequence(require_object(inputs.get("panphon_sequence"), f"{packet_path}.inputs.panphon_sequence"), path=f"{packet_path}.inputs.panphon_sequence")
|
|
|
|
prosody = require_object(inputs.get("prosody_emphasis_hints"), f"{packet_path}.inputs.prosody_emphasis_hints")
|
|
require(len(require_list(prosody.get("labels"), f"{packet_path}.inputs.prosody_emphasis_hints.labels")) == PROSODY_DIMENSIONALITY, f"{packet_path}.inputs.prosody_emphasis_hints.labels must contain {PROSODY_DIMENSIONALITY} labels")
|
|
require_dimensional_vector(
|
|
prosody,
|
|
path=f"{packet_path}.inputs.prosody_emphasis_hints",
|
|
dimensionality=PROSODY_DIMENSIONALITY,
|
|
allow_ref=False,
|
|
packet_dir=packet_dir,
|
|
)
|
|
|
|
character_state = require_object(inputs.get("character_state_vector"), f"{packet_path}.inputs.character_state_vector")
|
|
require_dimensional_vector(
|
|
character_state,
|
|
path=f"{packet_path}.inputs.character_state_vector",
|
|
dimensionality=CHARACTER_STATE_DIMENSIONALITY,
|
|
allow_ref=True,
|
|
packet_dir=packet_dir,
|
|
)
|
|
|
|
validate_pending_aquasynth_vector(
|
|
require_object(packet.get("phonetic_realization_embedding"), f"{packet_path}.phonetic_realization_embedding"),
|
|
path=f"{packet_path}.phonetic_realization_embedding",
|
|
dimensionality=PHONETIC_REALIZATION_DIMENSIONALITY,
|
|
)
|
|
validate_pending_aquasynth_vector(
|
|
require_object(packet.get("utterance_embedding"), f"{packet_path}.utterance_embedding"),
|
|
path=f"{packet_path}.utterance_embedding",
|
|
dimensionality=UTTERANCE_EMBEDDING_DIMENSIONALITY,
|
|
)
|
|
|
|
require_object(packet.get("audit_projection"), f"{packet_path}.audit_projection")
|
|
require_object(packet.get("training_refs"), f"{packet_path}.training_refs")
|
|
require_object(packet.get("trace"), f"{packet_path}.trace")
|
|
|
|
|
|
def validate_batch(batch_path: Path) -> list[Path]:
|
|
batch = require_object(load_json(batch_path), str(batch_path))
|
|
require(batch.get("schema_version") == BATCH_SCHEMA_VERSION, f"{batch_path}.schema_version must be {BATCH_SCHEMA_VERSION}")
|
|
require(batch.get("packet_schema_version") == PACKET_SCHEMA_VERSION, f"{batch_path}.packet_schema_version must be {PACKET_SCHEMA_VERSION}")
|
|
batch_id = batch.get("batch_id")
|
|
require(isinstance(batch_id, str) and batch_id, f"{batch_path}.batch_id must not be empty")
|
|
|
|
packet_refs = require_list(batch.get("packet_refs"), f"{batch_path}.packet_refs")
|
|
require(packet_refs, f"{batch_path}.packet_refs must not be empty")
|
|
|
|
packet_paths = [(batch_path.parent / ref).resolve() for ref in packet_refs]
|
|
for packet_path in packet_paths:
|
|
validate_packet(packet_path, batch_id)
|
|
return packet_paths
|
|
|
|
|
|
def main(argv: list[str]) -> int:
|
|
batch_path = Path(argv[1]).resolve() if len(argv) > 1 else DEFAULT_BATCH
|
|
try:
|
|
packet_paths = validate_batch(batch_path)
|
|
except ValidationError as exc:
|
|
print(f"validation failed: {exc}", file=sys.stderr)
|
|
return 1
|
|
|
|
print(f"validated {len(packet_paths)} packet(s) from {batch_path}")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main(sys.argv))
|