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, *plugins):
"""Load and initialize plugins by names.
Similar setting a list of plugins in the configuration. Make
sure you call ``unload_plugins()`` afterwards.
"""
beets.config['plugins'] = plugins
beets.plugins.load_plugins(plugins)
beets.plugins.find_plugins()
def setUp(self):
# A "clean" source list including only the defaults.
beets.config.sources = []
beets.config.read(user=False, defaults=True)
# Direct paths to a temporary directory. Tests can also use this
# temporary directory.
self.temp_dir = tempfile.mkdtemp()
beets.config['statefile'] = os.path.join(self.temp_dir, 'state.pickle')
beets.config['library'] = os.path.join(self.temp_dir, 'library.db')
beets.config['directory'] = os.path.join(self.temp_dir, 'libdir')
# Set $HOME, which is used by confit's `config_dir()` to create
# directories.
self._old_home = os.environ.get('HOME')
os.environ['HOME'] = self.temp_dir
# Initialize, but don't install, a DummyIO.
self.io = DummyIO()
def _stream_encoding(stream, default='utf-8'):
"""A helper for `_in_encoding` and `_out_encoding`: get the stream's
preferred encoding, using a configured override or a default
fallback if neither is not specified.
"""
# Configured override?
encoding = config['terminal_encoding'].get()
if encoding:
return encoding
# For testing: When sys.stdout or sys.stdin is a StringIO under the
# test harness, it doesn't have an `encoding` attribute. Just use
# UTF-8.
if not hasattr(stream, 'encoding'):
return default
# Python's guessed output stream encoding, or UTF-8 as a fallback
# (e.g., when piped to a file).
return stream.encoding or default
def update(self, lib):
"""When the client exists try to send refresh request to Emby.
"""
self._log.info(u'Updating Emby library...')
host = config['emby']['host'].get()
port = config['emby']['port'].get()
username = config['emby']['username'].get()
password = config['emby']['password'].get()
# Get user information from the Emby API.
user = get_user(host, port, username)
if not user:
self._log.warning(u'User {0} could not be found.'.format(username))
return
# Create Authentication data and headers.
auth_data = password_data(username, password)
headers = create_headers(user[0]['Id'])
# Get authentication token.
token = get_token(host, port, headers, auth_data)
if not token:
self._log.warning(
def __init__(self):
super(PlaylistPlugin, self).__init__()
self.config.add({
'auto': False,
'playlist_dir': '.',
'relative_to': 'library',
})
self.playlist_dir = self.config['playlist_dir'].as_filename()
self.changes = {}
if self.config['relative_to'].get() == 'library':
self.relative_to = beets.util.bytestring_path(
beets.config['directory'].as_filename())
elif self.config['relative_to'].get() != 'playlist':
self.relative_to = beets.util.bytestring_path(
self.config['relative_to'].as_filename())
else:
self.relative_to = None
if self.config['auto']:
self.register_listener('item_moved', self.item_moved)
self.register_listener('item_removed', self.item_removed)
self.register_listener('cli_exit', self.cli_exit)
def _stream_encoding(stream, default='utf-8'):
"""A helper for `_in_encoding` and `_out_encoding`: get the stream's
preferred encoding, using a configured override or a default
fallback if neither is not specified.
"""
# Configured override?
encoding = config['terminal_encoding'].get()
if encoding:
return encoding
# For testing: When sys.stdout or sys.stdin is a StringIO under the
# test harness, it doesn't have an `encoding` attribute. Just use
# UTF-8.
if not hasattr(stream, 'encoding'):
return default
# Python's guessed output stream encoding, or UTF-8 as a fallback
# (e.g., when piped to a file).
return stream.encoding or default
def func(lib, opts, args):
# The "write to files" option corresponds to the
# import_write config value.
write = config['import']['write'].get(bool)
for item in lib.items(ui.decargs(args)):
fetch_item_tempo(lib, logging.INFO, item, write)
if opts.printbpm and item.bpm:
ui.print_('{0} BPM'.format(item.bpm))
cmd.func = func
u'See changelog & documentation.',
old_key,
new_key,
)
config[new_key].set(config[old_key])
config_path = config.user_config_path()
if os.path.isfile(config_path):
log.debug(u'user configuration: {0}',
util.displayable_path(config_path))
else:
log.debug(u'no user configuration found at {0}',
util.displayable_path(config_path))
log.debug(u'data directory: {0}',
util.displayable_path(config.config_dir()))
return config
# Add any additional config files specified with --config. This
# special handling lets specified plugins get loaded before we
# finish parsing the command line.
if getattr(options, 'config', None) is not None:
config_path = options.config
del options.config
config.set_file(config_path)
config.set_args(options)
# Configure the logger.
if config['verbose'].get(int):
log.set_global_level(logging.DEBUG)
else:
log.set_global_level(logging.INFO)
config_path = config.user_config_path()
if os.path.isfile(config_path):
log.debug(u'user configuration: {0}',
util.displayable_path(config_path))
else:
log.debug(u'no user configuration found at {0}',
util.displayable_path(config_path))
log.debug(u'data directory: {0}',
util.displayable_path(config.config_dir()))
return config
@LazyClassProperty
def _weights(cls):
"""A dictionary from keys to floating-point weights.
"""
weights_view = config['match']['distance_weights']
weights = {}
for key in weights_view.keys():
weights[key] = weights_view[key].as_number()
return weights