""" Duration Gap task generator for temporal reasoning dataset. Multi-hop inter-task: Combines event-duration reasoning with silence-gap reasoning by comparing sound duration to nearby silent intervals. Uses PreprocessedESC50Dataset for accurate effective durations. """ import random from pathlib import Path from typing import Dict, List, Optional from utils import ( setup_logger, set_random_seed, concatenate_to_target_duration, generate_controlled_gap_durations, create_preprocessed_dataset, ) from tasks.multihop_base import MultihopBaseGenerator class DurationGapTaskGenerator(MultihopBaseGenerator): """Generates duration_gap task dataset samples.""" TASK_NAME = "duration_gap" def __init__(self, config: dict, logger=None): super().__init__(config, logger) # Override dataset with preprocessed version preprocessed_path = self.task_config.get( "preprocessed_data_path", config["tasks"].get("duration", {}).get("preprocessed_data_path", ""), ) self.dataset = create_preprocessed_dataset(config, preprocessed_path=preprocessed_path) def generate_sample( self, sample_id: int, target_duration_seconds: float = None, question_type: str = None, ) -> Optional[Dict]: """ Generate a single duration_gap sample. Pipeline: 1. Build scene with 3-6 events with controlled silences 2. For each event, know its effective duration and adjacent gap durations 3. Compare event duration vs gap duration 4. Generate question """ n_events = random.randint( self.task_config.get("min_events", 3), self.task_config.get("max_events", 6), ) n_unique = min(n_events, len(self.dataset.CATEGORIES)) categories = self.dataset.sample_categories(n_unique) while len(categories) < n_events: categories.append(random.choice(categories[:n_unique])) random.shuffle(categories) # Load audio with known durations from pydub import AudioSegment as PydubSegment source_files = [] audio_segments = [] effective_durations_ms = [] for cat in categories: fname, fpath, eff_dur = self.dataset.sample_file_from_category_with_duration(cat) audio = self.audio_processor.load_audio(fpath) audio = concatenate_to_target_duration(audio, max(self.source_clip_duration, eff_dur)) audio_segments.append(audio) source_files.append(fname) effective_durations_ms.append(int(eff_dur * 1000)) # Generate controlled gaps (wider range for comparison) num_gaps = n_events - 1 gap_min = self.task_config.get("min_gap_ms", 500) gap_max = self.task_config.get("max_gap_ms", 4000) if num_gaps > 0: gap_durations = generate_controlled_gap_durations( num_gaps, min_gap_ms=gap_min, max_gap_ms=gap_max, gap_multiplier=2.0, ) else: gap_durations = [] # Assemble audio result = audio_segments[0] current_ms = len(audio_segments[0]) events_meta = [ {"index": 0, "category": categories[0], "duration_ms": len(audio_segments[0]), "effective_duration_ms": effective_durations_ms[0], "gap_after_ms": gap_durations[0] if gap_durations else 0} ] for i in range(1, n_events): gap_ms = gap_durations[i - 1] result = result + PydubSegment.silent(duration=gap_ms) current_ms += gap_ms result = result + audio_segments[i] current_ms += len(audio_segments[i]) gap_after = gap_durations[i] if i < len(gap_durations) else 0 gap_before = gap_durations[i - 1] events_meta.append( {"index": i, "category": categories[i], "duration_ms": len(audio_segments[i]), "effective_duration_ms": effective_durations_ms[i], "gap_before_ms": gap_before, "gap_after_ms": gap_after} ) # Also set gap_before for first event events_meta[0]["gap_before_ms"] = 0 output_path = self.audio_output / f"{sample_id}.wav" result.export(str(output_path), format="wav") if question_type is None: question_type = random.choice(self.task_config["question_types"]) mcq_data, open_data, q_meta = self._generate_question( question_type, categories, events_meta, gap_durations ) if mcq_data is None: return None metadata = { "id": sample_id, "audio_path": str(output_path.relative_to(self.output_base.parent)), "n_events": n_events, "categories": categories, "source_files": source_files, "question_type": question_type, "gap_durations_ms": gap_durations, "effective_durations_ms": effective_durations_ms, "target_duration_s": target_duration_seconds, "actual_duration_s": len(result) / 1000.0, "mcq_question": mcq_data["question"], "mcq_options": mcq_data["options"], "mcq_correct_answer": mcq_data["correct_answer"], "open_text_question": open_data["question"], "open_text_answer": open_data["correct_answer"], **q_meta, } self.logger.info( f"Generated duration_gap sample {sample_id}: " f"{n_events} events, type={question_type}" ) return metadata def _generate_question(self, question_type, categories, events_meta, gap_durations): """Generate duration vs gap comparison question.""" n = len(categories) if question_type == "event_vs_after_gap": # Pick an event that has a gap after it (not the last event) valid = [e for e in events_meta if e["index"] < n - 1] if not valid: return None, None, {} event = random.choice(valid) sound1 = event["category"] event_dur = event["effective_duration_ms"] gap_after = event["gap_after_ms"] if event_dur >= gap_after: correct = sound1 else: correct = f"silence after {sound1}" mcq_text = self.task_config["mcq_questions"]["event_vs_after_gap"].format(sound1=sound1) open_text = self.task_config["open_text_questions"]["event_vs_after_gap"].format(sound1=sound1) options = [sound1, f"silence after {sound1}"] other = [c for c in self.dataset.CATEGORIES if c != sound1][:2] options.extend(other) random.shuffle(options) options = options[:4] if correct not in options: options[0] = correct option_labels = ["A", "B", "C", "D"] correct_label = option_labels[options.index(correct)] option_map = {l: v for l, v in zip(option_labels, options)} mcq_data = {"question": mcq_text, "options": option_map, "correct_answer": correct_label} open_data = {"question": open_text, "correct_answer": correct} q_meta = {"event_duration_ms": event_dur, "gap_after_ms": gap_after} return mcq_data, open_data, q_meta elif question_type == "event_vs_before_gap": valid = [e for e in events_meta if e["index"] > 0] if not valid: return None, None, {} event = random.choice(valid) sound1 = event["category"] event_dur = event["effective_duration_ms"] gap_before = event["gap_before_ms"] if event_dur >= gap_before: correct = sound1 else: correct = f"silence before {sound1}" mcq_text = self.task_config["mcq_questions"]["event_vs_before_gap"].format(sound1=sound1) open_text = self.task_config["open_text_questions"]["event_vs_before_gap"].format(sound1=sound1) options = [sound1, f"silence before {sound1}"] other = [c for c in self.dataset.CATEGORIES if c != sound1][:2] options.extend(other) random.shuffle(options) options = options[:4] if correct not in options: options[0] = correct option_labels = ["A", "B", "C", "D"] correct_label = option_labels[options.index(correct)] option_map = {l: v for l, v in zip(option_labels, options)} mcq_data = {"question": mcq_text, "options": option_map, "correct_answer": correct_label} open_data = {"question": open_text, "correct_answer": correct} q_meta = {"event_duration_ms": event_dur, "gap_before_ms": gap_before} return mcq_data, open_data, q_meta elif question_type == "gap_vs_event": if n < 3 or len(gap_durations) < 1: return None, None, {} # Pick a gap and compare with a third event's duration gap_idx = random.randint(0, len(gap_durations) - 1) gap_dur = gap_durations[gap_idx] sound1 = categories[gap_idx] sound2 = categories[gap_idx + 1] # Pick a third event other_indices = [i for i in range(n) if i != gap_idx and i != gap_idx + 1] if not other_indices: return None, None, {} third_idx = random.choice(other_indices) sound3 = categories[third_idx] third_dur = events_meta[third_idx]["effective_duration_ms"] if gap_dur >= third_dur: correct = f"the pause between {sound1} and {sound2}" else: correct = sound3 mcq_text = self.task_config["mcq_questions"]["gap_vs_event"].format(sound1=sound1, sound2=sound2, sound3=sound3) open_text = self.task_config["open_text_questions"]["gap_vs_event"].format(sound1=sound1, sound2=sound2, sound3=sound3) options = [f"the pause between {sound1} and {sound2}", sound3] other = [c for c in self.dataset.CATEGORIES if c not in [sound1, sound2, sound3]][:2] options.extend(other) random.shuffle(options) options = options[:4] if correct not in options: options[0] = correct option_labels = ["A", "B", "C", "D"] correct_label = option_labels[options.index(correct)] option_map = {l: v for l, v in zip(option_labels, options)} mcq_data = {"question": mcq_text, "options": option_map, "correct_answer": correct_label} open_data = {"question": open_text, "correct_answer": correct} q_meta = {"gap_duration_ms": gap_dur, "third_event_duration_ms": third_dur} return mcq_data, open_data, q_meta return None, None, {} def main(config_path: str = None): import yaml if config_path is None: config_path = Path(__file__).parent.parent / "config.yaml" with open(config_path, "r") as f: config = yaml.safe_load(f) set_random_seed(config["random_seed"]) logger = setup_logger( "duration_gap_task", log_file=str(Path(config["output"]["base_path"]) / config["logging"]["log_file"]), level=config["logging"]["level"], console_output=config["logging"]["console_output"], ) generator = DurationGapTaskGenerator(config, logger) generator.generate_dataset() if __name__ == "__main__": main()