from kivy.uix.relativelayout import RelativeLayout
-from kivy.properties import NumericProperty, ListProperty
+from kivy.properties import NumericProperty, ListProperty, StringProperty
from kivy.core.window import Window
from kivy.clock import Clock
import threading
import yaml
import sys
+from collections import defaultdict
-from .music_file import *
+from transitions.extensions import HierarchicalMachine as Machine
+
+from .music_file import MusicFile
from .mixer import Mixer
-from . import Config, gain, error_print
-from .music_effect import GainEffect
+from . import Config, gain, error_print, warn_print
+from .action import Action
class Mapping(RelativeLayout):
- expected_keys = NumericProperty(0)
+ STATES = [
+ 'initial',
+ 'configuring',
+ 'configured',
+ 'loading',
+ 'loaded',
+ 'failed'
+ ]
+
+ TRANSITIONS = [
+ {
+ 'trigger': 'configure',
+ 'source': 'initial',
+ 'dest': 'configuring'
+ },
+ {
+ 'trigger': 'fail',
+ 'source': 'configuring',
+ 'dest': 'failed'
+ },
+ {
+ 'trigger': 'success',
+ 'source': 'configuring',
+ 'dest': 'configured',
+ 'after': 'load'
+ },
+ {
+ 'trigger': 'load',
+ 'source': 'configured',
+ 'dest': 'loading'
+ },
+ {
+ 'trigger': 'fail',
+ 'source': 'loading',
+ 'dest': 'failed'
+ },
+ {
+ 'trigger': 'success',
+ 'source': 'loading',
+ 'dest': 'loaded'
+ },
+ {
+ 'trigger': 'reload',
+ 'source': 'loaded',
+ 'dest': 'configuring'
+ }
+ ]
+
master_volume = NumericProperty(100)
ready_color = ListProperty([1, 165/255, 0, 1])
+ state = StringProperty("")
def __init__(self, **kwargs):
+ self.keys = []
+ self.running = []
+ self.wait_ids = {}
+ self.open_files = {}
+
+ Machine(model=self, states=self.STATES,
+ transitions=self.TRANSITIONS, initial='initial',
+ ignore_invalid_triggers=True, queued=True)
+ super(Mapping, self).__init__(**kwargs)
+ self.keyboard = Window.request_keyboard(self.on_keyboard_closed, self)
+ self.keyboard.bind(on_key_down=self.on_keyboard_down)
+
+ self.configure()
+
+ def on_enter_configuring(self):
if Config.builtin_mixing:
self.mixer = Mixer()
else:
self.mixer = None
- 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)
- self.running = []
- Clock.schedule_interval(self.not_all_keys_ready, 1)
- @property
- def master_gain(self):
- return gain(self.master_volume)
+ try:
+ self.key_config, self.open_files = self.parse_config()
+ except Exception as e:
+ error_print("Error while loading configuration: {}".format(e),
+ with_trace=True)
+ sys.exit()
+ else:
+ self.success()
- def set_master_volume(self, value, delta=False, fade=0):
- [db_gain, self.master_volume] = gain(
- value + int(delta) * self.master_volume,
- self.master_volume)
+ def on_enter_loading(self):
+ for key in self.keys:
+ key.reload()
+ self.success()
- for music in self.open_files.values():
- if not (music.is_loaded_playing() or music.is_loaded_paused()):
- continue
+ # Kivy events
+ def add_widget(self, widget, index=0):
+ if type(widget).__name__ == "Key" and widget not in self.keys:
+ self.keys.append(widget)
+ return super(Mapping, self).add_widget(widget, index)
- if fade > 0:
- music.gain_effects.append(GainEffect(
- "fade",
- music.current_audio_segment,
- music.current_loop,
- music.sound_position,
- music.sound_position + fade,
- gain=db_gain))
- else:
- music.set_gain(db_gain)
+ def remove_widget(self, widget, index=0):
+ if type(widget).__name__ == "Key" and widget in self.keys:
+ self.keys.remove(widget)
+ return super(Mapping, self).remove_widget(widget, index)
- def _keyboard_closed(self):
- self._keyboard.unbind(on_key_down=self._on_keyboard_down)
- self._keyboard = None
+ def on_keyboard_closed(self):
+ self.keyboard.unbind(on_key_down=self.on_keyboard_down)
+ self.keyboard = None
- def _on_keyboard_down(self, keyboard, keycode, text, modifiers):
+ def on_keyboard_down(self, keyboard, keycode, text, modifiers):
key = self.find_by_key_code(keycode)
- if len(modifiers) == 0 and key is not None:
- threading.Thread(name="MSKeyAction", target=key.do_actions).start()
+ if self.allowed_modifiers(modifiers) and key is not None:
+ modifiers.sort()
+ threading.Thread(name="MSKeyAction", target=key.run,
+ args=['-'.join(modifiers)]).start()
elif 'ctrl' in modifiers and (keycode[0] == 113 or keycode[0] == '99'):
+ self.stop_all_running()
for thread in threading.enumerate():
if thread.getName()[0:2] != "MS":
continue
thread.join()
sys.exit()
+ elif 'ctrl' in modifiers and keycode[0] == 114:
+ threading.Thread(name="MSReload", target=self.reload).start()
return True
+ # Helpers
+ def allowed_modifiers(self, modifiers):
+ allowed = []
+ return len([a for a in modifiers if a not in allowed]) == 0
+
def find_by_key_code(self, key_code):
if "Key_" + str(key_code[0]) in self.ids:
return self.ids["Key_" + str(key_code[0])]
return None
- def not_all_keys_ready(self, dt):
- for key in self.children:
- if not type(key).__name__ == "Key":
- continue
- if not key.is_key_ready:
- return True
- self.ready_color = [0, 1, 0, 1]
- return False
+ def all_keys_ready(self):
+ partial = False
+ for key in self.keys:
+ if not key.is_loaded_or_failed():
+ return "not_ready"
+ partial = partial or key.is_failed()
+ if partial:
+ return "partial"
+ else:
+ return "success"
+
+ # Callbacks
+ def key_loaded_callback(self):
+ result = self.all_keys_ready()
+ if result == "success":
+ self.ready_color = [0, 1, 0, 1]
+ elif result == "partial":
+ self.ready_color = [1, 0, 0, 1]
+ else:
+ self.ready_color = [1, 165/255, 0, 1]
+
+ ## Some global actions
def stop_all_running(self):
running = self.running
self.running = []
for (key, start_time) in running:
- key.interrupt_action()
+ key.interrupt()
+
+ # Master volume methods
+ @property
+ def master_gain(self):
+ return gain(self.master_volume)
+ def set_master_volume(self, value, delta=False, fade=0):
+ [db_gain, self.master_volume] = gain(
+ value + int(delta) * self.master_volume,
+ self.master_volume)
+
+ for music in self.open_files.values():
+ music.set_gain_with_effect(db_gain, fade=fade)
+
+ # Wait handler methods
+ def add_wait_id(self, wait_id, action_or_wait):
+ self.wait_ids[wait_id] = action_or_wait
+
+ def interrupt_wait(self, wait_id):
+ if wait_id in self.wait_ids:
+ action_or_wait = self.wait_ids[wait_id]
+ del(self.wait_ids[wait_id])
+ if isinstance(action_or_wait, Action):
+ action_or_wait.interrupt()
+ else:
+ action_or_wait.set()
+
+ # Methods to control running keys
def start_running(self, key, start_time):
self.running.append((key, start_time))
if (key, start_time) in self.running:
self.running.remove((key, start_time))
+ # YML config parser
def parse_config(self):
+ def update_alias(prop_hash, aliases, key):
+ if isinstance(aliases[key], dict):
+ prop_hash.update(aliases[key], **prop_hash)
+ else:
+ warn_print("Alias {} is not a hash, ignored".format(key))
+
+ def include_aliases(prop_hash, aliases):
+ if 'include' not in prop_hash:
+ return
+
+ included = prop_hash['include']
+ del(prop_hash['include'])
+ if isinstance(included, str):
+ update_alias(prop_hash, aliases, included)
+ elif isinstance(included, list):
+ for included_ in included:
+ if isinstance(included_, str):
+ update_alias(prop_hash, aliases, included_)
+ else:
+ warn_print("Unkown alias include type, ignored: "
+ "{} in {}".format(included_, included))
+ else:
+ warn_print("Unkown alias include type, ignored: {}"
+ .format(included))
+
+ def check_key_property(key_property, key):
+ if 'description' in key_property:
+ desc = key_property['description']
+ if not isinstance(desc, list):
+ warn_print("description in key_property '{}' is not "
+ "a list, ignored".format(key))
+ del(key_property['description'])
+ if 'color' in key_property:
+ color = key_property['color']
+ if not isinstance(color, list)\
+ or len(color) != 3\
+ or not all(isinstance(item, int) for item in color)\
+ or any(item < 0 or item > 255 for item in color):
+ warn_print("color in key_property '{}' is not "
+ "a list of 3 valid integers, ignored".format(key))
+ del(key_property['color'])
+
+ def check_key_properties(config):
+ if 'key_properties' in config:
+ if isinstance(config['key_properties'], dict):
+ return config['key_properties']
+ else:
+ warn_print("key_properties config is not a hash, ignored")
+ return {}
+ else:
+ return {}
+
+ def check_mapped_keys(config):
+ if 'keys' in config:
+ if isinstance(config['keys'], dict):
+ return config['keys']
+ else:
+ warn_print("keys config is not a hash, ignored")
+ return {}
+ else:
+ return {}
+
+ def check_mapped_key(mapped_keys, key):
+ if not isinstance(mapped_keys[key], list):
+ warn_print("key config '{}' is not an array, ignored"
+ .format(key))
+ return []
+ else:
+ return mapped_keys[key]
+
+ def check_music_property(music_property, filename):
+ if not isinstance(music_property, dict):
+ warn_print("music_property config '{}' is not a hash, ignored"
+ .format(filename))
+ return {}
+ if 'name' in music_property:
+ music_property['name'] = str(music_property['name'])
+ if 'gain' in music_property:
+ try:
+ music_property['gain'] = float(music_property['gain'])
+ except ValueError as e:
+ del(music_property['gain'])
+ warn_print("gain for music_property '{}' is not "
+ "a float, ignored".format(filename))
+ return music_property
+
stream = open(Config.yml_file, "r")
try:
- config = yaml.load(stream)
+ config = yaml.safe_load(stream)
except Exception as e:
error_print("Error while loading config file: {}".format(e))
sys.exit()
stream.close()
- aliases = config['aliases']
+ if not isinstance(config, dict):
+ raise Exception("Top level config is supposed to be a hash")
+
+ if 'aliases' in config and isinstance(config['aliases'], dict):
+ aliases = config['aliases']
+ else:
+ aliases = defaultdict(dict)
+ if 'aliases' in config:
+ warn_print("aliases config is not a hash, ignored")
+
+ music_properties = defaultdict(dict)
+ if 'music_properties' in config and\
+ isinstance(config['music_properties'], dict):
+ music_properties.update(config['music_properties'])
+ elif 'music_properties' in config:
+ warn_print("music_properties config is not a hash, ignored")
+
seen_files = {}
- key_properties = {}
+ key_properties = defaultdict(lambda: {
+ "actions": [],
+ "properties": {},
+ "files": []
+ })
- for key in config['key_properties']:
- if key not in key_properties:
- key_prop = config['key_properties'][key]
- if 'include' in key_prop:
- included = key_prop['include']
- del(key_prop['include'])
+ for key in check_key_properties(config):
+ key_prop = config['key_properties'][key]
+
+ if not isinstance(key_prop, dict):
+ warn_print("key_property '{}' is not a hash, ignored"
+ .format(key))
+ continue
+
+ include_aliases(key_prop, aliases)
+ check_key_property(key_prop, key)
+
+ key_properties[key]["properties"] = key_prop
+
+ for mapped_key in check_mapped_keys(config):
+ for index, action in enumerate(check_mapped_key(
+ config['keys'], mapped_key)):
+ if not isinstance(action, dict) or\
+ not len(action) == 1 or\
+ not isinstance(list(action.values())[0] or {}, dict):
+ warn_print("action number {} of key '{}' is invalid, "
+ "ignored".format(index + 1, mapped_key))
+ continue
- if isinstance(included, str):
- key_prop.update(aliases[included], **key_prop)
- else:
- for included_ in included:
- key_prop.update(aliases[included_], **key_prop)
-
- key_properties[key] = {
- "actions": [],
- "properties": key_prop,
- "files": []
- }
-
- for mapped_key in config['keys']:
- if mapped_key not in key_properties:
- key_properties[mapped_key] = {
- "actions": [],
- "properties": {},
- "files": []
- }
- for action in config['keys'][mapped_key]:
action_name = list(action)[0]
action_args = {}
if action[action_name] is None:
- action[action_name] = []
+ action[action_name] = {}
- if 'include' in action[action_name]:
- included = action[action_name]['include']
- del(action[action_name]['include'])
-
- if isinstance(included, str):
- action[action_name].update(
- aliases[included],
- **action[action_name])
- else:
- for included_ in included:
- action[action_name].update(
- aliases[included_],
- **action[action_name])
+ include_aliases(action[action_name], aliases)
for argument in action[action_name]:
if argument == 'file':
- filename = action[action_name]['file']
+ filename = str(action[action_name]['file'])
if filename not in seen_files:
- if filename in config['music_properties']:
- seen_files[filename] = MusicFile(
- filename,
- self,
- **config['music_properties'][filename])
+ music_property = check_music_property(
+ music_properties[filename],
+ filename)
+
+ if filename in self.open_files:
+ self.open_files[filename]\
+ .reload_properties(**music_property)
+
+ seen_files[filename] = self.open_files[filename]
else:
seen_files[filename] = MusicFile(
- self,
- filename)
+ filename, self, **music_property)
if filename not in key_properties[mapped_key]['files']:
key_properties[mapped_key]['files'] \
.append(seen_files[filename])
action_args['music'] = seen_files[filename]
-
else:
action_args[argument] = action[action_name][argument]
projects / perso / Immae / Projets / Python / MusicSampler.git / blobdiff