Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def _load_plugins(self):
self.log.debug("Loading plugins...")
# TODO: This should be cleaned up. Create a Plugins base class and
# classmethods to determine if the plugins should be used.
for plugin in Util.string_to_list(
self.config['mpf']['plugins']):
self.log.debug("Loading '%s' plugin", plugin)
plugin_obj = Util.string_to_class(plugin)(self)
self.plugins.append(plugin_obj)
def get_express_config(self, value):
"""Parse short config."""
return self.get_list_config(Util.string_to_list(value))
# Validate sound attributes and provide default values
if 'volume' in self.config:
self._volume = min(max(float(self.config['volume']), 0.0), 1.0)
if 'priority' in self.config:
self.priority = int(self.config['priority'])
if 'max_queue_time' in self.config and self.config['max_queue_time'] is not None:
self._max_queue_time = AudioInterface.string_to_secs(self.config['max_queue_time'])
if 'loops' in self.config:
self._loops = int(self.config['loops'])
if 'events_when_played' in self.config and isinstance(
self.config['events_when_played'], str):
self._events_when_played = Util.string_to_list(self.config['events_when_played'])
if 'events_when_stopped' in self.config and isinstance(
self.config['events_when_stopped'], str):
self._events_when_stopped = Util.string_to_list(self.config['events_when_stopped'])
if 'ducking' in self.config:
self._ducking = DuckingSettings(self.machine, self.config['ducking'])
# An attenuation value of exactly 1.0 does absolutely nothing so
# there is no point in keeping the ducking settings for this
# sound when attenuation is 1.0.
if self._ducking.attenuation == 1.0:
self._ducking = None
# Add sound to a dictionary of sound objects keyed by sound id
if not hasattr(self.machine, 'sounds_by_id'):
try:
key = leds[0]
except (TypeError, IndexError):
return False
for step in script:
if step.get('fade', None):
color = str(step['color']) + "-f" + str(step['time'])
else:
color = str(step['color'])
current_action = {'time': step['time']}
if lights:
current_action['lights'] = dict()
for light in Util.string_to_list(lights):
current_action['lights'][light] = color
if leds:
current_action['leds'] = dict()
for led in Util.string_to_list(leds):
current_action['leds'][led] = color
show_actions.append(current_action)
show = Show(machine=self.machine, name='Script', file=None,
config=None, steps=show_actions)
self.machine.show_controller.play_show(show=show, loops=loops,
callback=callback, **kwargs)
return show
self._scope = self.config['scope']
if 'crossfade_time' in self.config and self.config['crossfade_time'] is not None:
self._crossfade_time = Util.string_to_secs(self.config['crossfade_time'])
if 'events_when_played' in self.config and isinstance(
self.config['events_when_played'], str):
self._events_when_played = Util.string_to_list(self.config['events_when_played'])
if 'events_when_stopped' in self.config and isinstance(
self.config['events_when_stopped'], str):
self._events_when_stopped = Util.string_to_list(self.config['events_when_stopped'])
if 'events_when_sound_played' in self.config and isinstance(
self.config['events_when_sound_played'], str):
self._events_when_sound_played = Util.string_to_list(self.config['events_when_sound_played'])
if 'events_when_sound_stopped' in self.config and isinstance(
self.config['events_when_sound_stopped'], str):
self._events_when_sound_stopped = Util.string_to_list(self.config['events_when_sound_stopped'])
if 'sounds' not in self.config or self.config['sounds'] is None:
self.log.error("A playlist must contain at least one sound. "
"Could not create playlist '%s' asset.", name)
raise AudioException("A playlist must contain at least one sound. "
"Could not create playlist '{}' asset.".format(name))
# self.config['sounds'] contains a list of sounds to include in the playlist. Optionally,
# each item in the list can also include an integer weight value delimited by a pipe (|)
# character.
# Build list of weighted sound names
def __init__(self, machine: MachineController, name: str, player: Player, config: dict):
"""Initialise sequence."""
super().__init__(machine, name, player, config)
self.log = logging.getLogger('Sequence.' + name)
self.log.debug("Creating Sequence LogicBlock")
# split events for each step
self.config['events'][:] = [Util.string_to_list(x) for x in self.config['events']]
if not self.config['player_variable']:
self.config['player_variable'] = self.name + '_step'
if not self.config['persist_state']:
self.player[self.config['player_variable']] = 0
if item == 'item not in config!@#':
if default == 'default required!@#':
raise ValueError('Required setting missing from config file. '
'Run with verbose logging and look for the last '
'ConfigProcessor entry above this line to see where the '
'problem is. {} {}'.format(spec,
validation_failure_info))
else:
item = default
if item_type == 'single':
item = self.validate_item(item, validation,
validation_failure_info)
elif item_type == 'list':
item = Util.string_to_list(item)
new_list = list()
for i in item:
new_list.append(
self.validate_item(i, validation,
validation_failure_info))
item = new_list
elif item_type == 'set':
item = set(Util.string_to_list(item))
new_set = set()
for i in item:
def _expand_device(self, device):
"""Idempotently expand device if it is a placeholder."""
if not isinstance(device, str):
return [device]
device_or_tag_names = Util.string_to_list(device)
if not self.device_collection:
return device_or_tag_names
device_list = []
for device_name in device_or_tag_names:
try:
devices = self.device_collection.items_tagged(device_name)
if not devices:
device_list.append(self.device_collection[device_name])
else:
device_list.extend(devices)
except KeyError:
if not self.__class__.allow_placeholders_in_keys or "(" not in device_name:
# no placeholders
return self.raise_config_error(
def play(self, settings, context, calling_context, priority=0, **kwargs):
"""Flash flashers."""
del kwargs
for flasher, s in settings.items():
if isinstance(flasher, str):
flasher_names = Util.string_to_list(flasher)
for flasher_name in flasher_names:
self._flash(self.machine.lights[flasher_name],
duration_ms=s['ms'],
key=context)
else:
self._flash(flasher, duration_ms=s['ms'], key=context)
Args:
machine: The main ``MachineController`` instance.
config: Dictionary which contains configuration settings for the
OPC client.
"""
super().__init__(machine, config)
self.log = logging.getLogger('FadeCandyClient')
self.update_every_tick = True
self.config = self.machine.config_validator.validate_config('fadecandy',
self.machine.config['fadecandy'])
self.gamma = self.config['gamma']
self.whitepoint = Util.string_to_list(self.config['whitepoint'])
self.whitepoint[0] = float(self.whitepoint[0])
self.whitepoint[1] = float(self.whitepoint[1])
self.whitepoint[2] = float(self.whitepoint[2])
self.linear_slope = self.config['linear_slope']
self.linear_cutoff = self.config['linear_cutoff']
self.keyframe_interpolation = self.config['keyframe_interpolation']
self.dithering = self.config['dithering']
if not self.keyframe_interpolation:
self.update_every_tick = False