Spaces:
Sleeping
Sleeping
File size: 3,290 Bytes
5e84ffc |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 |
"""Console application for all practices."""
import time
from abc import ABC, abstractmethod
from typing import Optional
import numpy as np
from improvisation_lab.application.base_app import BasePracticeApp
from improvisation_lab.config import Config
from improvisation_lab.infrastructure.audio import DirectAudioProcessor
from improvisation_lab.presentation.console_view import ConsolePracticeView
from improvisation_lab.service.base_practice_service import BasePracticeService
class ConsoleBasePracticeApp(BasePracticeApp, ABC):
"""Console application class for all practices."""
def __init__(self, service: BasePracticeService, config: Config):
"""Initialize the application using console UI.
Args:
service: PracticeService instance.
config: Config instance.
"""
super().__init__(service, config)
self.audio_processor = DirectAudioProcessor(
sample_rate=config.audio.sample_rate,
callback=self._process_audio_callback,
buffer_duration=config.audio.buffer_duration,
)
self.ui: Optional[ConsolePracticeView] = None
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 self.phrases is None:
return
current_note = self._get_current_note()
result = self.service.process_audio(audio_data, current_note)
if self.ui is not None:
self.ui.display_pitch_result(result)
# Progress to next note if current note is complete
if result.remaining_time <= 0:
self._advance_to_next_note()
def _advance_to_next_note(self):
"""Advance to the next note or phrase."""
if self.phrases is None:
return
self.current_note_idx += 1
if self.current_note_idx >= len(self._get_current_phrase()):
self.current_note_idx = 0
self.current_phrase_idx += 1
if self.current_phrase_idx >= len(self.phrases):
self.current_phrase_idx = 0
self.ui.display_phrase_info(self.current_phrase_idx, self.phrases)
def launch(self):
"""Launch the application."""
self.ui.launch()
self.phrases = self._generate_melody()
self.current_phrase_idx = 0
self.current_note_idx = 0
self.is_running = True
if not self.audio_processor.is_recording:
try:
self.audio_processor.start_recording()
self.ui.display_phrase_info(self.current_phrase_idx, self.phrases)
while True:
time.sleep(0.1)
except KeyboardInterrupt:
print("\nStopping...")
finally:
self.audio_processor.stop_recording()
@abstractmethod
def _get_current_note(self):
"""Return the current note to be processed."""
pass
@abstractmethod
def _get_current_phrase(self):
"""Return the current phrase to be processed."""
pass
@abstractmethod
def _generate_melody(self):
"""Generate melody specific to the practice type."""
pass
|