""" Event Density task generator for temporal reasoning dataset. Multi-hop task: Compares event counts or event frequency across temporal regions of the audio (first half vs second half, before vs after anchor). """ 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, ) from tasks.multihop_base import MultihopBaseGenerator class EventDensityTaskGenerator(MultihopBaseGenerator): """Generates event_density task dataset samples.""" TASK_NAME = "event_density" def generate_sample( self, sample_id: int, target_duration_seconds: float = None, question_type: str = None, ) -> Optional[Dict]: """ Generate a single event_density sample. Pipeline: 1. Build scene with 6-10 events, intentionally asymmetric distribution 2. Compute event counts per region 3. Generate question about which region has more events """ n_events = random.randint( self.task_config.get("min_events", 6), self.task_config.get("max_events", 10), ) # Sample categories — allow repeats for density counting n_unique = random.randint( max(2, n_events // 3), min(n_events, len(self.dataset.CATEGORIES)) ) categories_pool = self.dataset.sample_categories(n_unique) categories = list(categories_pool) while len(categories) < n_events: categories.append(random.choice(categories_pool)) # Intentionally create asymmetry by biasing distribution # Put more events of some categories in first or second half random.shuffle(categories) # Build audio final_audio, source_files, events_meta = self._build_scene( categories, target_duration_seconds ) output_path = self.audio_output / f"{sample_id}.wav" final_audio.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 ) 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, "target_duration_s": target_duration_seconds, "actual_duration_s": len(final_audio) / 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 event_density sample {sample_id}: " f"{n_events} events, type={question_type}" ) return metadata def _build_scene(self, categories, target_duration_s): """Build sequential audio scene.""" from pydub import AudioSegment as PydubSegment n = len(categories) num_gaps = n - 1 per_event_s = max(self.source_clip_duration, (target_duration_s or 30) / n) source_files = [] audio_segments = [] for cat in categories: fname, fpath = self.dataset.sample_file_from_category(cat) audio = self.audio_processor.load_audio(fpath) audio_segments.append(audio) source_files.append(fname) if num_gaps > 0: gap_durations = generate_controlled_gap_durations( num_gaps, min_gap_ms=200, max_gap_ms=1000, gap_multiplier=1.5 ) else: gap_durations = [] result = audio_segments[0] events_meta = [ {"index": 0, "category": categories[0], "start_ms": 0, "end_ms": len(audio_segments[0])} ] current_ms = len(audio_segments[0]) for i in range(1, n): gap_ms = gap_durations[i - 1] result = result + PydubSegment.silent(duration=gap_ms) current_ms += gap_ms event_start = current_ms result = result + audio_segments[i] current_ms += len(audio_segments[i]) events_meta.append( {"index": i, "category": categories[i], "start_ms": event_start, "end_ms": current_ms} ) return result, source_files, events_meta def _generate_question(self, question_type, categories, events_meta): """Generate event density question.""" n = len(categories) midpoint = n // 2 if question_type == "half_density": first_half = categories[:midpoint] second_half = categories[midpoint:] count_first = len(first_half) count_second = len(second_half) if count_first > count_second: correct = "first half" elif count_second > count_first: correct = "second half" else: correct = "equal" mcq_text = self.task_config["mcq_questions"]["half_density"] open_text = self.task_config["open_text_questions"]["half_density"] options = ["first half", "second half", "equal"] other = [c for c in self.dataset.CATEGORIES[:2] if c not in options] options.append(f"cannot determine") 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 = {"count_first_half": count_first, "count_second_half": count_second} return mcq_data, open_data, q_meta elif question_type == "before_after_density": # Pick anchor anchor_idx = random.randint(1, n - 2) if n > 2 else n // 2 anchor_sound = categories[anchor_idx] count_before = anchor_idx count_after = n - anchor_idx - 1 if count_before > count_after: correct = f"before {anchor_sound}" elif count_after > count_before: correct = f"after {anchor_sound}" else: correct = "equal" mcq_text = self.task_config["mcq_questions"]["before_after_density"].format(anchor_sound=anchor_sound) open_text = self.task_config["open_text_questions"]["before_after_density"].format(anchor_sound=anchor_sound) options = [f"before {anchor_sound}", f"after {anchor_sound}", "equal"] options.append("cannot determine") 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 = { "anchor_sound": anchor_sound, "count_before": count_before, "count_after": count_after, } return mcq_data, open_data, q_meta elif question_type == "label_density": # Pick anchor and target anchor_idx = random.randint(1, n - 2) if n > 2 else n // 2 anchor_sound = categories[anchor_idx] unique_cats = list(set(categories)) target_sound = random.choice(unique_cats) count_before = sum(1 for i in range(anchor_idx) if categories[i] == target_sound) count_after = sum(1 for i in range(anchor_idx + 1, n) if categories[i] == target_sound) if count_before > count_after: correct = f"before {anchor_sound}" elif count_after > count_before: correct = f"after {anchor_sound}" else: correct = "equal" mcq_text = self.task_config["mcq_questions"]["label_density"].format(target_sound=target_sound, anchor_sound=anchor_sound) open_text = self.task_config["open_text_questions"]["label_density"].format(target_sound=target_sound, anchor_sound=anchor_sound) options = [f"before {anchor_sound}", f"after {anchor_sound}", "equal"] options.append("cannot determine") 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 = { "anchor_sound": anchor_sound, "target_sound": target_sound, "count_before": count_before, "count_after": count_after, } return mcq_data, open_data, q_meta return None, None, {} def main(config_path: str = None): """Main entry point for event_density task generation.""" 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( "event_density_task", log_file=str(Path(config["output"]["base_path"]) / config["logging"]["log_file"]), level=config["logging"]["level"], console_output=config["logging"]["console_output"], ) generator = EventDensityTaskGenerator(config, logger) generator.generate_dataset() if __name__ == "__main__": main()