import threading import pydub import time from transitions.extensions import HierarchicalMachine as Machine import os.path import audioop from .lock import Lock from . import Config, gain, debug_print, error_print from .mixer import Mixer from .music_effect import GainEffect file_lock = Lock("file") class MusicFile(Machine): def __init__(self, filename, mapping, name=None, gain=1): states = [ 'initial', 'loading', 'failed', { 'name': 'loaded', 'children': ['stopped', 'playing', 'paused', 'stopping'] } ] transitions = [ { 'trigger': 'load', 'source': 'initial', 'dest': 'loading' }, { 'trigger': 'fail', 'source': 'loading', 'dest': 'failed' }, { 'trigger': 'success', 'source': 'loading', 'dest': 'loaded_stopped' }, { 'trigger': 'start_playing', 'source': 'loaded_stopped', 'dest': 'loaded_playing' }, { 'trigger': 'pause', 'source': 'loaded_playing', 'dest': 'loaded_paused' }, { 'trigger': 'unpause', 'source': 'loaded_paused', 'dest': 'loaded_playing' }, { 'trigger': 'stop_playing', 'source': ['loaded_playing','loaded_paused'], 'dest': 'loaded_stopping' }, { 'trigger': 'stopped', 'source': 'loaded_stopping', 'dest': 'loaded_stopped', 'after': 'trigger_stopped_events' } ] Machine.__init__(self, states=states, transitions=transitions, initial='initial') self.volume = 100 self.mapping = mapping self.filename = filename self.name = name or filename self.audio_segment = None self.audio_segment_frame_width = 0 self.initial_volume_factor = gain self.music_lock = Lock("music__" + filename) self.wait_event = threading.Event() self.db_gain = 0 self.gain_effects = [] threading.Thread(name="MSMusicLoad", target=self.load).start() def on_enter_loading(self): with file_lock: try: debug_print("Loading « {} »".format(self.name)) self.mixer = self.mapping.mixer or Mixer() initial_db_gain = gain(self.initial_volume_factor * 100) self.audio_segment = pydub.AudioSegment \ .from_file(self.filename) \ .set_frame_rate(Config.frame_rate) \ .set_channels(Config.channels) \ .set_sample_width(Config.sample_width) \ .apply_gain(initial_db_gain) self.audio_segment_frame_width = self.audio_segment.frame_width self.sound_duration = self.audio_segment.duration_seconds except Exception as e: error_print("failed to load « {} »: {}".format(self.name, e)) self.loading_error = e self.fail() else: self.success() debug_print("Loaded « {} »".format(self.name)) def check_is_loaded(self): return self.state.startswith('loaded_') def is_not_stopped(self): return self.check_is_loaded() and not self.is_loaded_stopped() def is_paused(self): return self.is_loaded_paused() @property def sound_position(self): if self.is_not_stopped(): return self.current_frame / self.current_audio_segment.frame_rate else: return 0 def play(self, fade_in=0, volume=100, loop=0, start_at=0): self.set_gain(gain(volume) + self.mapping.master_gain, absolute=True) self.volume = volume self.current_loop = 0 self.loop_left = loop with self.music_lock: self.current_audio_segment = self.audio_segment self.current_frame = int(start_at * self.audio_segment.frame_rate) if fade_in > 0: db_gain = gain(self.volume, 0)[0] self.set_gain(-db_gain) self.gain_effects.append(GainEffect( "fade", self.current_audio_segment, self.current_loop, self.sound_position, self.sound_position + fade_in, gain=db_gain)) self.start_playing() def on_enter_loaded_playing(self): self.mixer.add_file(self) def finished_callback(self): if self.is_loaded_playing(): self.stop_playing() if self.is_loaded_stopping(): self.stopped() def trigger_stopped_events(self): self.mixer.remove_file(self) self.wait_event.set() def play_callback(self, out_data_length, frame_count): if self.is_loaded_paused(): return b'\0' * out_data_length with self.music_lock: [data, nb_frames] = self.get_next_sample(frame_count) if nb_frames < frame_count: if self.is_loaded_playing() and self.loop_left != 0: self.loop_left -= 1 self.current_loop += 1 self.current_frame = 0 [new_data, new_nb_frames] = self.get_next_sample( frame_count - nb_frames) data += new_data nb_frames += new_nb_frames elif nb_frames == 0: # FIXME: too slow threading.Thread( name="MSFinishedCallback", target=self.finished_callback).start() return data.ljust(out_data_length, b'\0') def get_next_sample(self, frame_count): fw = self.audio_segment_frame_width data = b"" nb_frames = 0 segment = self.current_audio_segment max_val = int(segment.frame_count()) start_i = max(self.current_frame, 0) end_i = min(self.current_frame + frame_count, max_val) data += segment._data[start_i*fw : end_i*fw] nb_frames += end_i - start_i self.current_frame += end_i - start_i volume_factor = self.volume_factor(self.effects_next_gain(nb_frames)) data = audioop.mul(data, Config.sample_width, volume_factor) return [data, nb_frames] def seek(self, value=0, delta=False): # We don't want to do that while stopping if not (self.is_loaded_playing() or self.is_loaded_paused()): return with self.music_lock: self.abandon_all_effects() self.current_frame = max( 0, int(delta) * self.current_frame + int(value * self.audio_segment.frame_rate)) # FIXME: si on fait un seek + delta, adapter le "loop" def effects_next_gain(self, frame_count): db_gain = 0 for gain_effect in self.gain_effects: [new_gain, last_gain] = gain_effect.get_next_gain( self.current_frame, self.current_loop, frame_count) if last_gain: self.set_gain(new_gain) self.gain_effects.remove(gain_effect) else: db_gain += new_gain return db_gain def abandon_all_effects(self): db_gain = 0 for gain_effect in self.gain_effects: db_gain += gain_effect.get_last_gain() self.gain_effects = [] self.set_gain(db_gain) def stop(self, fade_out=0, wait=False): if self.is_loaded_playing(): ms = int(self.sound_position * 1000) ms_fo = max(1, int(fade_out * 1000)) new_audio_segment = self.current_audio_segment[: ms+ms_fo] \ .fade_out(ms_fo) with self.music_lock: self.current_audio_segment = new_audio_segment self.stop_playing() if wait: self.wait_end() else: self.stop_playing() self.stopped() def volume_factor(self, additional_gain): return 10 ** ( (self.db_gain + additional_gain) / 20) def set_gain(self, db_gain, absolute=False): if absolute: self.db_gain = db_gain else: self.db_gain += db_gain def set_volume(self, value, delta=False, fade=0): [db_gain, self.volume] = gain( value + int(delta) * self.volume, self.volume) if fade > 0: self.gain_effects.append(GainEffect( "fade", self.current_audio_segment, self.current_loop, self.sound_position, self.sound_position + fade, gain=db_gain)) else: self.set_gain(db_gain) def wait_end(self): self.wait_event.clear() self.wait_event.wait()