import pygame
import math
import time
+from transitions.extensions import HierarchicalMachine as Machine
-class MusicFile:
+class MusicFile(Machine):
def __init__(self, filename, lock, channel_id, name = None, gain = 1):
+ states = [
+ 'initial',
+ 'loading',
+ 'failed',
+ { 'name': 'loaded', 'children': ['stopped', 'playing', 'paused'] }
+ ]
+ transitions = [
+ { 'trigger': 'load', 'source': 'initial', 'dest': 'loading'},
+ { 'trigger': 'fail', 'source': 'loading', 'dest': 'failed'},
+ { 'trigger': 'success', 'source': 'loading', 'dest': 'loaded_stopped'},
+ #{ 'trigger': 'play', 'source': 'loaded_stopped', 'dest': 'loaded_playing'},
+ #{ 'trigger': 'pause', 'source': 'loaded_playing', 'dest': 'loaded_paused'},
+ #{ 'trigger': 'stop', 'source': ['loaded_playing','loaded_paused'], 'dest': 'loaded_stopped'}
+ ]
+
+ Machine.__init__(self, states=states, transitions=transitions, initial='initial')
+
self.filename = filename
self.channel_id = channel_id
self.name = name or filename
self.raw_data = None
self.gain = gain
- self.loaded = False
self.flag_paused = False
- threading.Thread(name = "MSMusicLoad", target = self.load_sound, args = [lock]).start()
+ threading.Thread(name = "MSMusicLoad", target = self.load, kwargs = {'lock': lock}).start()
- def load_sound(self, lock):
+ def on_enter_loading(self, lock=None):
lock.acquire()
- print("Loading « {} »".format(self.name))
- volume_factor = 20 * math.log10(self.gain)
- audio_segment = pydub.AudioSegment.from_file(self.filename).set_frame_rate(44100).apply_gain(volume_factor)
- self.sound_duration = audio_segment.duration_seconds
- self.raw_data = audio_segment.raw_data
- print("Loaded « {} »".format(self.name))
- self.loaded = True
- lock.release()
+ try:
+ print("Loading « {} »".format(self.name))
+ volume_factor = 20 * math.log10(self.gain)
+ audio_segment = pydub.AudioSegment.from_file(self.filename).set_frame_rate(44100).apply_gain(volume_factor)
+ self.sound_duration = audio_segment.duration_seconds
+ self.raw_data = audio_segment.raw_data
+ except Exception as e:
+ print("failed to load « {} »: {}".format(self.name, e))
+ self.loading_error = e
+ self.fail()
+ else:
+ self.success()
+ print("Loaded « {} »".format(self.name))
+ finally:
+ lock.release()
+
+ def check_is_loaded(self):
+ return self.state.startswith('loaded_')
def is_playing(self):
return self.channel().get_busy()
return 0
def play(self, fade_in = 0, volume = 100, start_at = 0):
- self.channel().set_endevent()
- self.channel().set_endevent(pygame.USEREVENT)
self.set_volume(volume)
if start_at > 0: