Add ASR loading logic of paired glob files

docs/loaders/asr.md

@@ -2,7 +2,7 @@
 
 Loader for **ASR** (Automatic Speech Recognition) datasets.
 
-There are two parsing strategies for ASR datasets, controlled by the `root_strategy` field in the schema.
+There are three parsing strategies for ASR datasets, controlled by the `root_strategy` field in the schema.
 
 ## Strategies
 
@@ -32,6 +32,33 @@ Each split (e.g. `train`, `dev`, `test`) is stored in a separate file. The loade
 | `columns` | ✗ | *(optional)* Column mappings applied to every split frame. |
 | `base_audio_path` | ✗ | *(optional)* Directory prefix or list of directories used to resolve `file_path` dtype columns. |
 
+### Paired-glob JSON strategy (`root_strategy: "paired_glob"`)
+
+For datasets with **no central index file**, where each audio file is paired
+with a JSON sidecar (e.g. `recording.merged.json` + `recording.wav`). The
+loader globs for the JSON files, flattens each one into rows, and applies the
+regular column mappings.
+
+When `record_path` is set, the named top-level JSON key must hold a **list of
+records** (e.g. time-aligned utterances) and each record becomes one DataFrame
+row. The remaining top-level keys are flattened with dot notation
+(`audio.filename`, `metadata.speaker2_gender`, …) and repeated on every row of
+that file. Without `record_path`, each JSON file yields a single row.
+
+Audio pairing does not rely on filename-stem matching: source the `audio_path`
+column from a filename field inside the JSON and resolve it with
+`path_match_strategy: "exact"`.
+
+**Controlled by:**
+
+| Field | Required | Description |
+|---|---|---|
+| `format` | ✓ | Must be `"json"`. |
+| `file_pattern` | ✓ | Glob pattern to find the JSON sidecars (e.g. `"**/*.merged.json"`). |
+| `columns` | ✓ | Mapping of logical column names to (dot-notation) source columns and dtypes. |
+| `record_path` | ✗ | *(optional)* Top-level JSON key holding the list of records; one row per record. |
+| `audio_extension` | ✗ | *(optional)* Extension of the paired audio files (e.g. `".wav"`), documentation / fallback. |
+
 ---
 
 ## Examples
