led-matrix-player/main.py

import sys
import spotipy
import spotipy.util as util
import config as cfg
from PIL import Image, ImageDraw, ImageFont
import asyncio
from rgbmatrix import RGBMatrix, RGBMatrixOptions
import threading
import requests
from io import BytesIO
import time
from datetime import datetime

client_id = cfg.client_id
client_secret = cfg.client_secret
redirect_uri = cfg.redirect_uri
username = cfg.username

scope = 'user-read-currently-playing'

options = RGBMatrixOptions()
options.rows = 32
options.cols = 64
options.chain_length = 6
options.parallel = 2
options.hardware_mapping = 'regular'
options.pixel_mapper_config = 'U-mapper'
options.limit_refresh_rate_hz = 240
options.gpio_slowdown = 4
options.brightness = 50
options.disable_hardware_pulsing = True
screen = RGBMatrix(options = options)

song_name=""
artist_name=""

cache_path="/home/pi/.cache-"+username

is_playing = False

is_scrolling_art = False

cover_art_offset_x = 0
cover_art_offset_y = 96

last_cover_art = Image.new('RGB',(32,32),color='white')
last_song = ''
last_artist = ''

current_cover_art = Image.new('RGB',(32,32),color='white')
current_song = ''
current_artist = ''

token = None #oauth2.SpotifyOAuth(username, scope, client_id, client_secret, redirect_uri)

#token = util.prompt_for_user_token(username, scope, client_id, client_secret, redirect_uri)

font = ImageFont.load_default()
clock_font = ImageFont.truetype(font='BebasNeue_Regular_1.otf', size=75)


def get_track():
    print('getting current song info')
    sp = spotipy.Spotify(auth=token)
    track = sp.currently_playing()
    if track is None:
        return Image.new('RGB',(32,32),color='black')
    global is_playing
    is_playing = track['is_playing']
    global song_name
    if track is None:
        return Image.new('RGB',(32,32),color='black')
    song_name = track['item']['name']
    global artist_name
    artist_name = track['item']['artists'][0]['name']
    for i in range(len(track['item']['artists'])-1):
        artist_name +=', ' + track['item']['artists'][i+1]['name']
    print('got info: ' + track['item']['name'] + ' - ' + track['item']['artists'][0]['name'])
    #print('getting image')
    image_url = track['item']['album']['images'][0]['url']
    response = requests.get(image_url)
    image = Image.open(BytesIO(response.content))
    #print(image_url)
    image = image.resize((32,32))
    return image

def showImage(image, offset_x = 0, offset_y = 0):
    image = image.convert('RGB')
    width, height = image.size
    for x in range(width):
        for y in range(height):
            r, g, b = image.getpixel((x, y))
            screen.SetPixel(x+offset_x,y+offset_y,r,g,b)

def changeSong():
    scrollDownSongInfo(song_name=last_song,artist_name=last_artist,cover_art=last_cover_art)
    scrollUpSongInfo(song_name=current_song,artist_name=current_artist,cover_art=current_cover_art)


def drawCanvas():
    canvas = Image.new('RGB', (192, 128), color = 'black')
    global last_song
    global last_artist
    was_playing = is_playing
    
    cover_art = get_track()
    
    global is_scrolling_art
    if (cover_art is None):
        return
    
    if not (last_song == song_name):
        change_song_scroll = threading.Thread(target=changeSong)
        change_song_scroll.start()
    last_song=song_name
    last_artist = artist_name
    global last_cover_art
    last_cover_art = cover_art
    global current_cover_art
    current_cover_art = cover_art
    global current_song
    current_song = song_name
    global current_artist
    current_artist = artist_name
    if(was_playing and not is_playing):
        is_scrolling_art = True
        scroll = threading.Thread(target=scrollDownSongInfo)
        scroll.start()
        
    if(not was_playing and is_playing):
        is_scrolling_art = True
        scroll = threading.Thread(target=scrollUpSongInfo)
        scroll.start()
    if is_scrolling_art:
        return
    if(is_playing):
        canvas.paste(cover_art,(cover_art_offset_x,cover_art_offset_y))
        songinfo = ImageDraw.Draw(canvas)
        songinfo_offset_x = 33
        songinfo_offset_y = 101
        songinfo.text((songinfo_offset_x,songinfo_offset_y+11), song_name)
        songinfo.text((songinfo_offset_x,songinfo_offset_y), artist_name)
        
        #scrolling_text(song_name,songinfo_offset_x, songinfo_offset_y+10)
        if(font.getsize(song_name)[0]>160):
            x = threading.Thread(target=scrolling_text,args=(song_name,songinfo_offset_x,songinfo_offset_y+11))
            x.start()
            print('scrolling text started')
        if(font.getsize(artist_name)[0]>160):
            
            y = threading.Thread(target=scrolling_text,args=(artist_name,songinfo_offset_x,songinfo_offset_y))
            y.start()
            
    #showImage(canvas)

