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
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
|
import threading
import pydub
import pygame
import math
import time
from transitions.extensions import HierarchicalMachine as Machine
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.flag_paused = False
threading.Thread(name = "MSMusicLoad", target = self.load, kwargs = {'lock': lock}).start()
def on_enter_loading(self, lock=None):
lock.acquire()
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()
def is_paused(self):
return self.flag_paused
@property
def sound_position(self):
if self.is_playing() and not self.is_paused():
return min(time.time() - self.started_at, self.sound_duration)
elif self.is_playing() and self.is_paused():
return min(self.paused_at - self.started_at, self.sound_duration)
else:
return 0
def play(self, fade_in = 0, volume = 100, start_at = 0):
self.set_volume(volume)
if start_at > 0:
raw_data_length = len(self.raw_data)
start_offset = int((raw_data_length / self.sound_duration) * start_at)
start_offset = start_offset - (start_offset % 2)
sound = pygame.mixer.Sound(self.raw_data[start_offset:])
else:
sound = pygame.mixer.Sound(self.raw_data)
self.started_at = time.time()
self.channel().play(sound, fade_ms = int(fade_in * 1000))
self.flag_paused = False
def pause(self):
self.paused_at = time.time()
self.channel().pause()
self.flag_paused = True
def unpause(self):
self.started_at += (time.time() - self.paused_at)
self.channel().unpause()
self.flag_paused = False
def stop(self, fade_out = 0):
if fade_out > 0:
self.channel().fadeout(int(fade_out * 1000))
else:
self.channel().stop()
def set_volume(self, value):
if value < 0:
value = 0
if value > 100:
value = 100
self.channel().set_volume(value / 100)
def wait_end(self):
pass
def channel(self):
return pygame.mixer.Channel(self.channel_id)
|
