file_lock = Lock("file")
class MusicFile:
+ STATES = [
+ 'initial',
+ 'loading',
+ 'failed',
+ {
+ 'name': 'loaded',
+ 'children': [
+ 'playing',
+ 'paused',
+ 'stopping'
+ ]
+ }
+ ]
+ TRANSITIONS = [
+ {
+ 'trigger': 'load',
+ 'source': 'initial',
+ 'dest': 'loading'
+ },
+ {
+ 'trigger': 'fail',
+ 'source': 'loading',
+ 'dest': 'failed'
+ },
+ {
+ 'trigger': 'success',
+ 'source': 'loading',
+ 'dest': 'loaded'
+ },
+ {
+ 'trigger': 'start_playing',
+ 'source': 'loaded',
+ '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': '*',
+ 'dest': 'loaded',
+ 'before': 'trigger_stopped_events'
+ }
+ ]
+
def __init__(self, filename, mapping, name=None, gain=1):
- states = [
- 'initial',
- 'loading',
- 'failed',
- {
- 'name': 'loaded',
- 'children': [
- 'stopped',
- 'playing',
- 'paused',
- 'stopping',
- 'stopped'
- ]
- }
- ]
- 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(model=self, states=states,
- transitions=transitions, initial='initial',
+ Machine(model=self, states=self.STATES,
+ transitions=self.TRANSITIONS, initial='initial',
ignore_invalid_triggers=True)
- 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()
+ # Machine related events
def on_enter_loading(self):
with file_lock:
try:
.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.success()
debug_print("Loaded « {} »".format(self.name))
- def check_is_loaded(self):
- return self.state.startswith('loaded_')
+ def on_enter_loaded(self):
+ self.gain_effects = []
+ self.set_gain(0, absolute=True)
+ self.current_audio_segment = None
+ self.volume = 100
+ self.wait_event = threading.Event()
+ self.current_loop = 0
+
+ def on_enter_loaded_playing(self):
+ self.mixer.add_file(self)
- def is_not_stopped(self):
- return self.check_is_loaded() and not self.is_loaded_stopped()
+ # Machine related states
+ def is_in_use(self):
+ return self.is_loaded(allow_substates=True) and not self.is_loaded()
- def is_paused(self):
- return self.is_loaded_paused()
+ def is_in_use_not_stopping(self):
+ return self.is_loaded_playing() or self.is_loaded_paused()
+ # Machine related triggers
+ def trigger_stopped_events(self):
+ self.mixer.remove_file(self)
+ self.wait_event.set()
+
+ # Actions and properties called externally
@property
def sound_position(self):
- if self.is_not_stopped():
+ if self.is_in_use():
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):
- # FIXME: create a "reinitialize" method
- self.gain_effects = []
self.set_gain(gain(volume) + self.mapping.master_gain, absolute=True)
self.volume = volume
- self.current_loop = 0
if loop < 0:
self.last_loop = float('inf')
else:
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)
+ if fade_in > 0:
+ db_gain = gain(self.volume, 0)[0]
+ self.set_gain(-db_gain)
+ self.add_fade_effect(db_gain, fade_in)
- def finished_callback(self):
+ def seek(self, value=0, delta=False):
+ if not self.is_in_use_not_stopping():
+ return
+
+ with self.music_lock:
+ self.abandon_all_effects()
+ if delta:
+ frame_count = int(self.audio_segment.frame_count())
+ frame_diff = int(value * self.audio_segment.frame_rate)
+ self.current_frame += frame_diff
+ while self.current_frame < 0:
+ self.current_loop -= 1
+ self.current_frame += frame_count
+ while self.current_frame > frame_count:
+ self.current_loop += 1
+ self.current_frame -= frame_count
+ if self.current_loop < 0:
+ self.current_loop = 0
+ self.current_frame = 0
+ if self.current_loop > self.last_loop:
+ self.current_loop = self.last_loop
+ self.current_frame = frame_count
+ else:
+ self.current_frame = max(
+ 0,
+ int(value * self.audio_segment.frame_rate))
+
+ def stop(self, fade_out=0, wait=False, set_wait_id=None):
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 self.is_loaded_stopping():
+ if wait:
+ if set_wait_id is not None:
+ self.mapping.add_wait_id(set_wait_id, self.wait_event)
+ self.wait_end()
+ else:
self.stopped()
- def trigger_stopped_events(self):
- self.mixer.remove_file(self)
- self.wait_event.set()
+ 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 set_volume(self, value, delta=False, fade=0):
+ [db_gain, self.volume] = gain(
+ value + int(delta) * self.volume,
+ self.volume)
+
+ self.set_gain_with_effect(db_gain, fade=fade)
+
+ def set_gain_with_effect(self, db_gain, fade=0):
+ if not self.is_in_use():
+ return
+
+ if fade > 0:
+ self.add_fade_effect(db_gain, fade)
+ else:
+ self.set_gain(db_gain)
+
+ def wait_end(self):
+ self.wait_event.clear()
+ self.wait_event.wait()
+
+ # Callbacks
+ def finished_callback(self):
+ self.stopped()
def play_callback(self, out_data_length, frame_count):
if self.is_loaded_paused():
return data.ljust(out_data_length, b'\0')
+ # Helpers
+ def set_gain(self, db_gain, absolute=False):
+ if absolute:
+ self.db_gain = db_gain
+ else:
+ self.db_gain += db_gain
+
def get_next_sample(self, frame_count):
- fw = self.audio_segment_frame_width
+ fw = self.audio_segment.frame_width
data = b""
nb_frames = 0
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()):
+ def add_fade_effect(self, db_gain, fade_duration):
+ if not self.is_in_use():
return
- with self.music_lock:
- self.abandon_all_effects()
- if delta:
- frame_count = int(self.audio_segment.frame_count())
- frame_diff = int(value * self.audio_segment.frame_rate)
- self.current_frame += frame_diff
- while self.current_frame < 0:
- self.current_loop -= 1
- self.current_frame += frame_count
- while self.current_frame > frame_count:
- self.current_loop += 1
- self.current_frame -= frame_count
- if self.current_loop < 0:
- self.current_loop = 0
- self.current_frame = 0
- if self.current_loop > self.last_loop:
- self.current_loop = self.last_loop
- self.current_frame = frame_count
- else:
- self.current_frame = max(
- 0,
- int(value * self.audio_segment.frame_rate))
+
+ self.gain_effects.append(GainEffect(
+ "fade",
+ self.current_audio_segment,
+ self.current_loop,
+ self.sound_position,
+ self.sound_position + fade_duration,
+ gain=db_gain))
def effects_next_gain(self, frame_count):
db_gain = 0
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, set_wait_id=None):
- 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:
- if set_wait_id is not None:
- self.mapping.add_wait_id(set_wait_id, self.wait_event)
- self.wait_end()
- else:
- self.stop_playing()
- self.stopped()
-
- def volume_factor(self, additional_gain):
+ def volume_factor(self, additional_gain=0):
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 not (self.is_loaded_playing() or self.is_loaded_paused()):
- return
-
- 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()
-
projects / perso / Immae / Projets / Python / MusicSampler.git / blobdiff