feat(audio-postprocessing): agent-ready timestamp mapper (refined-span + multi-speaker fixes)#2173
feat(audio-postprocessing): agent-ready timestamp mapper (refined-span + multi-speaker fixes)#2173shubhamNvidia wants to merge 2 commits into
Conversation
…idency resolver Shared base imported by the audio stage modules to make them agent-ready: - _agent_ready.py: AgentReady mixin, StageContract/IOSpec/Gates/Role - _residency.py: input residency resolver (file/waveform/auto) - common.py: agent-ready helpers (ensure_mono/ensure_waveform_2d) Excludes the agent orchestration layer (registry/catalog/conformance/planning/agent).
…escribe/StageContract + residency)
Greptile SummaryThis PR makes
Confidence Score: 3/5The infrastructure additions (AgentReady, residency helpers) and key-field parameterization are safe to merge. The timestamp logic in the new diar-with-mappings fallback path can produce corrupt records when diar segments cross original-file boundaries. The new nemo_curator/stages/audio/postprocessing/timestamp_mapper.py — specifically Important Files Changed
Flowchart%%{init: {'theme': 'neutral'}}%%
flowchart TD
A[process task] --> B{segment_mappings\nin metadata?}
B -- Yes --> C{start_ms AND\nend_ms present?}
B -- No --> NM[_build_output_item_no_mapping]
C -- Yes --> D{end_ms > start_ms?}
D -- No --> DROP1[return empty list\nlog warning]
D -- Yes --> E[_translate_to_original\nstart_ms, end_ms]
E --> F{len ranges?}
F -- "> 1" --> DROP2[return empty list\nmulti-mapping span]
F -- "== 1" --> OUT1[_build_output_item\noriginal coords]
F -- "== 0" --> DROP3[return empty list\nno overlap]
C -- No --> G{diar_segments\npresent?}
G -- No --> DROP4[return empty list\nno timing source]
G -- Yes --> H[_build_output_from_diar_and_mappings]
H --> I{any ranges\nreturned?}
I -- No --> DROP5[return empty list]
I -- Yes --> OUT2[union min/max across\nranges — ⚠️ may mix files]
NM --> J{start_ms + end_ms?}
J -- Yes --> OUT3[direct original coords]
J -- No --> K{diar_segments?}
K -- Yes --> OUT4[span first to last\n+ speaking_duration]
K -- No --> L{duration?}
L -- Yes --> OUT5[whole-file 0 to dur]
L -- No --> OUT6[zero-duration row\nlog warning]
%%{init: {'theme': 'base', 'themeVariables': {"darkMode": true, "background": "#0d1117", "primaryColor": "#21262d", "primaryTextColor": "#e6edf3", "primaryBorderColor": "#8b949e", "lineColor": "#8b949e", "textColor": "#e6edf3", "edgeLabelBackground": "#161b22", "actorBkg": "#21262d", "actorBorder": "#8b949e", "actorTextColor": "#e6edf3", "actorLineColor": "#8b949e", "signalColor": "#8b949e", "signalTextColor": "#e6edf3", "noteBkgColor": "#373320", "noteBorderColor": "#d4a72c", "noteTextColor": "#f0e6c0", "labelBoxBkgColor": "#21262d", "labelBoxBorderColor": "#8b949e", "labelTextColor": "#e6edf3", "loopTextColor": "#e6edf3", "activationBkgColor": "#30363d", "activationBorderColor": "#8b949e"}}}%%
flowchart TD
A[process task] --> B{segment_mappings\nin metadata?}
B -- Yes --> C{start_ms AND\nend_ms present?}
B -- No --> NM[_build_output_item_no_mapping]
C -- Yes --> D{end_ms > start_ms?}
D -- No --> DROP1[return empty list\nlog warning]
D -- Yes --> E[_translate_to_original\nstart_ms, end_ms]
E --> F{len ranges?}
F -- "> 1" --> DROP2[return empty list\nmulti-mapping span]
F -- "== 1" --> OUT1[_build_output_item\noriginal coords]
F -- "== 0" --> DROP3[return empty list\nno overlap]
C -- No --> G{diar_segments\npresent?}
G -- No --> DROP4[return empty list\nno timing source]
G -- Yes --> H[_build_output_from_diar_and_mappings]
H --> I{any ranges\nreturned?}
I -- No --> DROP5[return empty list]
I -- Yes --> OUT2[union min/max across\nranges — ⚠️ may mix files]
NM --> J{start_ms + end_ms?}
J -- Yes --> OUT3[direct original coords]
J -- No --> K{diar_segments?}
K -- Yes --> OUT4[span first to last\n+ speaking_duration]
K -- No --> L{duration?}
L -- Yes --> OUT5[whole-file 0 to dur]
L -- No --> OUT6[zero-duration row\nlog warning]
Reviews (1): Last reviewed commit: "feat(audio-postprocessing): make postpro..." | Re-trigger Greptile |
![greptile-apps[bot]](https://avatars.githubusercontent.com/in/867647?s=60&v=4){kind=small}
| result: dict[str, Any] = { | ||
| self.original_file_key: ranges[0]["original_file"], | ||
| self.original_start_ms_key: min(r["original_start_ms"] for r in ranges), | ||
| self.original_end_ms_key: max(r["original_end_ms"] for r in ranges), | ||
| } | ||
| result[self.duration_ms_key] = result[self.original_end_ms_key] - result[self.original_start_ms_key] | ||
| result[self.duration_key] = result[self.duration_ms_key] / 1000.0 |
There was a problem hiding this comment.
When diar_segments span a mapping boundary (i.e. the concat audio was composed from multiple original files), _translate_to_original returns entries with different original_file values. The current code picks ranges[0]["original_file"] as the file key, then computes min(r["original_start_ms"]) and max(r["original_end_ms"]) across all ranges — including those from different files. The result is a record whose original_start_ms/original_end_ms are timestamps from a file other than the one recorded in original_file.
The start_ms/end_ms code path avoids this by explicitly rejecting any segment that spans more than one mapping (len(original_ranges) > 1 → return []). The diar path has no equivalent guard, so a cross-file diar segment silently produces a corrupt record rather than being dropped.
| def _build_output_from_diar_and_mappings( | ||
| self, | ||
| item: dict[str, Any], | ||
| diar_segments: list, | ||
| mappings: list[dict[str, Any]], | ||
| ) -> dict[str, Any] | None: | ||
| """Compose concat-time diar segments through concat->original mappings. | ||
|
|
||
| Each diar segment is ``[start_sec, end_sec]`` in concatenated time. | ||
| We translate every segment to original-file time and span the union. | ||
| Returns ``None`` if no segment overlaps any mapping. | ||
| """ | ||
| ranges: list[dict[str, Any]] = [] | ||
| for seg in diar_segments: | ||
| try: | ||
| start_sec, end_sec = float(seg[0]), float(seg[1]) | ||
| except (TypeError, IndexError, ValueError): | ||
| continue | ||
| ranges.extend(_translate_to_original(mappings, int(start_sec * 1000), int(end_sec * 1000))) | ||
| if not ranges: | ||
| return None | ||
| result: dict[str, Any] = { | ||
| self.original_file_key: ranges[0]["original_file"], | ||
| self.original_start_ms_key: min(r["original_start_ms"] for r in ranges), | ||
| self.original_end_ms_key: max(r["original_end_ms"] for r in ranges), | ||
| } | ||
| result[self.duration_ms_key] = result[self.original_end_ms_key] - result[self.original_start_ms_key] | ||
| result[self.duration_key] = result[self.duration_ms_key] / 1000.0 | ||
| self._copy_passthrough(item, result) | ||
| return result |
There was a problem hiding this comment.
speaking_duration is never computed in the with-mappings diar path
_build_output_item_no_mapping computes speaking_duration and diar_segments from the raw diar segments and writes them as core output fields. _build_output_from_diar_and_mappings does not: those keys are only emitted if they happen to already exist in item and survive _copy_passthrough. Since diar_segments_key is absent from _DEFAULT_PASSTHROUGH_KEYS, it will always be missing in the with-mappings diar path. speaking_duration will be copied only if set by a prior stage. The inconsistency means downstream consumers cannot rely on these fields being present when the stage is used with segment mappings and diarization, despite describe() declaring both as write keys.
1 participant
Summary
Makes
TimestampMapperStageagent-ready and fixes two timestamp-correctness bugs.What the stage does
Translates per-segment times measured on concatenated/processed audio back to the original source timeline, using the
segment_mappingsmetadata produced upstream (see_translate_to_original).Fixes
start_ms/end_ms(e.g. from a VAD pass) AND coarserdiar_segments, the refined values are now mapped through and preferred; diarization spans are used only as a fallback when refined times are absent.Contract
describe()declares OR-shaped reads (start_ms+end_ms|diar_segments|duration) andmetadata_reads=[segment_mappings].Notes for reviewers
preserves_upstream_keys=Falseand REPLACEStask.data, keeping only apassthrough_keyswhitelist (defaults keep filter scores + speaker metadata likespeaker_id). Non-whitelisted keys (e.g.text,segment_num) are dropped — setpassthrough_keysto retain them._translate_to_original(overlap/offset math, ms units) and the refined-vs-diar branch.Test plan
pytest tests/stages/audio/postprocessing -qpytest tests/stages/audio -q -k timestamp_mapper