tasks/task_duration_gap.py

"""
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()