Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions .github/workflows/ci.yml
Original file line number Diff line number Diff line change
Expand Up @@ -41,6 +41,10 @@ jobs:
- name: Gate phase2-export.v1 asa-ableton handoff
run: python -m unittest tests.test_phase2_export.AsaAbletonHandoffContractTests
- run: python -m unittest discover -s tests
- name: Fundamentals gate (synthetic default corpus)
run: |
python scripts/build_synthetic_corpus.py --out-dir tests/fixtures/fundamentals_tracks --manifest tests/fixtures/fundamentals_eval_manifest.json --check --audio-only
python scripts/evaluate_fundamentals.py --fail-on-skip
- name: Assert no wrong-interpreter pyc files
run: |
BAD=$(
Expand Down
111 changes: 76 additions & 35 deletions apps/backend/fundamentals_evaluation.py
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,10 @@ class FundamentalsCheck:
name: str
passed: bool
message: str
# Known-gap checks still run and report their score but do not gate the
# summary — they keep a real baseline weakness visible in every report
# without holding the regression gate red.
informational: bool = False


@dataclass
Expand Down Expand Up @@ -67,6 +71,7 @@ def run_fundamentals_evaluation(
failed_subprocess = 0
passed_checks = 0
failed_checks = 0
informational_checks = 0
track_reports: list[dict[str, Any]] = []

for raw_track in tracks:
Expand All @@ -75,8 +80,9 @@ def run_fundamentals_evaluation(
result = _evaluate_track(raw_track, tracks_dir, runner)
if result.status == "evaluated":
evaluated += 1
passed_checks += sum(1 for check in result.checks if check.passed)
failed_checks += sum(1 for check in result.checks if not check.passed)
passed_checks += sum(1 for check in result.checks if check.passed and not check.informational)
failed_checks += sum(1 for check in result.checks if not check.passed and not check.informational)
informational_checks += sum(1 for check in result.checks if check.informational)
elif result.status == "skipped_audio_missing":
skipped += 1
else:
Expand All @@ -91,6 +97,7 @@ def run_fundamentals_evaluation(
"tracksAnalyzeFailed": failed_subprocess,
"checksPassed": passed_checks,
"checksFailed": failed_checks,
"checksInformational": informational_checks,
"failOnSkip": fail_on_skip,
"allPassed": failed_checks == 0 and (not fail_on_skip or skipped == 0),
}
Expand Down Expand Up @@ -119,6 +126,8 @@ def _evaluate_track(raw: dict[str, Any], tracks_dir: Path, runner: Runner) -> Fu
thresholds = raw.get("thresholds") if isinstance(raw.get("thresholds"), dict) else {}
flags = raw.get("analyzeFlags")
analyze_flags = [str(flag) for flag in flags] if isinstance(flags, list) else None
raw_gaps = raw.get("knownGaps")
known_gaps = {str(name) for name in raw_gaps} if isinstance(raw_gaps, list) else set()

if not audio_rel:
return FundamentalsTrackResult(
Expand Down Expand Up @@ -161,6 +170,9 @@ def _evaluate_track(raw: dict[str, Any], tracks_dir: Path, runner: Runner) -> Fu
)

checks = _evaluate_expected(payload, expected, thresholds)
for check in checks:
if check.name in known_gaps:
check.informational = True
return FundamentalsTrackResult(
track_id,
str(audio_path),
Expand All @@ -169,7 +181,7 @@ def _evaluate_track(raw: dict[str, Any], tracks_dir: Path, runner: Runner) -> Fu
"evaluated",
None,
checks,
all(check.passed for check in checks),
all(check.passed for check in checks if not check.informational),
)


Expand Down Expand Up @@ -334,7 +346,12 @@ def _track_result_to_report(result: FundamentalsTrackResult) -> dict[str, Any]:
"status": result.status,
"skipReason": result.skip_reason,
"checks": [
{"name": check.name, "passed": check.passed, "message": check.message}
{
"name": check.name,
"passed": check.passed,
"message": check.message,
"informational": check.informational,
}
for check in result.checks
],
"allPassed": result.all_passed,
Expand Down Expand Up @@ -370,36 +387,46 @@ def _normalize_key(value: str | None) -> str:
return re.sub(r"\s+", " ", normalized)


# Enharmonic-aware note-name → pitch-class map. KeyExtractor prefers sharp
# spellings ("C# Minor") while the Viterbi chord vocab and generated fixtures
# use flats ("Db"), so all key/chord comparisons fold to pitch classes.
_NOTE_TO_PC = {
"cb": 11, "c": 0, "c#": 1, "db": 1, "d": 2, "d#": 3, "eb": 3,
"e": 4, "e#": 5, "fb": 4, "f": 5, "f#": 6, "gb": 6, "g": 7,
"g#": 8, "ab": 8, "a": 9, "a#": 10, "bb": 10, "b": 11, "b#": 0,
}


def _parse_key_pc(value: str | None) -> tuple[int, str] | None:
parts = _normalize_key(value).split()
if not parts:
return None
pc = _NOTE_TO_PC.get(parts[0])
if pc is None:
return None
mode = parts[1] if len(parts) > 1 else "major"
return pc, mode


def _keys_match(actual: str | None, expected: str, *, allow_relative: bool) -> bool:
actual_norm = _normalize_key(actual)
expected_norm = _normalize_key(expected)
if actual_norm == expected_norm:
actual_pc = _parse_key_pc(actual)
expected_pc = _parse_key_pc(expected)
if actual_pc is None or expected_pc is None:
return _normalize_key(actual) == _normalize_key(expected)
if actual_pc == expected_pc:
return True
if not allow_relative:
return False
return _relative_key(actual_norm) == expected_norm or _relative_key(expected_norm) == actual_norm


def _relative_key(normalized: str) -> str:
minor_to_major = {
"a minor": "c major",
"e minor": "g major",
"b minor": "d major",
"f# minor": "a major",
"c# minor": "e major",
"g# minor": "b major",
"d# minor": "f# major",
"a# minor": "c# major",
"d minor": "f major",
"g minor": "bb major",
"c minor": "eb major",
"f minor": "ab major",
"bb minor": "db major",
"eb minor": "gb major",
"ab minor": "cb major",
}
major_to_minor = {major: minor for minor, major in minor_to_major.items()}
return minor_to_major.get(normalized) or major_to_minor.get(normalized) or ""
return _relative_of(actual_pc) == expected_pc or _relative_of(expected_pc) == actual_pc


def _relative_of(key: tuple[int, str]) -> tuple[int, str] | None:
pc, mode = key
if mode == "minor":
return (pc + 3) % 12, "major"
if mode == "major":
return (pc + 9) % 12, "minor"
return None


def _event_f1(actual: list[Any], expected: list[Any], *, tolerance_seconds: float) -> float:
Expand Down Expand Up @@ -449,21 +476,35 @@ def _chord_segment_accuracy(actual: list[Any], expected: list[Any]) -> float:
overlap = max(0.0, min(end, actual_end) - max(start, actual_start))
if (
overlap > 0.0
and _normalize_chord_label_for_compare(actual_label)
== _normalize_chord_label_for_compare(label)
and _normalize_chord_label(actual_label) == _normalize_chord_label(label)
):
matching_duration += overlap
return matching_duration / total_duration if total_duration > 0 else 0.0


def _normalize_chord_label_for_compare(label: str | None) -> str:
"""Normalize ASA triad labels for fixture comparisons without enharmonic folding."""
def _normalize_chord_label(label: str | None) -> str:
"""Fold a triad label to a canonical pitch-class form ("9m", "1", "N").

Enharmonic-aware (Db == C#) because the Viterbi vocab spells flats while
generated fixtures or hand labels may spell sharps. Named differently from
analyze_segments._normalize_chord_label_for_compare, which serves the
chordTimelineAgreement comparison and has its own rules.
"""
value = (label or "").strip()
value = re.sub(r"\s+", "", value)
value = value.replace(":maj", "").replace(":min", "m")
value = re.sub(r"(?i)major$", "", value)
value = re.sub(r"(?i)minor$", "m", value)
return value
if not value:
return ""
if value.upper() == "N":
return "N"
is_minor = value.endswith("m")
root = value[:-1] if is_minor else value
pc = _NOTE_TO_PC.get(root.lower())
if pc is None:
return value.lower()
return f"{pc}{'m' if is_minor else ''}"


def _note_f1(actual: list[Any], expected: list[Any]) -> float:
Expand Down
11 changes: 11 additions & 0 deletions apps/backend/fundamentals_quality.py
Original file line number Diff line number Diff line change
Expand Up @@ -102,6 +102,17 @@ def _tempo_quality(payload: dict[str, Any]) -> dict[str, Any]:
status = _status_from_confidence(confidence)
if agreement is False and payload.get("bpmDoubletime") is not True:
status = STATUS_AMBIGUOUS
elif (
status == STATUS_AMBIGUOUS
and agreement is True
and isinstance(confidence, (int, float))
and confidence >= 0.5
):
# Two independent estimators (RhythmExtractor2013 + Percival) agree
# within tolerance — the cross-check itself is the settling evidence
# even when the extractor's own confidence sits mid-range. Marking
# this "cross-check not strong enough" was factually wrong.
status = STATUS_AUTHORITATIVE
return _domain(
status=status,
plain=(
Expand Down
Loading
Loading