#!/usr/bin/env python3 # -*- coding: utf-8 -*- __name__ = "Luniebox" import sys import logging import subprocess import time from enum import Enum from configparser import ConfigParser import RPi.GPIO as GPIO from spotifydl import SpotifyDL, SpotifyDLStatus from mpd import MPDClient from ExtendedMFRC522 import ExtendedMFRC522 from spotify import Spotify import util defaultConfigFilePath = '../config/luniebox.cfg' class PlayerService(Enum): NONE = 0 SPOTIFY = 1 MPD = 2 class Luniebox(object): def __init__(self, configFilePath=defaultConfigFilePath): self.configFilePath = configFilePath self.read_config() GPIO.setwarnings(False) self.reader = ExtendedMFRC522(encoding="ascii") loglevel = 'INFO' if self.get_setting('logging', 'level'): loglevel = self.get_setting('logging', 'level') logger = logging.getLogger('luniebox') logger.setLevel(logging._nameToLevel[loglevel]) logFormatter = logging.Formatter( style='{', datefmt='%Y-%m-%d %H:%M:%S', fmt='{asctime} {levelname}: {message}') logstdoutHandler = logging.StreamHandler(sys.stdout) logstdoutHandler.setFormatter(logFormatter) logger.addHandler(logstdoutHandler) if self.get_setting('mpd', 'disabled', '') != 'True': self.mpd = MPDClient() self.mpd.host = "localhost" self.mpd.port = 6600 self.mpd.timeout = 10 if self.mpd_connect(): logging.getLogger('luniebox').debug("connected to mpd") else: logging.getLogger('luniebox').info("mpd disabled") self.mpd = False if self.get_setting('spotify', 'disabled', '') != 'True': self.spotify = Spotify(luniebox=self) if self.spotify_connect(): logging.getLogger('luniebox').debug("connected to spotify") self.zspotify_path = False self.spotifydl = False self.zspotify_path = self.get_setting('spotify', 'zspotify_path') if self.zspotify_path and self.mpd: self.spotifydl_connect() else: logging.getLogger('luniebox').info("spotify disabled") self.spotify = False if not self.config.has_option('api', 'api_key'): self.set_setting('api', 'api_key', util.randomString(64)) self.current = self.get_setting('luniebox', 'current') self.service = PlayerService.NONE if self.current != None: if self.current.startswith('spotify:'): self.service = PlayerService.SPOTIFY elif self.current.startswith('mpd:'): self.service = PlayerService.MPD self.volume_max = int(self.get_setting('luniebox', 'volume_max', 100)) self.volume = int(self.get_setting( 'luniebox', 'volume', self.volume_max)) self.volume_step = int(self.get_setting('luniebox', 'volume_step', 5)) self.wind_step = int(self.get_setting('luniebox', 'wind_step', 10)) self.resume = True def mpd_connect(self): if not self.get_setting('luniebox', 'setup'): return False if not self.mpd: logging.getLogger('luniebox').warn("mpd disabled!") return False try: self.mpd.disconnect() except: pass tries = 0 while tries < 5: try: self.mpd.connect(self.mpd.host, self.mpd.port) return True except: subprocess.call(["systemctl", "restart", "mpd"]) time.sleep(1) tries += 1 logging.getLogger('luniebox').error( "Could not connect to MPD service!") return False def spotify_connect(self, restart=False, max_tries=5): if not self.get_setting('luniebox', 'setup'): return False if not self.spotify: logging.getLogger('luniebox').warn("spotify disabled!") return False if restart: subprocess.run(["sudo", "systemctl", "restart", "spotifyd"]) time.sleep(20) tries = 0 spotifyd_status = subprocess.call( ["systemctl", "is-active", "--quiet", "spotifyd"]) while spotifyd_status != 0 and tries < max_tries: subprocess.call(["systemctl", "restart", "spotifyd"]) tries = tries + 1 time.sleep(10 * tries) spotifyd_status = subprocess.call( ["systemctl", "is-active", "--quiet", "spotifyd"]) if spotifyd_status == 0: return True logging.getLogger('luniebox').error("spotifyd service not running!") return False def spotifydl_connect(self): if not self.get_setting('luniebox', 'setup'): return False if self.spotifydl: return True if not self.zspotify_path or not self.mpd: logging.getLogger('luniebox').warn("spotifydl disabled!") return False try: self.spotifydl = SpotifyDL(self) logging.getLogger('luniebox').info("spotifydl enabled!") return True except Exception as exception: logging.getLogger('luniebox').warning( "error on setup spotifydl: " + str(exception)) return False def read_config(self): configParser = ConfigParser() dataset = configParser.read(self.configFilePath) if len(dataset) != 1: raise ValueError( "Config file {} not found!".format(self.configFilePath)) self.config = configParser def get_setting(self, section, key, default=None): if self.config.has_option(section, key): return self.config[section][key] return default def set_setting(self, section, key, value): if not self.config.has_section(section): self.config.add_section(section) self.config.set(section, key, str(value)) with open(self.configFilePath, 'w') as configfile: self.config.write(configfile) def rfid_write(self, data): self.reader.write(data.strip()) def rfid_readOnce(self): id, text = self.reader.read() logging.getLogger('luniebox').debug( "Once ID: %s Text: %s" % (id, text)) return text.strip() def vol_up(self): self.change_volume(self.volume_step) def vol_down(self): self.change_volume(-1 * self.volume_step) def change_volume(self, change): self.volume += change if self.volume > self.volume_max: self.volume = self.volume_max if self.volume < 0: self.volume = 0 if self.service == PlayerService.SPOTIFY and self.spotify_connect(): self.spotify.volume(self.volume) elif self.service == PlayerService.MPD and self.mpd_connect(): self.mpd.setvol(self.volume) self.set_setting('luniebox', 'volume', str(self.volume)) logging.getLogger('luniebox').debug("vol: " + str(self.volume)) def fastforward(self): if not self.resume: logging.getLogger('luniebox').debug('seek ff') if self.service == PlayerService.SPOTIFY and self.spotify_connect(): self.spotify.fastforward(self.wind_step) elif self.service == PlayerService.MPD and self.mpd_connect(): self.mpd.seekcur(self.wind_step) def rewind(self): if not self.resume: logging.getLogger('luniebox').debug('seek rew') if self.service == PlayerService.SPOTIFY and self.spotify_connect(): self.spotify.rewind(self.wind_step) elif self.service == PlayerService.MPD and self.mpd_connect(): self.mpd.seekcur(-1 * self.wind_step) def next(self): if not self.resume: logging.getLogger('luniebox').debug('next') if self.service == PlayerService.SPOTIFY and self.spotify_connect(): self.spotify.next() elif self.service == PlayerService.MPD and self.mpd_connect(): self.mpd.next() def previous(self): if not self.resume: logging.getLogger('luniebox').debug('prev') if self.service == PlayerService.SPOTIFY and self.spotify_connect(): self.spotify.previous() elif self.service == PlayerService.MPD and self.mpd_connect(): self.mpd.previous() def pause(self): if not self.resume: if self.mpd_connect(): self.mpd.pause(1) logging.getLogger('luniebox').debug("pause mpd") if self.spotify_connect(max_tries=2): self.spotify.pause() logging.getLogger('luniebox').debug("pause spotify") self.resume = True def play(self, text): if text != "": if text.startswith('spotify:') and self.spotify: self.service = PlayerService.SPOTIFY elif text.startswith('mpd:') and self.mpd: self.service = PlayerService.MPD else: self.service = PlayerService.NONE if self.service == PlayerService.SPOTIFY and self.spotifydl_connect(): downloadStatus = self.spotifydl.downloadStatus(text) if downloadStatus == SpotifyDLStatus.FINISHED and self.mpd_connect(): self.mpd.update() self.mpd.idle('update') self.service = PlayerService.MPD elif self.get_setting('spotify', 'auto_download') == 'True' and downloadStatus == SpotifyDLStatus.NONE or downloadStatus == SpotifyDLStatus.ERROR: self.spotifydl.download(text) if self.service == PlayerService.SPOTIFY and self.spotify_connect(): if text != self.current: self.spotify.volume(self.volume) if self.spotify.play(text): self.current = text self.set_setting( 'luniebox', 'current', self.current) self.resume = False logging.getLogger('luniebox').debug( "play spotify: " + self.current) else: logging.getLogger('luniebox').warn( "cannot play spotify: " + self.current) elif self.resume and text == self.current: self.spotify.volume(self.volume) play = self.current if self.spotify.is_active(): play = None if self.spotify.play(play): self.resume = False logging.getLogger('luniebox').debug( "resume spotify: " + self.current) else: logging.getLogger('luniebox').warn( "cannot resume spotify: " + self.current) return True elif self.service == PlayerService.MPD and self.mpd_connect(): if text != self.current: try: self.mpd.setvol(self.volume) self.mpd.clear() mpd_uri = text.replace('mpd:', '') self.mpd.add(mpd_uri) self.mpd.play() except Exception as exception: logging.getLogger('luniebox').warning( "cannot not play mpd '" + text + "': " + str(exception)) return False self.current = text self.set_setting('luniebox', 'current', self.current) self.resume = False if text.startswith('spotify:'): logging.getLogger('luniebox').debug( "play spotify from mpd: " + text) else: logging.getLogger('luniebox').debug( "play mpd: " + self.current) elif self.resume and text == self.current: try: self.mpd.setvol(self.volume) self.mpd.play() except Exception as exception: logging.getLogger('luniebox').warning( "cannot not resume mpd '" + self.current + "': " + str(exception)) return False self.resume = False if text.startswith('spotify:'): logging.getLogger('luniebox').debug( "resume spotify from mpd: " + text) else: logging.getLogger('luniebox').debug( "resume mpd: " + self.current) return True elif text != None: logging.getLogger('luniebox').info( "invalid value(?): " + str(text)) return False return False def stop(self): self.pause() def sort_mpd(self, object): if 'directory' in object: return object['directory'] elif 'file' in object: return object['file'] return '' def mpd_list(self, path=''): if self.mpd_connect(): try: self.mpd.update(path) result = self.mpd.listfiles(path) return sorted(result, key=self.sort_mpd) except: return [] return None def mpd_status(self): if self.mpd_connect(): status = self.mpd.status() status['song'] = self.mpd.currentsong() return status return None luniebox = Luniebox()