diff --git a/demo/reporting_example_results.jsonl b/demo/reporting_example_results.jsonl new file mode 100644 index 0000000..969b398 --- /dev/null +++ b/demo/reporting_example_results.jsonl @@ -0,0 +1,6 @@ +{"key": "utt_001", "model": "baseline", "pesq": 2.1, "stoi": 0.72, "wer": 0.18, "spk_similarity": 0.81, "mcd": 6.4} +{"key": "utt_002", "model": "baseline", "pesq": 2.4, "stoi": 0.75, "wer": 0.15, "spk_similarity": 0.84, "mcd": 5.9} +{"key": "utt_003", "model": "baseline", "pesq": 1.9, "stoi": 0.69, "wer": 0.23, "spk_similarity": 0.79, "mcd": 7.1} +{"key": "utt_004", "model": "candidate", "pesq": 3.2, "stoi": 0.86, "wer": 0.09, "spk_similarity": 0.88, "mcd": 4.8} +{"key": "utt_005", "model": "candidate", "pesq": 3.4, "stoi": 0.88, "wer": 0.07, "spk_similarity": 0.91, "mcd": 4.5} +{"key": "utt_006", "model": "candidate", "pesq": 2.8, "stoi": 0.82, "wer": 0.12, "spk_similarity": 0.86, "mcd": 5.2} diff --git a/demo/run_reporting_example.sh b/demo/run_reporting_example.sh new file mode 100755 index 0000000..18a2129 --- /dev/null +++ b/demo/run_reporting_example.sh @@ -0,0 +1,17 @@ +#!/usr/bin/env bash +set -euo pipefail + +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +RESULTS="${SCRIPT_DIR}/reporting_example_results.jsonl" +OUT_DIR="${TMPDIR:-/tmp}/versa-reporting-example" + +mkdir -p "${OUT_DIR}" + +versa-visualize "${RESULTS}" \ + --out "${OUT_DIR}/report.html" \ + --csv "${OUT_DIR}/report.csv" \ + --markdown "${OUT_DIR}/report.md" \ + --group-by model + +echo "Wrote example reports to ${OUT_DIR}" +echo "Open ${OUT_DIR}/report.html to inspect the visualization report." diff --git a/docs/visualization.md b/docs/visualization.md index e40e599..217f907 100644 --- a/docs/visualization.md +++ b/docs/visualization.md @@ -1,4 +1,45 @@ ## Interactive Visualization of Versa Results + +### Packaged Report Command + +VERSA can generate a first-class report directly from a scoring JSONL file or a +directory of result files: + +``` +versa-visualize results.jsonl --out report.html +``` + +The HTML report includes a radar overview, category sunburst, summary tables, +mean/std, 95% confidence intervals, failure counts, best/worst examples, and +outlier examples. CSV and Markdown exports are also available: + +``` +versa-visualize results.jsonl --out report.html --csv report.csv --markdown report.md +``` + +For model comparisons, group records by a field and VERSA will add per-metric +rankings: + +``` +versa-visualize results.jsonl --out report.html --group-by model +``` + +The aggregation command can also write summary reports: + +``` +versa-aggregate results.jsonl --out metrics_report.csv +versa-aggregate results.jsonl --out metrics_report.md --format md +``` + +For a quick smoke test after installation, run the bundled toy example: + +``` +demo/run_reporting_example.sh +``` + +It reads ``demo/reporting_example_results.jsonl`` and writes an HTML report plus +CSV/Markdown summaries to ``${TMPDIR:-/tmp}/versa-reporting-example``. + ### Steps * Additional Package Dependency Installation ``` diff --git a/pyproject.toml b/pyproject.toml index 5c5091c..8a347b2 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -124,6 +124,8 @@ Homepage = "https://github.com/wavlab-speech/versa.git" [project.scripts] versa-score = "versa.bin.scorer:main" +versa-aggregate = "versa.bin.aggregate_results:main" +versa-visualize = "versa.bin.visualize:main" [tool.setuptools.packages.find] include = ["versa*"] diff --git a/test/test_reporting.py b/test/test_reporting.py new file mode 100644 index 0000000..e465518 --- /dev/null +++ b/test/test_reporting.py @@ -0,0 +1,74 @@ +import csv + +from versa.reporting import ( + analyze_records, + metric_category, + read_result_records, + write_csv_report, + write_html_report, + write_markdown_report, +) + + +def test_analyze_records_computes_ci_failures_rankings_and_outliers(): + records = [ + {"key": "a", "model": "m1", "pesq": 2.0, "wer": 0.2}, + {"key": "b", "model": "m1", "pesq": 2.5, "wer": 0.1}, + {"key": "c", "model": "m2", "pesq": 4.0, "wer": 0.4}, + {"key": "d", "model": "m2", "pesq": "bad", "wer": 2.0}, + {"key": "e", "model": "m2", "wer": 0.3}, + ] + + analysis = analyze_records(records, group_by="model") + summaries = {summary.name: summary for summary in analysis["metrics"]} + + assert analysis["record_count"] == 5 + assert summaries["pesq"].count == 3 + assert summaries["pesq"].missing == 1 + assert summaries["pesq"].invalid == 1 + assert summaries["pesq"].ci95_high > summaries["pesq"].ci95_low + assert summaries["pesq"].best_key == "c" + assert summaries["wer"].best_key == "b" + assert summaries["wer"].worst_key == "d" + assert analysis["groups"]["rankings"]["pesq"][0]["group"] == "m2" + assert analysis["groups"]["rankings"]["wer"][0]["group"] == "m1" + + +def test_read_result_records_accepts_jsonl_and_python_literals(tmp_path): + result_file = tmp_path / "results.txt" + result_file.write_text( + '{"key": "a", "pesq": 2.0}\n' "{'key': 'b', 'sir': inf}\n", + encoding="utf-8", + ) + + records = read_result_records(str(result_file)) + + assert [record["key"] for record in records] == ["a", "b"] + assert records[0]["_source_file"] == "results.txt" + + +def test_report_exports(tmp_path): + records = [ + {"key": "a", "pesq": 2.0, "wer": 0.2}, + {"key": "b", "pesq": 3.0, "wer": 0.1}, + ] + analysis = analyze_records(records) + csv_path = tmp_path / "report.csv" + md_path = tmp_path / "report.md" + html_path = tmp_path / "report.html" + + write_csv_report(analysis, str(csv_path)) + write_markdown_report(analysis, str(md_path)) + write_html_report(analysis, str(html_path)) + + with csv_path.open(newline="", encoding="utf-8") as handle: + rows = list(csv.DictReader(handle)) + assert {row["metric"] for row in rows} == {"pesq", "wer"} + assert "VERSA Results Report" in md_path.read_text(encoding="utf-8") + assert "Radar Overview" in html_path.read_text(encoding="utf-8") + assert "Category Sunburst" in html_path.read_text(encoding="utf-8") + + +def test_metric_category_strips_model_prefixes(): + assert metric_category("arecho_pesq") == "speech_enhancement" + assert metric_category("custom_wer") == "asr_wer_cer" diff --git a/versa/__init__.py b/versa/__init__.py index b9c466e..eb69add 100644 --- a/versa/__init__.py +++ b/versa/__init__.py @@ -20,6 +20,13 @@ flag in sys.argv for flag in ("--list-metrics", "--describe-metric", "--recommend-config") ) +_SKIP_OPTIONAL_METRIC_IMPORTS = _SKIP_OPTIONAL_METRIC_IMPORTS or any( + Path(arg).name + in {"versa-visualize", "versa-aggregate", "visualize.py", "aggregate_results.py"} + or arg.endswith("versa.bin.visualize") + or arg.endswith("versa.bin.aggregate_results") + for arg in sys.argv +) def _optional_metric_import(module_name, names, install_hint=None): diff --git a/versa/bin/aggregate_results.py b/versa/bin/aggregate_results.py index cbc7063..a8d6bd0 100644 --- a/versa/bin/aggregate_results.py +++ b/versa/bin/aggregate_results.py @@ -3,39 +3,70 @@ # Copyright 2024 Jiatong Shi # Apache 2.0 (http://www.apache.org/licenses/LICENSE-2.0) -"""Aggregate results.""" +"""Aggregate and report VERSA results.""" import argparse import json import logging +import os from tqdm import tqdm +from versa.reporting import ( + analyze_records, + read_result_records, + write_csv_report, + write_html_report, + write_markdown_report, +) + def get_parser() -> argparse.Namespace: """Get parser of aggregate results.""" parser = argparse.ArgumentParser( - description="Aggregate results.", + description="Aggregate chunked results or generate polished result reports.", formatter_class=argparse.ArgumentDefaultsHelpFormatter, ) + parser.add_argument( + "input", + nargs="?", + help="Input JSONL result file or directory. If omitted, --logdir/--scoredir/--nj mode is used.", + ) parser.add_argument( "--logdir", type=str, - required=True, + default=None, help="Input log directory.", ) parser.add_argument( "--scoredir", type=str, - required=True, + default=None, help="Output scoring directory.", ) parser.add_argument( "--nj", type=int, - required=True, + default=None, help="Number of sub jobs", ) + parser.add_argument( + "--out", + type=str, + default=None, + help="Output report path for input mode.", + ) + parser.add_argument( + "--format", + choices=["auto", "csv", "md", "html"], + default="auto", + help="Report format for input mode.", + ) + parser.add_argument( + "--group-by", + default=None, + help="Optional record field used for per-metric ranking.", + ) return parser @@ -61,11 +92,56 @@ def aggregate_results(logdir: str, scoredir: str, nj: int) -> None: logging.info("Done.") +def generate_report( + input_path: str, + output_path: str, + report_format: str = "auto", + group_by: str = None, +) -> None: + """Generate a CSV, Markdown, or HTML report from result records.""" + records = read_result_records(input_path) + analysis = analyze_records(records, group_by=group_by) + + if report_format == "auto": + suffix = os.path.splitext(output_path)[1].lower() + report_format = { + ".csv": "csv", + ".md": "md", + ".markdown": "md", + ".html": "html", + ".htm": "html", + }.get(suffix, "csv") + + os.makedirs(os.path.dirname(os.path.abspath(output_path)), exist_ok=True) + if report_format == "csv": + write_csv_report(analysis, output_path) + elif report_format == "md": + write_markdown_report(analysis, output_path) + else: + write_html_report(analysis, output_path) + + logging.info( + "Wrote %s report for %s utterances and %s metrics to %s", + report_format, + analysis["record_count"], + analysis["metric_count"], + output_path, + ) + + def main() -> None: """Run main function.""" parser = get_parser() args = parser.parse_args() logging.basicConfig(level=logging.INFO) + if args.input: + output_path = args.out or "metrics_report.csv" + generate_report(args.input, output_path, args.format, args.group_by) + return + + if args.logdir is None or args.scoredir is None or args.nj is None: + parser.error("either provide input, or provide --logdir, --scoredir, and --nj") + aggregate_results(args.logdir, args.scoredir, args.nj) diff --git a/versa/bin/visualize.py b/versa/bin/visualize.py new file mode 100644 index 0000000..df952bf --- /dev/null +++ b/versa/bin/visualize.py @@ -0,0 +1,111 @@ +#!/usr/bin/env python3 + +"""Generate VERSA result reports and visualizations.""" + +import argparse +import os +from pathlib import Path + +from versa.reporting import ( + analyze_records, + read_result_records, + write_csv_report, + write_html_report, + write_markdown_report, +) + + +def get_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + description="Create summary tables and visual reports from VERSA results.", + formatter_class=argparse.ArgumentDefaultsHelpFormatter, + ) + parser.add_argument( + "input", + help="Input JSONL result file or directory of result files.", + ) + parser.add_argument( + "--out", + default="report.html", + help="Output report path. Format is inferred from the extension.", + ) + parser.add_argument( + "--format", + choices=["auto", "html", "csv", "md"], + default="auto", + help="Report format.", + ) + parser.add_argument( + "--group-by", + default=None, + help="Optional record field used for per-metric ranking, e.g. _source_file.", + ) + parser.add_argument( + "--csv", + default=None, + help="Optional extra CSV summary export path.", + ) + parser.add_argument( + "--markdown", + default=None, + help="Optional extra Markdown summary export path.", + ) + parser.add_argument( + "--outlier-limit", + type=int, + default=3, + help="Maximum outlier examples to keep per metric.", + ) + return parser + + +def main() -> None: + parser = get_parser() + args = parser.parse_args() + + records = read_result_records(args.input) + analysis = analyze_records( + records, + group_by=args.group_by, + outlier_limit=args.outlier_limit, + ) + + output_path = Path(args.out) + output_path.parent.mkdir(parents=True, exist_ok=True) + report_format = args.format + if report_format == "auto": + suffix = output_path.suffix.lower() + report_format = { + ".html": "html", + ".htm": "html", + ".csv": "csv", + ".md": "md", + ".markdown": "md", + }.get(suffix, "html") + + if report_format == "html": + write_html_report(analysis, os.fspath(output_path)) + elif report_format == "csv": + write_csv_report(analysis, os.fspath(output_path)) + else: + write_markdown_report(analysis, os.fspath(output_path)) + + if args.csv: + Path(args.csv).parent.mkdir(parents=True, exist_ok=True) + write_csv_report(analysis, args.csv) + if args.markdown: + Path(args.markdown).parent.mkdir(parents=True, exist_ok=True) + write_markdown_report(analysis, args.markdown) + + print( + "Wrote {format} report for {records} utterances and {metrics} metrics to {path}".format( + format=report_format, + records=analysis["record_count"], + metrics=analysis["metric_count"], + path=output_path, + ) + ) + + +if __name__ == "__main__": + main() diff --git a/versa/reporting.py b/versa/reporting.py new file mode 100644 index 0000000..eca49c3 --- /dev/null +++ b/versa/reporting.py @@ -0,0 +1,781 @@ +"""Reporting helpers for VERSA scoring results.""" + +import ast +import csv +import html +import json +import math +import os +from collections import defaultdict +from dataclasses import dataclass +from pathlib import Path +from typing import Any, Dict, List, Optional, Sequence, Tuple + +IGNORED_FIELDS = {"key", "_source_file"} + + +METRIC_CATEGORIES = { + "audio_quality": [ + "dns_overall", + "dns_p808", + "nisqa", + "utmos", + "plcmos", + "singmos", + "sheet_ssqa", + "utmosv2", + "scoreq_nr", + "scoreq_ref", + "noresqa", + "torch_squim_mos", + "warpq", + "dnsmos_pro_bvcc", + "dnsmos_pro_nisqa", + "dnsmos_pro_vcc2018", + ], + "speech_enhancement": [ + "torch_squim_pesq", + "torch_squim_stoi", + "torch_squim_si_sdr", + "se_si_snr", + "se_ci_sdr", + "se_sar", + "se_sdr", + "pesq", + "stoi", + "sir", + "sar", + "sdr", + "ci-sdr", + "ci_sdr", + "si-snr", + "si_snr", + "visqol", + ], + "psychoacoustic": [ + "pysepm_fwsegsnr", + "pysepm_wss", + "pysepm_cd", + "pysepm_c_sig", + "pysepm_c_bak", + "pysepm_c_ovl", + "pysepm_csii_high", + "pysepm_csii_mid", + "pysepm_csii_low", + "pysepm_ncm", + "pysepm_llr", + "pam", + "pam_score", + "srmr", + ], + "asr_wer_cer": [ + "espnet_wer", + "espnet_cer", + "owsm_wer", + "owsm_cer", + "whisper_wer", + "whisper_cer", + "wer", + "cer", + "asr_match_error_rate", + ], + "semantic": [ + "speech_bert", + "speech_bleu", + "speech_token_distance", + "clap_score", + ], + "similarity": ["emotion_similarity", "spk_similarity", "singer_similarity"], + "pitch_f0": ["f0_corr", "f0corr", "f0_rmse", "f0rmse", "mcd"], + "audio_features": ["speaking_rate", "log_wmse"], + "aesthetics": [ + "audiobox_aesthetics_CE", + "audiobox_aesthetics_CU", + "audiobox_aesthetics_PC", + "audiobox_aesthetics_PQ", + ], + "security": ["asvspoof_score", "nomad"], +} + + +@dataclass +class MetricSummary: + name: str + category: str + count: int + missing: int + invalid: int + mean: float + median: float + std: float + stderr: float + ci95_low: float + ci95_high: float + minimum: float + maximum: float + higher_is_better: Optional[bool] + best_key: str + best_value: float + worst_key: str + worst_value: float + outliers: List[Tuple[str, float, float]] + + +def read_result_records(input_path: str) -> List[Dict[str, Any]]: + """Read VERSA result records from a file or directory.""" + paths = _collect_input_paths(input_path) + records: List[Dict[str, Any]] = [] + for path in paths: + with open(path, "r", encoding="utf-8") as handle: + for line_number, raw_line in enumerate(handle, start=1): + line = raw_line.strip() + if not line: + continue + try: + record = json.loads(line) + except json.JSONDecodeError: + try: + record = ast.literal_eval(line) + except (SyntaxError, ValueError): + try: + record = _literal_eval_with_special_floats(line) + except (SyntaxError, ValueError) as literal_exc: + raise ValueError( + f"Could not parse {path}:{line_number} as JSON or Python literal" + ) from literal_exc + if not isinstance(record, dict): + raise ValueError(f"Expected object in {path}:{line_number}") + record = dict(record) + record.setdefault("_source_file", path.name) + records.append(record) + return records + + +def analyze_records( + records: Sequence[Dict[str, Any]], + *, + group_by: Optional[str] = None, + outlier_limit: int = 3, +) -> Dict[str, Any]: + """Compute report-ready summaries from result records.""" + if not records: + raise ValueError("No result records were found") + + metrics = discover_numeric_metrics(records) + metric_summaries = [ + summarize_metric(metric, records, outlier_limit=outlier_limit) + for metric in sorted(metrics) + ] + categories: Dict[str, List[MetricSummary]] = defaultdict(list) + for summary in metric_summaries: + categories[summary.category].append(summary) + + groups = {} + if group_by: + groups = summarize_groups(records, metrics, group_by) + + return { + "records": records, + "metrics": metric_summaries, + "categories": dict(sorted(categories.items())), + "groups": groups, + "group_by": group_by, + "record_count": len(records), + "metric_count": len(metric_summaries), + } + + +def discover_numeric_metrics(records: Sequence[Dict[str, Any]]) -> List[str]: + metrics = set() + for record in records: + for key, value in record.items(): + if key in IGNORED_FIELDS or key.startswith("_") or "text" in key.lower(): + continue + if _to_float(value) is not None: + metrics.add(key) + return sorted(metrics) + + +def summarize_metric( + metric: str, records: Sequence[Dict[str, Any]], *, outlier_limit: int = 3 +) -> MetricSummary: + values: List[Tuple[str, float]] = [] + missing = 0 + invalid = 0 + for index, record in enumerate(records, start=1): + key = str(record.get("key") or f"utt_{index}") + if metric not in record: + missing += 1 + continue + value = _to_float(record[metric]) + if value is None or not math.isfinite(value): + invalid += 1 + continue + values.append((key, value)) + + numeric = [value for _, value in values] + count = len(numeric) + mean = sum(numeric) / count if count else 0.0 + sorted_values = sorted(numeric) + median = _median(sorted_values) + std = _sample_std(numeric, mean) + stderr = std / math.sqrt(count) if count else 0.0 + ci_delta = 1.96 * stderr + minimum = min(numeric) if numeric else 0.0 + maximum = max(numeric) if numeric else 0.0 + higher_is_better = metric_direction(metric) + + if values and higher_is_better is False: + best_key, best_value = min(values, key=lambda item: item[1]) + worst_key, worst_value = max(values, key=lambda item: item[1]) + elif values: + best_key, best_value = max(values, key=lambda item: item[1]) + worst_key, worst_value = min(values, key=lambda item: item[1]) + else: + best_key, best_value, worst_key, worst_value = "", 0.0, "", 0.0 + + outliers = [] + if count > 1 and std > 0: + scored = [ + (key, value, (value - mean) / std) + for key, value in values + if abs((value - mean) / std) >= 2.0 + ] + outliers = sorted(scored, key=lambda item: abs(item[2]), reverse=True)[ + :outlier_limit + ] + + return MetricSummary( + name=metric, + category=metric_category(metric), + count=count, + missing=missing, + invalid=invalid, + mean=mean, + median=median, + std=std, + stderr=stderr, + ci95_low=mean - ci_delta, + ci95_high=mean + ci_delta, + minimum=minimum, + maximum=maximum, + higher_is_better=higher_is_better, + best_key=best_key, + best_value=best_value, + worst_key=worst_key, + worst_value=worst_value, + outliers=outliers, + ) + + +def summarize_groups( + records: Sequence[Dict[str, Any]], metrics: Sequence[str], group_by: str +) -> Dict[str, Any]: + grouped: Dict[str, List[Dict[str, Any]]] = defaultdict(list) + for record in records: + grouped[str(record.get(group_by, "unknown"))].append(record) + + metric_rankings = {} + for metric in metrics: + direction = metric_direction(metric) + rows = [] + for group, group_records in grouped.items(): + summary = summarize_metric(metric, group_records, outlier_limit=0) + if summary.count: + rows.append( + { + "group": group, + "mean": summary.mean, + "std": summary.std, + "count": summary.count, + } + ) + reverse = direction is not False + metric_rankings[metric] = sorted( + rows, key=lambda row: row["mean"], reverse=reverse + ) + + return { + "sizes": {key: len(value) for key, value in grouped.items()}, + "rankings": metric_rankings, + } + + +def write_csv_report(analysis: Dict[str, Any], output_path: str) -> None: + fields = [ + "metric", + "category", + "count", + "missing", + "invalid", + "mean", + "median", + "std", + "stderr", + "ci95_low", + "ci95_high", + "min", + "max", + "higher_is_better", + "best_key", + "best_value", + "worst_key", + "worst_value", + "outliers", + ] + with open(output_path, "w", newline="", encoding="utf-8") as handle: + writer = csv.DictWriter(handle, fieldnames=fields) + writer.writeheader() + for summary in analysis["metrics"]: + writer.writerow(_summary_row(summary)) + + +def write_markdown_report(analysis: Dict[str, Any], output_path: str) -> None: + lines = [ + "# VERSA Results Report", + "", + f"- Utterances: {analysis['record_count']}", + f"- Metrics: {analysis['metric_count']}", + "", + "## Summary", + "", + "| Metric | Category | Count | Mean | Std | 95% CI | Missing | Invalid | Best | Worst |", + "| --- | --- | ---: | ---: | ---: | --- | ---: | ---: | --- | --- |", + ] + for summary in analysis["metrics"]: + lines.append( + "| {metric} | {category} | {count} | {mean} | {std} | {ci} | {missing} | {invalid} | {best} | {worst} |".format( + metric=summary.name, + category=summary.category, + count=summary.count, + mean=_fmt(summary.mean), + std=_fmt(summary.std), + ci=f"{_fmt(summary.ci95_low)} to {_fmt(summary.ci95_high)}", + missing=summary.missing, + invalid=summary.invalid, + best=f"{summary.best_key} ({_fmt(summary.best_value)})", + worst=f"{summary.worst_key} ({_fmt(summary.worst_value)})", + ) + ) + lines.extend(["", "## Outlier Examples", ""]) + any_outliers = False + for summary in analysis["metrics"]: + if not summary.outliers: + continue + any_outliers = True + lines.append(f"### {summary.name}") + for key, value, z_score in summary.outliers: + lines.append(f"- {key}: {_fmt(value)} (z={_fmt(z_score)})") + lines.append("") + if not any_outliers: + lines.append("No z-score outliers >= 2.0 were detected.") + with open(output_path, "w", encoding="utf-8") as handle: + handle.write("\n".join(lines) + "\n") + + +def write_html_report(analysis: Dict[str, Any], output_path: str) -> None: + category_rows = [] + for category, summaries in analysis["categories"].items(): + expected_values = analysis["record_count"] * len(summaries) + observed_values = sum(summary.count for summary in summaries) + missing_values = sum(summary.missing for summary in summaries) + invalid_values = sum(summary.invalid for summary in summaries) + category_rows.append( + { + "category": category, + "metrics": len(summaries), + "observed": observed_values, + "missing": missing_values, + "invalid": invalid_values, + "coverage": ( + observed_values / expected_values if expected_values else 0.0 + ), + } + ) + + top_metrics = sorted( + analysis["metrics"], key=lambda item: item.count, reverse=True + )[:12] + radar_svg = _radar_svg(top_metrics) + sunburst_svg = _sunburst_svg(category_rows) + rows_html = "\n".join(_metric_html_row(summary) for summary in analysis["metrics"]) + category_html = "\n".join( + f"{html.escape(row['category'])}{row['metrics']}{row['observed']}{row['missing']}{row['invalid']}{_fmt(row['coverage'] * 100)}%" + for row in category_rows + ) + outlier_html = _outlier_html(analysis["metrics"]) + ranking_html = _ranking_html(analysis) + + document = f""" + + + + +VERSA Results Report + + + +
+

