def command(self, command = "", **kwargs):
pass
- def pause(self, music = None, **kwargs):
+ def music_list(self, music):
if music is not None:
- music.pause()
+ return [music]
else:
- for music in self.key.parent.open_files.values():
- if music.is_playing() and not music.is_paused():
- music.pause()
+ return self.key.parent.open_files.values()
+
+ def pause(self, music = None, **kwargs):
+ for music in self.music_list(music):
+ if music.is_loaded_playing():
+ music.pause()
def unpause(self, music = None, **kwargs):
- if music is not None:
- music.unpause()
- else:
- for music in self.key.parent.open_files.values():
- if music.is_playing() and music.is_paused():
- music.unpause()
+ for music in self.music_list(music):
+ if music.is_loaded_paused():
+ music.unpause()
def play(self, music = None, fade_in = 0, start_at = 0,
restart_if_running = False, volume = 100, **kwargs):
if music is not None:
if restart_if_running:
- if music.is_playing():
+ if music.is_not_stopped():
music.stop()
music.play(volume = volume, fade_in = fade_in, start_at = start_at)
else:
- if not music.is_playing():
+ if not music.is_not_stopped():
music.play(volume = volume, fade_in = fade_in, start_at = start_at)
- else:
- pygame.mixer.unpause()
def stop(self, music = None, fade_out = 0, **kwargs):
- if music is not None:
- music.stop(fade_out = fade_out)
- else:
- if fade_out > 0:
- pygame.mixer.fadeout(int(fade_out * 1000))
- else:
- pygame.mixer.stop()
+ for music in self.music_list(music):
+ if music.is_loaded_paused() or music.is_loaded_playing():
+ music.stop(fade_out = fade_out)
def stop_all_actions(self, **kwargs):
self.key.parent.stop_all_running()
if music is not None:
music.set_volume(value)
else:
+ # FIXME: todo
pass
def wait(self, duration = 0, music = None, **kwargs):
from kivy.clock import Clock
import threading
-import pygame
import yaml
import sys
-from .lock import *
from .music_file import *
from . import yml_file
ready_color = ListProperty([1, 165/255, 0, 1])
def __init__(self, **kwargs):
- self.key_config, self.channel_number, self.open_files = self.parse_config()
+ self.key_config, self.open_files = self.parse_config()
super(Mapping, self).__init__(**kwargs)
self._keyboard = Window.request_keyboard(self._keyboard_closed, self)
self._keyboard.bind(on_key_down=self._on_keyboard_down)
Clock.schedule_interval(self.not_all_keys_ready, 1)
- pygame.mixer.init(frequency = 44100)
- pygame.mixer.set_num_channels(self.channel_number)
-
def _keyboard_closed(self):
self._keyboard.unbind(on_key_down=self._on_keyboard_down)
self._keyboard = None
continue
thread.join()
- pygame.quit()
sys.exit()
return True
aliases = config['aliases']
seen_files = {}
- file_lock = Lock("file")
-
- channel_id = 0
-
key_properties = {}
for key in config['key_properties']:
if filename in config['music_properties']:
seen_files[filename] = MusicFile(
filename,
- file_lock,
- channel_id,
**config['music_properties'][filename])
else:
seen_files[filename] = MusicFile(
- filename,
- file_lock,
- channel_id)
- channel_id = channel_id + 1
+ filename)
if filename not in key_properties[mapped_key]['files']:
key_properties[mapped_key]['files'].append(seen_files[filename])
key_properties[mapped_key]['actions'].append([action_name, action_args])
- return (key_properties, channel_id + 1, seen_files)
+ return (key_properties, seen_files)
import threading
import pydub
-import pygame
import math
import time
from transitions.extensions import HierarchicalMachine as Machine
+import pyaudio as pa
+import sounddevice as sd
+import os.path
+
+from .lock import Lock
+file_lock = Lock("file")
+
+pyaudio = pa.PyAudio()
+
class MusicFile(Machine):
- def __init__(self, filename, lock, channel_id, name = None, gain = 1):
+ def __init__(self, filename, name = None, gain = 1):
states = [
'initial',
'loading',
'failed',
- { 'name': 'loaded', 'children': ['stopped', 'playing', 'paused'] }
+ { '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': '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'}
+ { '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'}
]
Machine.__init__(self, states=states, transitions=transitions, initial='initial')
self.filename = filename
- self.channel_id = channel_id
+ self.stream = None
self.name = name or filename
- self.raw_data = None
+ self.audio_segment = None
self.gain = gain
+ self.music_lock = Lock("music__" + filename)
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()
+ threading.Thread(name = "MSMusicLoad", target = self.load).start()
+
+ def on_enter_loading(self):
+ with file_lock:
+ try:
+ print("Loading « {} »".format(self.name))
+ volume_factor = 20 * math.log10(self.gain)
+ self.audio_segment = pydub.AudioSegment.from_file(self.filename).set_frame_rate(44100).apply_gain(volume_factor)
+ self.sound_duration = self.audio_segment.duration_seconds
+ 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))
def check_is_loaded(self):
return self.state.startswith('loaded_')
- def is_playing(self):
- return self.channel().get_busy()
+ def is_not_stopped(self):
+ return self.check_is_loaded() and not self.is_loaded_stopped()
def is_paused(self):
- return self.flag_paused
+ return self.is_loaded_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)
+ 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, 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
+ self.db_gain = self.volume_to_gain(volume)
+ ms = int(start_at * 1000)
+ ms_fi = max(1, int(fade_in * 1000))
+ with self.music_lock:
+ self.current_audio_segment = (self.audio_segment + self.db_gain).fade(from_gain=-120, duration=ms_fi, start=ms)
+ self.before_loaded_playing(initial_frame = int(start_at * self.audio_segment.frame_rate))
+ self.start_playing()
+
+ def before_loaded_playing(self, initial_frame = 0):
+ self.current_frame = initial_frame
+ with self.music_lock:
+ segment = self.current_audio_segment
+
+ self.stream = sd.RawOutputStream(samplerate=segment.frame_rate,
+ channels=segment.channels,
+ dtype='int' + str(8*segment.sample_width), # FIXME: ?
+ latency=1.,
+ callback=self.play_callback,
+ finished_callback=self.finished_callback
+ )
+
+ def on_enter_loaded_playing(self):
+ self.stream.start()
+
+ def on_enter_loaded_paused(self):
+ self.stream.stop()
+
+ def finished_callback(self):
+ if self.is_loaded_playing():
+ self.stop_playing()
+ if self.is_loaded_stopping():
+ self.stopped()
+
+ def play_callback(self, out_data, frame_count, time_info, status_flags):
+ with self.music_lock:
+ audio_segment = self.current_audio_segment.get_sample_slice_data(
+ start_sample=self.current_frame,
+ end_sample=self.current_frame + frame_count
+ )
+ self.current_frame += frame_count
+ if len(audio_segment) == 0:
+ raise sd.CallbackStop
+
+ out_data[:] = audio_segment.ljust(len(out_data), b'\0')
def stop(self, fade_out = 0):
- if fade_out > 0:
- self.channel().fadeout(int(fade_out * 1000))
+ if self.is_loaded_playing():
+ ms = int(self.sound_position * 1000)
+ ms_fo = max(1, int(fade_out * 1000))
+
+ with self.music_lock:
+ self.current_audio_segment = self.current_audio_segment[:ms + ms_fo].fade_out(ms_fo)
+ self.stop_playing()
else:
- self.channel().stop()
+ self.stop_playing()
+ self.stopped()
def set_volume(self, value):
- if value < 0:
- value = 0
- if value > 100:
- value = 100
- self.channel().set_volume(value / 100)
+ if self.is_loaded_stopped():
+ return
+
+ db_gain = self.volume_to_gain(value)
+ new_audio_segment = self.current_audio_segment + (db_gain - self.db_gain)
+ self.db_gain = db_gain
+ with self.music_lock:
+ self.current_audio_segment = new_audio_segment
+
+ def volume_to_gain(self, volume):
+ return 20 * math.log10(max(volume, 0.0001) / 100)
def wait_end(self):
+ # FIXME: todo
pass
- def channel(self):
- return pygame.mixer.Channel(self.channel_id)
+# Add some more functions to AudioSegments
+def get_sample_slice_data(self, start_sample=0, end_sample=float('inf')):
+ max_val = int(self.frame_count())
+
+ start_i = max(start_sample, 0) * self.frame_width
+ end_i = min(end_sample, max_val) * self.frame_width
+
+ return self._data[start_i:end_i]
+
+pydub.AudioSegment.get_sample_slice_data = get_sample_slice_data
projects / perso / Immae / Projets / Python / MusicSampler.git / commitdiff
