improvisation_lab/application/interval_practice/web_interval_app.py

"""Web application for interval practice."""

import time
from typing import Any, List, Tuple

import numpy as np

from improvisation_lab.application.base_app import BasePracticeApp
from improvisation_lab.config import Config
from improvisation_lab.domain.music_theory import Intervals
from improvisation_lab.infrastructure.audio import WebAudioProcessor
from improvisation_lab.presentation.interval_practice import (
    IntervalViewTextManager, WebIntervalPracticeView)
from improvisation_lab.service import IntervalPracticeService


class WebIntervalPracticeApp(BasePracticeApp):
    """Web application class for interval practice."""

    def __init__(self, service: IntervalPracticeService, config: Config):
        """Initialize the application using web UI.

        Args:
            service: IntervalPracticeService instance.
            config: Config instance.
        """
        super().__init__(service, config)

        self.audio_processor = WebAudioProcessor(
            sample_rate=config.audio.sample_rate,
            callback=self._process_audio_callback,
            buffer_duration=config.audio.buffer_duration,
        )

        self.text_manager = IntervalViewTextManager()
        self.ui = WebIntervalPracticeView(
            on_generate_melody=self.start,
            on_end_practice=self.stop,
            on_audio_input=self.handle_audio,
            config=config,
        )
        self.base_note = "-"
        self.results_table: List[List[Any]] = []
        self.progress_timer: float = 0.0
        self.is_auto_advance = False
        self.note_duration = 1.5

    def _process_audio_callback(self, audio_data: np.ndarray):
        """Process incoming audio data and update the application state.

        Args:
            audio_data: Audio data to process.
        """
        if not self.is_running or not self.phrases:
            return

        current_note = self.phrases[self.current_phrase_idx][
            self.current_note_idx
        ].value

        result = self.service.process_audio(audio_data, current_note)

        # Update status display
        self.text_manager.update_pitch_result(result, self.is_auto_advance)

        # Progress to next note if current note is complete
        if self.is_auto_advance:
            current_time = time.time()
            if current_time - self.progress_timer >= self.note_duration:
                self._advance_to_next_note()
                self.progress_timer = current_time
        elif result.remaining_time <= 0:
            self._advance_to_next_note()

        self.text_manager.update_phrase_text(self.current_phrase_idx, self.phrases)

    def _advance_to_next_note(self):
        """Advance to the next note or phrase."""
        if self.phrases is None:
            return

        self.update_results_table()
        self.current_note_idx += 1
        if self.current_note_idx >= len(self.phrases[self.current_phrase_idx]):
            self.current_note_idx = 0
            self.current_phrase_idx += 1
            if self.current_phrase_idx >= len(self.phrases):
                self.current_phrase_idx = 0
            self.base_note = self.phrases[self.current_phrase_idx][
                self.current_note_idx
            ].value

    def handle_audio(self, audio: Tuple[int, np.ndarray]) -> Tuple[str, str, str, List]:
        """Handle audio input from Gradio interface.

        Args:
            audio: Audio data to process.

        Returns:
            Tuple[str, str, str, List]:
                The current base note including the next base note,
                target note, result text, and results table.
        """
        if not self.is_running:
            return "-", "Not running", "Start the session first", []

        self.audio_processor.process_audio(audio)

        return (
            self.base_note,
            self.text_manager.phrase_text,
            self.text_manager.result_text,
            self.results_table,
        )

    def start(
        self,
        interval: str,
        direction: str,
        number_problems: int,
        is_auto_advance: bool,
        note_duration: float,
    ) -> Tuple[str, str, str, List]:
        """Start a new practice session.

        Args:
            interval: Interval to move to and back.
            direction: Direction to move to and back.
            number_problems: Number of problems to generate.
            is_auto_advance: Whether to automatically advance to the next note.
            note_duration: Duration of each note in seconds.

        Returns:
            Tuple[str, str, str, List]:
                The current base note including the next base note,
                target note, result text, and results table.
        """
        semitone_interval = Intervals.INTERVALS_MAP.get(interval, 0)

        if direction == "Down":
            semitone_interval = -semitone_interval
        self.phrases = self.service.generate_melody(
            num_notes=number_problems, interval=semitone_interval
        )
        self.current_phrase_idx = 0
        self.current_note_idx = 0
        self.is_running = True

        present_note = self.phrases[0][0].value
        self.base_note = present_note

        if not self.audio_processor.is_recording:
            self.text_manager.initialize_text()
            self.audio_processor.start_recording()

        self.text_manager.update_phrase_text(self.current_phrase_idx, self.phrases)
        self.results_table = []

        self.is_auto_advance = is_auto_advance
        self.note_duration = note_duration
        self.progress_timer = time.time()

        return (
            self.base_note,
            self.text_manager.phrase_text,
            self.text_manager.result_text,
            self.results_table,
        )

    def stop(self) -> Tuple[str, str, str]:
        """Stop the current practice session.

        Returns:
            tuple[str, str, str]:
                The current base note including the next base note,
                target note, and result text.
        """
        self.is_running = False
        self.base_note = "-"
        if self.audio_processor.is_recording:
            self.audio_processor.stop_recording()
            self.text_manager.terminate_text()
        return (
            self.base_note,
            self.text_manager.phrase_text,
            self.text_manager.result_text,
        )

    def launch(self, **kwargs):
        """Launch the application."""
        self.ui.launch(**kwargs)

    def update_results_table(self):
        """Update the results table with the latest result."""
        if not self.is_auto_advance:
            return

        target_note = self.phrases[self.current_phrase_idx][self.current_note_idx].value
        if self.base_note == target_note:
            return
        detected_note = self.text_manager.result_text.split("|")[1].strip()
        detected_note = detected_note.replace("Your note: ", "").replace(" ", "")
        # Result determination
        result = "⭕️" if detected_note == target_note else "X"
        new_result = [
            self.current_phrase_idx + 1,
            self.base_note,
            target_note,
            detected_note,
            result,
        ]

        self.results_table.append(new_result)