VERSA Results Report

+
Generated from scoring results with summary statistics, confidence intervals, rankings, failures, and outlier examples.
+
+
+
+
{analysis['record_count']}utterances
+
{analysis['metric_count']}numeric metrics
+
{len(analysis['categories'])}metric categories
+
{sum(s.missing + s.invalid for s in analysis['metrics'])}missing or invalid metric values
+
+
+

Radar Overview

{radar_svg}
+

Category Sunburst

{sunburst_svg}
+
+

Category Summary

{category_html}
CategoryMetricsObserved ValuesMissingInvalidCoverage
+{ranking_html} +

Metric Summary

{rows_html}
MetricCategoryCountMeanStd95% CIMissingInvalidBestWorst
+

Outlier Examples

{outlier_html}
+
Tip: CSV and Markdown exports are available from the same CLI for downstream analysis.
+
+ + +""" + with open(output_path, "w", encoding="utf-8") as handle: + handle.write(document) + + +def metric_category(metric: str) -> str: + normalized = _strip_prefix(metric).lower() + for category, names in METRIC_CATEGORIES.items(): + if normalized in {name.lower() for name in names}: + return category + if any(token in normalized for token in ["wer", "cer"]): + return "asr_wer_cer" + if any(token in normalized for token in ["similarity", "sim"]): + return "similarity" + if any(token in normalized for token in ["mos", "quality", "nisqa", "utmos"]): + return "audio_quality" + if any(token in normalized for token in ["pesq", "stoi", "sdr", "snr"]): + return "speech_enhancement" + if any(token in normalized for token in ["f0", "pitch", "mcd"]): + return "pitch_f0" + if any(token in normalized for token in ["distance", "dtw", "rmse"]): + return "distance" + return "other" + + +def metric_direction(metric: str) -> Optional[bool]: + normalized = _strip_prefix(metric).lower() + if any(token in normalized for token in ["wer", "cer", "error", "rmse", "mcd"]): + return False + if "distance" in normalized and "token_distance" not in normalized: + return False + if any( + token in normalized + for token in [ + "similarity", + "sim", + "corr", + "mos", + "quality", + "nisqa", + "utmos", + "pesq", + "stoi", + "sdr", + "snr", + "bleu", + "bert", + "clap", + ] + ): + return True + return None + + +def _collect_input_paths(input_path: str) -> List[Path]: + path = Path(input_path) + if path.is_file(): + return [path] + if path.is_dir(): + paths: List[Path] = [] + for pattern in ("*.jsonl", "*.json", "*.txt", "*.jl"): + paths.extend(sorted(path.glob(pattern))) + if paths: + return paths + raise FileNotFoundError(f"No result files found at {input_path}") + + +def _literal_eval_with_special_floats(line: str) -> Any: + class SpecialFloatTransformer(ast.NodeTransformer): + def visit_Name(self, node: ast.Name) -> ast.AST: + if node.id in {"inf", "Infinity"}: + return ast.copy_location(ast.Constant(float("inf")), node) + if node.id in {"nan", "NaN"}: + return ast.copy_location(ast.Constant(float("nan")), node) + return node + + tree = ast.parse(line, mode="eval") + tree = SpecialFloatTransformer().visit(tree) + ast.fix_missing_locations(tree) + return ast.literal_eval(tree) + + +def _to_float(value: Any) -> Optional[float]: + if isinstance(value, bool): + return None + if isinstance(value, (int, float)): + return float(value) + return None + + +def _sample_std(values: Sequence[float], mean: float) -> float: + if len(values) <= 1: + return 0.0 + variance = sum((value - mean) ** 2 for value in values) / (len(values) - 1) + return math.sqrt(max(variance, 0.0)) + + +def _median(sorted_values: Sequence[float]) -> float: + count = len(sorted_values) + if not count: + return 0.0 + midpoint = count // 2 + if count % 2: + return sorted_values[midpoint] + return (sorted_values[midpoint - 1] + sorted_values[midpoint]) / 2 + + +def _strip_prefix(metric: str) -> str: + parts = metric.split("_") + if len(parts) > 1 and parts[0] not in {"se", "si", "ci", "f0"}: + candidate = "_".join(parts[1:]) + known = {name.lower() for names in METRIC_CATEGORIES.values() for name in names} + if candidate.lower() in known: + return candidate + return metric + + +def _fmt(value: Any) -> str: + if isinstance(value, float): + if not math.isfinite(value): + return str(value) + return f"{value:.4g}" + return str(value) + + +def _summary_row(summary: MetricSummary) -> Dict[str, Any]: + return { + "metric": summary.name, + "category": summary.category, + "count": summary.count, + "missing": summary.missing, + "invalid": summary.invalid, + "mean": summary.mean, + "median": summary.median, + "std": summary.std, + "stderr": summary.stderr, + "ci95_low": summary.ci95_low, + "ci95_high": summary.ci95_high, + "min": summary.minimum, + "max": summary.maximum, + "higher_is_better": summary.higher_is_better, + "best_key": summary.best_key, + "best_value": summary.best_value, + "worst_key": summary.worst_key, + "worst_value": summary.worst_value, + "outliers": "; ".join( + f"{key}:{_fmt(value)} (z={_fmt(z_score)})" + for key, value, z_score in summary.outliers + ), + } + + +def _metric_html_row(summary: MetricSummary) -> str: + ci = f"{_fmt(summary.ci95_low)} to {_fmt(summary.ci95_high)}" + best = f"{html.escape(summary.best_key)} ({_fmt(summary.best_value)})" + worst = f"{html.escape(summary.worst_key)} ({_fmt(summary.worst_value)})" + return ( + "" + f"{html.escape(summary.name)}" + f"{html.escape(summary.category)}" + f"{summary.count}" + f"{_fmt(summary.mean)}" + f"{_fmt(summary.std)}" + f"{html.escape(ci)}" + f"{summary.missing}" + f"{summary.invalid}" + f"{best}" + f"{worst}" + "" + ) + + +def _outlier_html(summaries: Sequence[MetricSummary]) -> str: + blocks = [] + for summary in summaries: + if not summary.outliers: + continue + items = "".join( + f"
  • {html.escape(key)}: {_fmt(value)} (z={_fmt(z_score)})
    • " + for key, value, z_score in summary.outliers + ) + blocks.append(f"

    {html.escape(summary.name)}

    ") + return ( + "\n".join(blocks) + if blocks + else '

    No z-score outliers >= 2.0 were detected.

    ' + ) + + +def _ranking_html(analysis: Dict[str, Any]) -> str: + groups = analysis.get("groups") or {} + rankings = groups.get("rankings") or {} + if not rankings: + return "" + rows = [] + for metric, ranking in rankings.items(): + if not ranking: + continue + top = ranking[0] + rows.append( + f"{html.escape(metric)}{html.escape(top['group'])}{_fmt(top['mean'])}{top['count']}" + ) + if not rows: + return "" + group_by = html.escape(str(analysis.get("group_by"))) + return f"

    Per-Metric Ranking by {group_by}

    {''.join(rows)}
    MetricTop GroupMeanCount
    " + + +def _radar_svg(summaries: Sequence[MetricSummary]) -> str: + if not summaries: + return '

    No metrics available.

    ' + width = 420 + height = 360 + cx = width / 2 + cy = height / 2 + radius = 118 + max_mean = max(abs(summary.mean) for summary in summaries) or 1.0 + points = [] + labels = [] + for index, summary in enumerate(summaries): + angle = -math.pi / 2 + 2 * math.pi * index / len(summaries) + scaled = min(abs(summary.mean) / max_mean, 1.0) + x = cx + math.cos(angle) * radius * scaled + y = cy + math.sin(angle) * radius * scaled + points.append(f"{x:.2f},{y:.2f}") + lx = cx + math.cos(angle) * (radius + 34) + ly = cy + math.sin(angle) * (radius + 34) + labels.append( + f'{html.escape(summary.name[:18])}' + ) + rings = [] + for factor in (0.25, 0.5, 0.75, 1.0): + ring_points = [] + for index in range(len(summaries)): + angle = -math.pi / 2 + 2 * math.pi * index / len(summaries) + ring_points.append( + f"{cx + math.cos(angle) * radius * factor:.2f},{cy + math.sin(angle) * radius * factor:.2f}" + ) + rings.append( + f"" + ) + axes = [] + for index in range(len(summaries)): + angle = -math.pi / 2 + 2 * math.pi * index / len(summaries) + axes.append( + f'' + ) + return ( + f'' + + "".join(rings) + + "".join(axes) + + f"" + + "".join(labels) + + "" + ) + + +def _sunburst_svg(category_rows: Sequence[Dict[str, Any]]) -> str: + if not category_rows: + return '

    No categories available.

    ' + width = 420 + height = 360 + cx = width / 2 + cy = height / 2 + total = sum(row["metrics"] for row in category_rows) or 1 + start = -math.pi / 2 + colors = [ + "#1f8a70", + "#c05621", + "#4267ac", + "#8a6f2a", + "#287f9e", + "#9a4f7a", + "#5f7f32", + "#555f6f", + ] + paths = [] + labels = [] + for index, row in enumerate(category_rows): + sweep = 2 * math.pi * row["metrics"] / total + end = start + sweep + paths.append( + _arc_path(cx, cy, 58, 132, start, end, colors[index % len(colors)]) + ) + mid = (start + end) / 2 + labels.append( + f"{html.escape(row['category'][:16])}" + ) + start = end + return ( + f'' + f'' + f'VERSA' + f'metrics' + + "".join(paths) + + "".join(labels) + + "" + ) + + +def _arc_path( + cx: float, + cy: float, + inner: float, + outer: float, + start: float, + end: float, + color: str, +) -> str: + large = 1 if end - start > math.pi else 0 + p1 = (cx + math.cos(start) * outer, cy + math.sin(start) * outer) + p2 = (cx + math.cos(end) * outer, cy + math.sin(end) * outer) + p3 = (cx + math.cos(end) * inner, cy + math.sin(end) * inner) + p4 = (cx + math.cos(start) * inner, cy + math.sin(start) * inner) + d = ( + f"M {p1[0]:.2f} {p1[1]:.2f} " + f"A {outer} {outer} 0 {large} 1 {p2[0]:.2f} {p2[1]:.2f} " + f"L {p3[0]:.2f} {p3[1]:.2f} " + f"A {inner} {inner} 0 {large} 0 {p4[0]:.2f} {p4[1]:.2f} Z" + ) + return f''