@@ -280,3 +307,68 @@ columns:
     optional: true
 
 ```
+
+### Paired-glob JSON schema
+
+Each `*.merged.json` sidecar describes one WAV recording: an `audio` block
+(with the exact WAV filename), a flat `metadata` block, and a `transcriptions`
+array of time-aligned utterances. The schema below yields one row per
+utterance, with the per-recording `audio.*` / `metadata.*` fields repeated on
+every row:
+
+```yaml
+dataset_id: "xxx"
+task: "ASR"
+root_strategy: "paired_glob"
+format: "json"
+
+file_pattern: "**/*.merged.json"
+audio_extension: ".wav"
+
+record_path: "transcriptions"
+
+columns:
+  audio_path:
+    source_column: "audio.filename"
+    dtype: "file_path"
+    path_match_strategy: "exact"
+  transcription:
+    source_column: "text"
+    dtype: "string"
+  utterance_id:
+    source_column: "utt_id"
+    dtype: "string"
+    optional: true
+  speaker_id:
+    source_column: "speaker"
+    dtype: "category"
+    optional: true
+  start_time:
+    source_column: "start_time"
+    dtype: "float"
+    optional: true
+  end_time:
+    source_column: "end_time"
+    dtype: "float"
+    optional: true
+  audio_duration_sec:
+    source_column: "audio.duration_sec"
+    dtype: "float"
+    optional: true
+  sample_rate_hz:
+    source_column: "audio.sample_rate_hz"
+    dtype: "int"
+    optional: true
+  gender:
+    source_column: "metadata.speaker2_gender"
+    dtype: "category"
+    optional: true
+  topic:
+    source_column: "metadata.user_topic"
+    dtype: "string"
+    optional: true
+  validation_id:
+    source_column: "metadata.val_id"
+    dtype: "string"
+    optional: true
+```

docs/schema_documentation.md

@@ -67,7 +67,7 @@ strategy is inferred from the fields present in the schema:
 |---|---------------------------------------------------------------------|---|
 | **Index-based** (default) | A metadata file (CSV / TSV / pipe-delimited) lists each sample.     | `index_file`, `columns` |
 | **Multi-split** | Multiple split files (train, dev, test, …) each containing samples. | `root_strategy: "multi_split"`, `splits` |
-| **Paired-glob** | Each audio file has a matching `.txt` file, no index file at all.   | `root_strategy: "paired_glob"`, `file_pattern`, `audio_extension` |
+| **Paired-glob** | Each audio file has a matching sidecar file (`.txt` for TTS, JSON for ASR), no index file at all. | `root_strategy: "paired_glob"`, `file_pattern`, `audio_extension` (TTS) / `format: "json"`, `record_path`, `columns` (ASR) |
 | **Glob** | Directory-structured dataset with metadata encoded in the path hierarchy. | `root_strategy: "glob"`, `file_pattern` |
 
 ### Index-based fields
@@ -94,12 +94,30 @@ strategy is inferred from the fields present in the schema:
 
 ### Paired-glob fields
 
+**TTS (text sidecars)** — each audio file has a matching `.txt` file with the
+transcription; pairing is done on the filename stem:
+
 | Field | Default | Required | Description |
 |---|---|---|---|
 | `root_strategy` | — | ✓ | Must be `"paired_glob"`. |
 | `file_pattern` | — | ✓ | Glob pattern to find text files (e.g. `"**/*.txt"`). |
 | `audio_extension` | — | ✓ | Extension of the matching audio files (e.g. `".webm"`). |
 
+**ASR (JSON sidecars)** — each audio file has a matching JSON file holding the
+audio filename, metadata, and (optionally) a list of time-aligned utterance
+records:
+
+| Field | Default | Required | Description |
+|---|---|---|---|
+| `root_strategy` | — | ✓ | Must be `"paired_glob"`. |
+| `format` | — | ✓ | Must be `"json"`. |
+| `file_pattern` | — | ✓ | Glob pattern to find the JSON sidecars (e.g. `"**/*.merged.json"`). |
+| `columns` | — | ✓ | Column mappings over the flattened JSON; nested keys use dot notation (e.g. `audio.filename`, `metadata.speaker2_gender`). |
+| `record_path` | — | ✗ | Top-level JSON key holding a list of records (e.g. `"transcriptions"`); each record becomes one row and the remaining top-level keys are repeated per row. When omitted, each JSON file yields one row. |
+| `audio_extension` | — | ✗ | Extension of the paired audio files (e.g. `".wav"`). Pairing normally comes from a filename field inside the JSON, mapped as a `file_path` column with `path_match_strategy: "exact"`. |
+
+See [ASR loader](./loaders/asr.md) for a complete example.
+
 ### Glob fields
 
 | Field | Default | Required | Description |

src/datacollective/schema.py

@@ -125,6 +125,13 @@ class DatasetSchema(BaseModel):
     content_mapping: ContentMapping | None = Field(
         default=None, description="Mapping for glob-based content extraction"
     )
+    record_path: str | None = Field(
+        default=None,
+        description=(
+            "for JSON sidecar files: top-level key holding the list of records "
+            '(e.g. "transcriptions"); one DataFrame row per record'
+        ),
+    )
 
     # --- Multi-split strategy (e.g. Common Voice) ---
     splits: list[str] | None = Field(
@@ -267,6 +274,7 @@ def _parse_schema(raw: str | dict[str, Any] | Path) -> DatasetSchema:
         "file_pattern",
         "audio_extension",
         "content_mapping",
+        "record_path",
         "splits",
         "splits_file_pattern",
         "sections",
@@ -290,6 +298,7 @@ def _parse_schema(raw: str | dict[str, Any] | Path) -> DatasetSchema:
         file_pattern=data.get("file_pattern"),
         audio_extension=data.get("audio_extension"),
         content_mapping=content_mapping,
+        record_path=data.get("record_path"),
         splits=data.get("splits"),
         splits_file_pattern=data.get("splits_file_pattern"),
         sections=data.get("sections"),

src/datacollective/schema_loaders/tasks/asr.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import json
 from pathlib import Path
 
 import pandas as pd
@@ -21,6 +22,16 @@ def __init__(self, schema: DatasetSchema, extract_dir: Path) -> None:
                 raise ValueError(
                     "ASR multi_split schema must specify 'splits' (list of split names)"
                 )
+        elif schema.root_strategy == Strategy.PAIRED_GLOB:
+            if (schema.format or "").casefold() != "json":
+                raise ValueError("ASR paired_glob schema only supports 'format: json'")
+            if not schema.file_pattern:
+                raise ValueError("ASR paired_glob schema must specify 'file_pattern'")
+            if not schema.columns:
+                raise ValueError(
+                    "ASR paired_glob schema must specify at least two column mappings "
+                    "for audio and transcription"
+                )
         else:
             if not schema.index_file:
                 raise ValueError("ASR schema must specify 'index_file'")
@@ -32,9 +43,62 @@ def __init__(self, schema: DatasetSchema, extract_dir: Path) -> None:
     def load(self) -> pd.DataFrame:
         if self.schema.root_strategy == Strategy.MULTI_SPLIT:
             return self._load_multi_split()
+        if self.schema.root_strategy == Strategy.PAIRED_GLOB:
+            return self._load_paired_glob_json()
         raw_df = self._load_index_file()
         return self._apply_column_mappings(raw_df)
 
+    def _load_paired_glob_json(self) -> pd.DataFrame:
+        """
+        Load an ASR dataset where each audio file is paired with a JSON sidecar
+        (matched via ``file_pattern``) instead of a central index file.
+
+        When ``record_path`` is set, the JSON key it names must hold a list of
+        records (e.g. time-aligned utterances) and each record becomes one row;
+        the remaining top-level keys are flattened with dot notation
+        (audio.filename, ...) and repeated on
+        every row of that file.  Without ``record_path`` each JSON file yields
+        a single row.  Column mappings are then applied as for index files, so
+        ``file_path`` columns (typically sourced from a filename field inside
+        the JSON) resolve through the usual audio-path machinery.
+        """
+        assert self.schema.file_pattern is not None
+
+        json_files = sorted(self.extract_dir.rglob(self.schema.file_pattern))
+        json_files = [p for p in json_files if not p.name.startswith("._")]
+        if not json_files:
+            raise FileNotFoundError(
+                f"No files matching '{self.schema.file_pattern}' "
+                f"found under '{self.extract_dir}'"
+            )
+
+        logger.debug(
+            f"Found {len(json_files)} JSON files matching '{self.schema.file_pattern}'"
+        )
+
+        record_path = self.schema.record_path
+        frames: list[pd.DataFrame] = []
+        for path in json_files:
+            data = json.loads(path.read_text(encoding=self.schema.encoding))
+            if record_path:
+                if record_path not in data:
+                    raise KeyError(
+                        f"record_path '{record_path}' not found in '{path}'. "
+                        f"Available keys: {list(data)}"
+                    )
+                frame = pd.json_normalize(data, record_path=record_path)
+                meta = pd.json_normalize(
+                    {key: value for key, value in data.items() if key != record_path}
+                )
+                for column in meta.columns:
+                    frame[column] = meta[column].iloc[0]
+            else:
+                frame = pd.json_normalize(data)
+            frames.append(frame)
+
+        raw_df = pd.concat(frames, ignore_index=True)
+        return self._apply_column_mappings(raw_df)
+
     def _load_multi_split(self) -> pd.DataFrame:
         """
         Load all split TSV/CSV files whose stems match the ``splits`` list,