Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def get_class_name(name, postfix=''):
name = identifier(name)
return ''.join(map(capitalize, name.split('_'))) + postfix
def __init__(self, path, channels=None, sample_rate_hz=None):
self.path = path
self.channels = channels
self.sample_rate_hz = sample_rate_hz
if self.channels is None:
self._load_channels()
headings = [chan.label for chan in self.channels]
self.data_tuple = collections.namedtuple('csv_entry',
map(identifier, headings))
def has(self, modname):
return hasattr(self, identifier(modname))
def get_plugin(self, name=None, kind=None, *args, **kwargs):
return self.plugin_cache.get_plugin(identifier(name), kind, *args, **kwargs)
def install(cls, target, **params):
if cls.kind is not None:
attr_name = identifier(cls.kind)
else:
attr_name = identifier(cls.name)
if hasattr(target, attr_name):
existing_module = getattr(target, attr_name)
existing_name = getattr(existing_module, 'name', str(existing_module))
message = 'Attempting to install module "{}" which already exists (new: {}, existing: {})'
raise ValueError(message.format(attr_name, cls.name, existing_name))
setattr(target, attr_name, cls(target, **params))
def install(cls, target, **params):
if cls.kind is not None:
attr_name = identifier(cls.kind)
else:
attr_name = identifier(cls.name)
if hasattr(target, attr_name):
existing_module = getattr(target, attr_name)
existing_name = getattr(existing_module, 'name', str(existing_module))
message = 'Attempting to install module "{}" which already exists (new: {}, existing: {})'
raise ValueError(message.format(attr_name, cls.name, existing_name))
setattr(target, attr_name, cls(target, **params))
attr = getattr(cls, attr_name)
if name == attr:
return attr
try:
return Enum.from_pod(name)
except ValueError:
raise ValueError('Invalid enum value: {}'.format(repr(name)))
reserved = ['values', 'levels', 'names']
levels = []
n = start
for v in args:
id_v = identifier(v)
if id_v in reserved:
message = 'Invalid enum level name "{}"; must not be in {}'
raise ValueError(message.format(v, reserved))
name = caseless_string(id_v)
lv = level(v, n)
setattr(Enum, name, lv)
levels.append(lv)
n += step
setattr(Enum, 'levels', levels)
setattr(Enum, 'values', [lvl.value for lvl in levels])
setattr(Enum, 'names', [lvl.name for lvl in levels])
return Enum
value = pop_aliased_param(cfg_point, raw)
if value is not None:
logger.debug('Setting run "{}" to "{}"'.format(cfg_point.name, value))
state.run_config.set(cfg_point.name, value)
# Get global job spec configuration
for cfg_point in JobSpec.configuration.values():
value = pop_aliased_param(cfg_point, raw)
if value is not None:
logger.debug('Setting global "{}" to "{}"'.format(cfg_point.name, value))
state.jobs_config.set_global_value(cfg_point.name, value)
for name, values in raw.items():
# Assume that all leftover config is for a plug-in or a global
# alias it is up to PluginCache to assert this assumption
logger.debug('Caching "{}" with "{}"'.format(identifier(name), values))
state.plugin_cache.add_configs(identifier(name), values, source)
except ConfigError as e:
if wrap_exceptions:
raise ConfigError('Error in "{}":\n{}'.format(source, str(e)))
else:
raise e
finally:
log.dedent()