def scrollDownSongInfo(offset_x=0,offset_y=96, song_name=song_name, artist_name=artist_name, cover_art=last_cover_art):
    global is_scrolling_art
    #cover_art = get_track()
    if (cover_art is None):
        return
    for i in range(33):
        song_info_canvas = Image.new('RGB', (192,32), color = 'black')
        songinfo = ImageDraw.Draw(song_info_canvas)
        songinfo_offset_x = 33
        songinfo.text((songinfo_offset_x,15+i), song_name)
        songinfo.text((songinfo_offset_x,4+i), artist_name)
        song_info_canvas.paste(cover_art,(0,i))
        showImage(song_info_canvas, offset_x, offset_y)
        time.sleep(0.001)
    is_scrolling_art = False

def scrollUpSongInfo(offset_x=0, offset_y=96, song_name=song_name, artist_name=artist_name, cover_art=last_cover_art):
    global is_scrolling_art
    #cover_art = get_track()
    if (cover_art is None):
        return
    for i in range(33):
        song_info_canvas = Image.new('RGB', (192,32), color = 'black')
        songinfo = ImageDraw.Draw(song_info_canvas)
        songinfo_offset_x = 33
        songinfo.text((songinfo_offset_x,15+(33-i)), song_name)
        songinfo.text((songinfo_offset_x,4+(33-i)), artist_name)
        song_info_canvas.paste(cover_art,(0,(32-i)))
        showImage(song_info_canvas, offset_x, offset_y)
        time.sleep(0.001)
    is_scrolling_art = False

def scrolling_text(text,offset_x=0,offset_y=0):
    scroll_frame_time = (6/2)/(font.getsize(text)[0]-160)
    #print(scroll_frame_time)
    if(scroll_frame_time>0.03):
        scroll_frame_time = 0.03
    for i in range((font.getsize(text)[0]-160)):
        scrolled_text_canvas = Image.new('RGB', (160, 11), color = 'black')
        scrolled_text = ImageDraw.Draw(scrolled_text_canvas)
        scrolled_text.text((-i,0), text)
        showImage(scrolled_text_canvas, offset_x, offset_y)
        time.sleep(scroll_frame_time)
        #print('scrolled '+str(i)+' px')
    if((5-(font.getsize(text)[0]-160)*2*scroll_frame_time)>0):
        time.sleep(5-(font.getsize(text)[0]-160)*2*scroll_frame_time)
    for i in range(font.getsize(text)[0]-160):
        scrolled_text_canvas = Image.new('RGB', (160, 11), color = 'black')
        scrolled_text = ImageDraw.Draw(scrolled_text_canvas)
        scrolled_text.text(((-(font.getsize(text)[0]-160))+i+1,0), text)
        showImage(scrolled_text_canvas, offset_x, offset_y)
        time.sleep(scroll_frame_time)
        #print('scrolled '+str(i)+' px')
    
        
def main():
    get_access_token()
    clock = threading.Thread(target=drawClock)
    clock.start()
    print('led-matrix-viewer started')
    while True:
        get_access_token()
        drawCanvas()
        time.sleep(10)
    
def drawClock():
    while True:
        canvas = Image.new('RGB', (128,64), color = 'black')
        clock_text = ImageDraw.Draw(canvas)
        clock_text.text((0,0), datetime.now().strftime("%H:%M"), font = clock_font, align = "center")
        showImage(canvas, 32, 0)
        time.sleep(20)
    
    
    
def get_access_token():
    global token
    sp_oauth = spotipy.SpotifyOAuth(
        client_id,
        client_secret,
        redirect_uri,
        scope=scope,
        #cache_path=cache_path,
        username=username,
        open_browser=False
    )
    token_info = sp_oauth.get_cached_token()

    if not token_info:
        code = sp_oauth.get_auth_response()
        token = sp_oauth.get_access_token(code, as_dict=False)
    else:
        if sp_oauth.is_token_expired(token_info):
            token_info = refresh_access_token(token_info)
    token =  token_info["access_token"]
    

if __name__ == "__main__":
    main()