Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def process_config2(self, config_spec, source=None, section_name=None,
target=None, result_type='dict', base_spec=None,
add_missing_keys=True):
# config_spec, str i.e. "device:shot"
# source is dict
# section_name is str used for logging failures
if not self.config_spec:
self.load_config_spec()
if source is None:
source = CaseInsensitiveDict()
if not section_name:
section_name = config_spec # str
validation_failure_info = (config_spec, section_name)
orig_spec = config_spec # str
# build up the actual config spec we're going to use
this_spec = self.config_spec
config_spec = config_spec.split(':')
for i in range(len(config_spec)):
this_spec = this_spec[config_spec[i]]
if not isinstance(this_spec, dict):
this_spec = dict()
self.log.warning("Received request to set machine_var '%s', but "
"that is not a valid machine_var.", name)
return
prev_value = self.machine_vars[name]['value']
self.machine_vars[name]['value'] = value
try:
change = value - prev_value
except TypeError:
change = prev_value != value
if change or force_events:
if self.machine_vars[name]['persist'] and self.config['mpf']['save_machine_vars_to_disk']:
disk_var = CaseInsensitiveDict()
disk_var['value'] = value
if self.machine_vars[name]['expire_secs']:
disk_var['expire'] = self.clock.get_time() + self.machine_vars[name]['expire_secs']
self.machine_var_data_manager.save_key(name, disk_var)
self.log.debug("Setting machine_var '%s' to: %s, (prior: %s, "
"change: %s)", name, value, prev_value,
change)
self.events.post('machine_var_' + name,
value=value,
prev_value=prev_value,
change=change)
'''event: machine_var_(name)
def load_machine_config(config_file_list, machine_path,
config_path='config', existing_config=None):
for num, config_file in enumerate(config_file_list):
if not existing_config:
machine_config = CaseInsensitiveDict()
else:
machine_config = existing_config
if not (config_file.startswith('/') or
config_file.startswith('\\')):
config_file = os.path.join(machine_path, config_path,
config_file)
logging.info("Machine config file #%s: %s", num + 1, config_file)
machine_config = Util.dict_merge(machine_config,
Config.load_config_file(
config_file))
return machine_config
self.log.info("Starting clock at %sHz", Clock._max_fps)
self._boot_holds = set()
self.mpf_path = os.path.dirname(mpf.__file__)
self.modes = CaseInsensitiveDict()
self.player_list = list()
self.player = None
self.num_players = 0
self.bcp_client_connected = False
self.animation_configs = dict()
self.active_slides = dict()
self.scriptlets = list()
self.register_boot_hold('init')
self.displays = CaseInsensitiveDict()
self.machine_vars = CaseInsensitiveDict()
self.machine_var_monitor = False
self.monitors = dict()
self.targets = dict()
"""Dict which contains all the active slide frames in the machine that
a slide can target. Will always contain an entry called 'default'
which will be used if a slide doesn't specify targeting.
"""
self.keyboard = None
self.physical_dmds = []
self.physical_rgb_dmds = []
self.crash_queue = queue.Queue()
self.ticks = 0
self.start_time = 0
self.is_init_done = False
def create_machine_var(self, name, value=0, persist=False, expire_secs=None, silent=False):
"""Create a new machine variable.
Args:
name: String name of the variable.
value: The value of the variable. This can be any Type.
persist: Boolean as to whether this variable should be saved to
disk so it's available the next time MPF boots.
expire_secs: Optional number of seconds you'd like this variable
to persist on disk for. When MPF boots, if the expiration time
of the variable is in the past, it will be loaded with a value
of 0. For example, this lets you write the number of credits on
the machine to disk to persist even during power off, but you
could set it so that those only stay persisted for an hour.
"""
var = CaseInsensitiveDict()
var['value'] = value
var['persist'] = persist
var['expire_secs'] = expire_secs
self.machine_vars[name] = var
if not silent:
self.set_machine_var(name, value, force_events=True)
def validate_config(self, config_spec, source, section_name=None,
base_spec=None, add_missing_keys=True):
# config_spec, str i.e. "device:shot"
# source is dict
# section_name is str used for logging failures
if source is None:
source = CaseInsensitiveDict()
if not section_name:
section_name = config_spec # str
validation_failure_info = (config_spec, section_name)
this_spec = self._build_spec(config_spec, base_spec)
if '__allow_others__' not in this_spec:
self.check_for_invalid_sections(this_spec, source,
validation_failure_info)
processed_config = source
if not isinstance(source, (list, dict)):
self.validation_error("", validation_failure_info, "Source should be list or dict but is {}".format(
self.log.warning("Received request to set machine_var '%s', but "
"that is not a valid machine_var.", name)
return
prev_value = self.machine_vars[name]['value']
self.machine_vars[name]['value'] = value
try:
change = value - prev_value
except TypeError:
change = prev_value != value
if change or force_events:
if self.machine_vars[name]['persist'] and self.config['mpf']['save_machine_vars_to_disk']:
disk_var = CaseInsensitiveDict()
disk_var['value'] = value
if self.machine_vars[name]['expire_secs']:
disk_var['expire'] = self.clock.get_time() + self.machine_vars[name]['expire_secs']
self.machine_var_data_manager.save_key(name, disk_var)
self.log.debug("Setting machine_var '%s' to: %s, (prior: %s, "
"change: %s)", name, value, prev_value,
change)
self.events.post('machine_var_' + name,
value=value,
prev_value=prev_value,
change=change)
'''event: machine_var_(name)
self.log.debug("Command line arguments: %s", self.options)
self.verify_system_info()
self._exception = None
self._boot_holds = set()
self.is_init_done = False
self.register_boot_hold('init')
self._done = False
self.monitors = dict()
self.plugins = list()
self.scriptlets = list()
self.modes = DeviceCollection(self, 'modes', None)
self.game = None
self.active_debugger = dict()
self.machine_vars = CaseInsensitiveDict()
self.machine_var_monitor = False
self.machine_var_data_manager = None
self.thread_stopper = threading.Event()
self.delayRegistry = DelayManagerRegistry(self)
self.delay = DelayManager(self.delayRegistry)
self.crash_queue = queue.Queue()
self.config = None
self.events = None
self.machine_config = None
self._set_machine_path()
self.config_validator = ConfigValidator(self)
# pylint: disable-msg=protected-access
self.log.info("Starting clock at %sHz", Clock._max_fps)
self._boot_holds = set()
self.mpf_path = os.path.dirname(mpf.__file__)
self.modes = CaseInsensitiveDict()
self.player_list = list()
self.player = None
self.num_players = 0
self.bcp_client_connected = False
self.animation_configs = dict()
self.active_slides = dict()
self.scriptlets = list()
self.register_boot_hold('init')
self.displays = CaseInsensitiveDict()
self.machine_vars = CaseInsensitiveDict()
self.machine_var_monitor = False
self.monitors = dict()
self.targets = dict()
"""Dict which contains all the active slide frames in the machine that
a slide can target. Will always contain an entry called 'default'
which will be used if a slide doesn't specify targeting.
"""
self.keyboard = None
self.physical_dmds = []
self.physical_rgb_dmds = []
self.crash_queue = queue.Queue()
self.ticks = 0
self.start_time = 0
self.is_init_done = False