"""
def __init__(self, conf):
- ## Set daemon
+ # Set daemon
Daemon.__init__(self, conf.get('daemon', 'pidfile'))
self.enabled = True
self.config = conf
@property
def plugins(self):
- return [plugin[1] for plugin in sorted(self._plugins, key=lambda pl: pl[0], reverse=True)]
+ return [plugin[1] for plugin in
+ sorted(self._plugins, key=lambda pl: pl[0], reverse=True)]
def need_tracks(self):
"""Is the player in need for tracks"""
self.log.debug(err)
continue
except PlayerError as err:
- #TODO: unhandled Player exceptions
+ # TODO: unhandled Player exceptions
self.log.warning('Unhandled player exception: %s', err)
self.log.info('Got reconnected')
break
"""Dispatching callbacks to plugins
"""
# hanging here until a monitored event is raised in the player
- if getattr(self, 'changed', False): # first iteration exception
+ if getattr(self, 'changed', False): # first iteration exception
self.changed = self.player.monitor()
else: # first iteration goes through else
self.changed = ['playlist', 'player', 'skipped']
for cplgn in core_plugins:
logger.debug('Register core %(name)s (%(doc)s)', cplgn.info())
sima.register_core_plugin(cplgn)
- logger.debug('core loaded, prioriy: %s', ' > '.join(map(str, sima.core_plugins)))
+ logger.debug('core loaded, prioriy: %s',
+ ' > '.join(map(str, sima.core_plugins)))
# Loading internal plugins
load_plugins(sima, 'internal')
except Exception: # Unhandled exception
exception_log()
+
# Script starts here
def main():
"""Entry point"""
LOG_FORMATS = {
DEBUG: '[{process}]{filename: >11}:{lineno: <3} {levelname: <7}: {message}',
INFO: '{levelname: <7}: {message}',
- #logging.DEBUG: '{asctime} {filename}:{lineno}({funcName}) '
- #'{levelname}: {message}',
}
DATE_FMT = "%Y-%m-%d %H:%M:%S"
# separator. It is used then to split back the string to tags list.
SEPARATOR = chr(0x1F) # ASCII Unit Separator
+
def is_uuid4(uuid):
"""Controls MusicBrainz UUID4 format
return True
return False
+
class MetaException(Exception):
"""Generic Meta Exception"""
@property
def names(self):
"""aliases + name"""
- return self.__aliases | {self.__name,}
+ return self.__aliases | {self.__name, }
@property
@serialize
__author__ = 'Jack Kaliko'
-
from sima import LFM
from sima.lib.meta import Artist
from sima.lib.track import Track
:param BaseCache cache: Set a cache, defaults to `False`.
"""
- stats = {'etag':0,
- 'ccontrol':0,
- 'total':0}
+ stats = {'etag': 0,
+ 'ccontrol': 0,
+ 'total': 0}
def __init__(self):
self.http = HttpClient(cache=self.cache, stats=self.stats)
def _forge_payload(self, artist, method='similar', track=None):
"""Build payload
"""
- payloads = dict({'similar': {'method':'artist.getsimilar',},
- 'top': {'method':'artist.gettoptracks',},
- 'track': {'method':'track.getsimilar',},
- 'info': {'method':'artist.getinfo',},
- })
+ payloads = dict({'similar': {'method': 'artist.getsimilar',},
+ 'top': {'method': 'artist.gettoptracks',},
+ 'track': {'method': 'track.getsimilar',},
+ 'info': {'method': 'artist.getinfo',},
+ })
payload = payloads.get(method)
payload.update(api_key=LFM.get('apikey'), format='json')
if not isinstance(artist, Artist):
ans = self.http(self.root_url, payload)
try:
ans.json()
- except ValueError as err:
+ except ValueError:
# Corrupted/malformed cache? cf. gitlab issue #35
raise WSError('Malformed json, try purging the cache: %s')
- self._controls_answer(ans.json()) # pylint: disable=no-member
+ self._controls_answer(ans.json()) # pylint: disable=no-member
# Artist might be found but return no 'artist' list…
# cf. "Mulatu Astatqe" vs. "Mulatu Astatqé" with autocorrect=0
# json format is broken IMHO, xml is more consistent IIRC
# Here what we got:
# >>> {"similarartists":{"#text":"\n","artist":"Mulatu Astatqe"}}
# autocorrect=1 should fix it, checking anyway.
- simarts = ans.json().get('similarartists').get('artist') # pylint: disable=no-member
+ simarts = ans.json().get('similarartists').get('artist') # pylint: disable=no-member
if not isinstance(simarts, list):
raise WSError('Artist found but no similarities returned')
- for art in ans.json().get('similarartists').get('artist'): # pylint: disable=no-member
+ for art in ans.json().get('similarartists').get('artist'): # pylint: disable=no-member
yield Artist(name=art.get('name'), mbid=art.get('mbid', None))
def get_toptrack(self, artist):
"""
payload = self._forge_payload(artist, method='top')
ans = self.http(self.root_url, payload)
- self._controls_answer(ans.json()) # pylint: disable=no-member
- tops = ans.json().get('toptracks').get('track') # pylint: disable=no-member
+ self._controls_answer(ans.json()) # pylint: disable=no-member
+ tops = ans.json().get('toptracks').get('track') # pylint: disable=no-member
art = {'artist': artist.name,
'musicbrainz_artistid': artist.mbid,}
for song in tops:
"""
sea = SimaStr.reg_lead.search(self.stripped)
if sea:
- #print sea.groupdict()
self.stripped = sea.group('root0')
sea = SimaStr.reg_midl.search(self.stripped)
if sea:
- #print sea.groupdict()
self.stripped = str().join([sea.group('root0'), ' ',
sea.group('root1')])
sea = SimaStr.reg_trail.search(self.stripped)
if sea:
- #print sea.groupdict()
self.stripped = sea.group('root0')
def remove_diacritics(self):
@property
def length(self):
"""Get a fancy duration as ``%H:%M:%S`` (use :attr:`duration` to get duration in second only)"""
- temps = time.gmtime(self.duration) #TODO: returns a date not a duration
+ temps = time.gmtime(self.duration) # TODO: returns a date not a duration
if temps.tm_hour:
fmt = '%H:%M:%S'
else:
return result
return wrapper
+
def set_artist_mbid(func):
def wrapper(*args, **kwargs):
cls = args[0]
return result
return wrapper
+
def tracks_wrapper(func):
"""Convert plain track mapping as returned by MPDClient into :py:obj:`sima.lib.track.Track`
objects. This decorator accepts single track or list of tracks as input.
# local import
from sima.lib.plugin import Plugin
+
class PlaceHolder(Plugin):
"""
Placeholder contrib plugin
pass
-
# VIM MODLINE
# vim: ai ts=4 sw=4 sts=4 expandtab
# local import
from ...lib.plugin import Plugin
+
class History(Plugin):
"""
History management
"""
# standard library import
-from os import getpid
+from os import getpid
from socket import getfqdn
# third parties components
def fetch_genres(self):
"""Fetches ,at most, nb-depth genre from history,
and returns the nbgenres most present"""
- depth = 10 # nb of genre to fetch from history for analysis
- nbgenres = 2 # nb of genre to return
+ depth = 10 # nb of genre to fetch from history for analysis
+ nbgenres = 2 # nb of genre to return
genres = [g[0] for g in self.sdb.fetch_genres_history(limit=depth)]
if not genres:
self.log.debug('No genre found in current track history')
persitent_cache = daemon.config.getboolean('lastfm', 'cache')
if persitent_cache:
CacheController.CACHE_ANYWAY = True
- self.log.debug('Persistant cache enabled in %s', join(vardir, 'http', 'LastFM'))
+ self.log.debug('Persistant cache enabled in %s',
+ join(vardir, 'http', 'LastFM'))
SimaFM.cache = FileCache(join(vardir, 'http', 'LastFM'))
self.ws = SimaFM()
if not tags_config.get('filter', None) and \
config_tags.isdisjoint(sup_tags):
log.warning('Found no config for Tags plugin! '
- 'Need at least "filter" or a supported tag')
+ 'Need at least "filter" or a supported tag')
log.info('Supported Tags are : %s', ', '.join(sup_tags))
# raise PluginException('plugin misconfiguration')
return False
if config_tags.difference(sup_tags):
log.error('Found unsupported tag in config: %s',
- config_tags.difference(sup_tags))
+ config_tags.difference(sup_tags))
# raise PluginException('plugin misconfiguration')
return False
return True
while True:
try:
self.filedsc = os.open(self.lockfile,
- os.O_CREAT|os.O_EXCL|os.O_RDWR)
+ os.O_CREAT | os.O_EXCL | os.O_RDWR)
break
except OSError as err:
if err.errno != errno.EEXIST:
#
"""Computes levenshtein distance/ratio"""
+
def levenshtein(a_st, b_st):
"""Computes the Levenshtein distance between two strings."""
n_a, m_b = len(a_st), len(b_st)
return current[n_a]
+
def levenshtein_ratio(string, strong):
"""
Compute levenshtein ratio.
#
#
-from argparse import ArgumentParser, RawDescriptionHelpFormatter, SUPPRESS
+from argparse import ArgumentParser, RawDescriptionHelpFormatter
from .utils import Wfile, Rfile, Wdir
if isdir(self._file):
self.parser.error('need a file not a directory: {}'.format(self._file))
if not exists(self._dir):
- #raise ArgumentError(self, '"{0}" does not exist'.format(self._dir))
self.parser.error('directory does not exist: {0}'.format(self._dir))
if not exists(self._file):
# Is parent directory writable then
# Copyright (C) 2013 Vinay Sajip. New BSD License.
# Copyright (C) 2014, 2020 kaliko <kaliko@azylum.org>
#
-from __future__ import print_function
REQ_VER = (3,4)
import sys
home / dev / repositories / mpd-sima.git / commitdiff