mirror of
https://github.com/clockworkpi/launcher.git
synced 2025-12-12 17:58:50 +01:00
428 lines
13 KiB
Python
428 lines
13 KiB
Python
# -*- coding: utf-8 -*-
|
|
|
|
import os
|
|
import time
|
|
import pygame
|
|
|
|
import numpy
|
|
import math
|
|
|
|
import gobject
|
|
|
|
from beeprint import pp
|
|
|
|
## local UI import
|
|
from UI.constants import Width,Height,ICON_TYPES
|
|
from UI.page import Page,PageSelector
|
|
from UI.label import Label
|
|
from UI.fonts import fonts
|
|
from UI.util_funcs import midRect
|
|
from UI.keys_def import CurKeys
|
|
from UI.icon_item import IconItem
|
|
from UI.icon_pool import MyIconPool
|
|
from UI.skin_manager import SkinManager
|
|
|
|
from threading import Thread
|
|
|
|
from list_item import ListItem
|
|
|
|
import myvars
|
|
|
|
class PIFI(object):
|
|
_MPD_FIFO = "/tmp/mpd.fifo"
|
|
_SAMPLE_SIZE = 1024
|
|
_SAMPLING_RATE = 44100
|
|
_FIRST_SELECTED_BIN = 5
|
|
_NUMBER_OF_SELECTED_BINS = 1024
|
|
|
|
_samples_buffer = None
|
|
|
|
def __init__(self):
|
|
self.sampleSize = self._SAMPLE_SIZE
|
|
self.samplingRate = self._SAMPLING_RATE
|
|
|
|
def GetSpectrum(self,fifoFile,trim_by=4,log_scale=False,div_by=100):
|
|
try:
|
|
rawSamples = os.read(fifoFile,self.sampleSize) # will return empty lines (non-blocking)
|
|
if len(rawSamples) < 1:
|
|
# print("Read error")
|
|
pass
|
|
else:
|
|
self._samples_buffer = rawSamples
|
|
except Exception,e:
|
|
pass
|
|
|
|
if self._samples_buffer == None:
|
|
return ""
|
|
|
|
data = numpy.fromstring(self._samples_buffer, dtype=numpy.int16)
|
|
|
|
data = data * numpy.hanning(len(data))
|
|
|
|
left,right = numpy.split(numpy.abs(numpy.fft.fft(data)),2)
|
|
spec_y = numpy.add(left,right[::-1])
|
|
|
|
if log_scale:
|
|
spec_y=numpy.multiply(20,numpy.log10(spec_y))
|
|
if trim_by:
|
|
i=int((self.sampleSize/2)/trim_by)
|
|
spec_y=spec_y[:i]
|
|
if div_by:
|
|
spec_y=spec_y/float(div_by)
|
|
|
|
return spec_y
|
|
|
|
|
|
class MPDSpectrumPage(Page):
|
|
|
|
_Icons = {}
|
|
_Selector=None
|
|
_FootMsg = ["Nav","","","Back",""]
|
|
_MyList = []
|
|
_ListFont = fonts["veramono12"]
|
|
_SongFont = fonts["notosanscjk12"]
|
|
_PIFI = None
|
|
_FIFO = None
|
|
_Color = SkinManager().GiveColor('Front')
|
|
_GobjectIntervalId = -1
|
|
_Queue = None
|
|
_KeepReading = True
|
|
_ReadingThread = None
|
|
|
|
_BGpng = None
|
|
_BGwidth = 320
|
|
_BGheight = 200
|
|
|
|
_SheepHead = None
|
|
_SheepHeadW = 69
|
|
_SheepHeadH = 66
|
|
|
|
_SheepBody = None
|
|
_SheepBodyW = 105
|
|
_SheepBodyH = 81
|
|
|
|
_RollCanvas = None
|
|
_RollW = 180
|
|
_RollH = 18
|
|
|
|
_freq_count = 0
|
|
_head_dir = 0
|
|
|
|
_Neighbor = None
|
|
|
|
|
|
_bby = []
|
|
_bbs = []
|
|
_capYPositionArray = []
|
|
_frames = 0
|
|
read_retry = 0
|
|
_queue_data = []
|
|
_vis_values = []
|
|
|
|
def __init__(self):
|
|
Page.__init__(self)
|
|
self._Icons = {}
|
|
self._CanvasHWND = None
|
|
self._MyList = []
|
|
self._PIFI = PIFI()
|
|
|
|
def Init(self):
|
|
self._PosX = self._Index * self._Screen._Width
|
|
self._Width = self._Screen._Width
|
|
self._Height = self._Screen._Height
|
|
|
|
self._CanvasHWND = self._Screen._CanvasHWND
|
|
self._RollCanvas = pygame.Surface(( self._RollW,self._RollH))
|
|
|
|
"""
|
|
self._BGpng = IconItem()
|
|
self._BGpng._ImgSurf = MyIconPool._Icons["sheep_bg"]
|
|
self._BGpng._MyType = ICON_TYPES["STAT"]
|
|
self._BGpng._Parent = self
|
|
self._BGpng.Adjust(0,0,self._BGwidth,self._BGheight,0)
|
|
|
|
self._SheepHead = IconItem()
|
|
self._SheepHead._ImgSurf = MyIconPool._Icons["sheep_head"]
|
|
self._SheepHead._MyType = ICON_TYPES["STAT"]
|
|
self._SheepHead._Parent = self
|
|
self._SheepHead.Adjust(0,0,self._SheepHeadW,self._SheepHeadH,0)
|
|
|
|
self._SheepBody = IconItem()
|
|
self._SheepBody._ImgSurf = MyIconPool._Icons["sheep_body"]
|
|
self._SheepBody._MyType = ICON_TYPES["STAT"]
|
|
self._SheepBody._Parent = self
|
|
self._SheepBody.Adjust(0,0,self._SheepBodyW,self._SheepBodyH,0)
|
|
"""
|
|
|
|
self._cwp_png = IconItem()
|
|
self._cwp_png._ImgSurf = MyIconPool._Icons["tape"]
|
|
self._cwp_png._MyType = ICON_TYPES["STAT"]
|
|
self._cwp_png._Parent = self
|
|
self._cwp_png.Adjust(0,0,79,79,0)
|
|
|
|
|
|
self._song_title = Label()
|
|
self._song_title.SetCanvasHWND(self._RollCanvas)
|
|
self._song_title.Init("Untitled",self._SongFont,SkinManager().GiveColor('White'))
|
|
|
|
|
|
self._title = Label()
|
|
self._title.SetCanvasHWND(self._CanvasHWND)
|
|
self._title.Init("Title:",self._ListFont,SkinManager().GiveColor('White'))
|
|
|
|
self._time = Label()
|
|
self._time.SetCanvasHWND(self._CanvasHWND)
|
|
self._time.Init("Time:",self._ListFont,SkinManager().GiveColor('White'))
|
|
|
|
|
|
self._time2 = Label()
|
|
self._time2.SetCanvasHWND(self._CanvasHWND)
|
|
self._time2.Init("00:00-00:00", self._ListFont,
|
|
SkinManager().GiveColor('White'))
|
|
|
|
|
|
self.Start()
|
|
|
|
def Start(self):
|
|
|
|
if self._Screen.CurPage() != self:
|
|
return
|
|
|
|
try:
|
|
self._FIFO = os.open(self._PIFI._MPD_FIFO, os.O_RDONLY | os.O_NONBLOCK)
|
|
|
|
t = Thread(target=self.GetSpectrum)
|
|
t.daemon = True # thread dies with the program
|
|
t.start()
|
|
self._ReadingThread = t
|
|
|
|
except IOError:
|
|
print("open %s failed"%self._PIFI._MPD_FIFO)
|
|
self._FIFO = None
|
|
return
|
|
|
|
|
|
def GetSpectrum(self):
|
|
while self._KeepReading and self._FIFO != None:
|
|
raw_samples = self._PIFI.GetSpectrum(self._FIFO)
|
|
if len(raw_samples) < 1:
|
|
#print("sleeping... 0.01")
|
|
time.sleep(0.01)
|
|
self.read_retry+=1
|
|
if self.read_retry > 20:
|
|
os.close(self._FIFO)
|
|
self._FIFO = os.open(self._PIFI._MPD_FIFO, os.O_RDONLY | os.O_NONBLOCK)
|
|
self.read_retry = 0
|
|
|
|
self.Playing()
|
|
|
|
else:
|
|
self.read_retry = 0
|
|
self._queue_data = raw_samples
|
|
self.Playing()
|
|
|
|
|
|
def Playing(self):
|
|
|
|
self._Screen.Draw()
|
|
self._Screen.SwapAndShow()
|
|
|
|
|
|
def ClearCanvas(self):
|
|
self._CanvasHWND.fill(SkinManager().GiveColor('Black'))
|
|
|
|
def SgsSmooth(self):
|
|
passes = 1
|
|
points = 3
|
|
origs = self._bby[:]
|
|
for p in range(0,passes):
|
|
pivot = int(points/2.0)
|
|
|
|
for i in range(0,pivot):
|
|
self._bby[i] = origs[i]
|
|
self._bby[ len(origs) -i -1 ] = origs[ len(origs) -i -1 ]
|
|
|
|
smooth_constant = 1.0/(2.0*pivot+1.0)
|
|
for i in range(pivot, len(origs)-pivot):
|
|
_sum = 0.0
|
|
for j in range(0,(2*pivot)+1):
|
|
_sum += (smooth_constant * origs[i+j-pivot]) +j -pivot
|
|
|
|
self._bby[i] = _sum
|
|
|
|
if p < (passes - 1):
|
|
origs = self._bby[:]
|
|
|
|
def OnLoadCb(self):
|
|
if self._Neighbor != None:
|
|
pass
|
|
|
|
if self._KeepReading == False:
|
|
self._KeepReading = True
|
|
|
|
if self._FIFO == None:
|
|
self.Start()
|
|
|
|
|
|
def KeyDown(self,event):
|
|
if event.key == CurKeys["Menu"] or event.key == CurKeys["A"]:
|
|
try:
|
|
os.close(self._FIFO)
|
|
self._FIFO = None
|
|
|
|
except Exception, e:
|
|
print(e)
|
|
|
|
self._KeepReading = False
|
|
self._ReadingThread.join()
|
|
self._ReadingThread = None
|
|
|
|
self.ReturnToUpLevelPage()
|
|
self._Screen.Draw()
|
|
self._Screen.SwapAndShow()
|
|
|
|
if event.key == CurKeys["Start"]:
|
|
self._Screen.Draw()
|
|
self._Screen.SwapAndShow()
|
|
|
|
if event.key == CurKeys["Enter"]:
|
|
pass
|
|
|
|
|
|
def Draw(self):
|
|
self.ClearCanvas()
|
|
self._frames+=1
|
|
|
|
bw = 10
|
|
gap = 2
|
|
margin_bottom = 72
|
|
|
|
spects = None
|
|
meterNum = self._Width / float(bw +gap ) ## 320/12= 26
|
|
meter_left = meterNum - int(meterNum)
|
|
meter_left = meter_left*int(bw+gap)
|
|
margin_left = meter_left / 2 + gap
|
|
meterNum = int(meterNum)
|
|
|
|
self._cwp_png.NewCoord(43,159)
|
|
self._cwp_png.Draw()
|
|
|
|
if self._Neighbor != None:
|
|
if self._Neighbor._CurSongName != "":
|
|
self._song_title.SetText(self._Neighbor._CurSongName)
|
|
if self._Neighbor._CurSongTime != "":
|
|
times = self._Neighbor._CurSongTime
|
|
times_ = times.split(":")
|
|
if len(times_)> 1:
|
|
cur = int(times_[0])
|
|
end = int(times_[1])
|
|
if cur > 3600:
|
|
cur_text = time.strftime('%H:%M:%S', time.gmtime(cur))
|
|
else:
|
|
cur_text = time.strftime('%M:%S', time.gmtime(cur))
|
|
|
|
if end > 3600:
|
|
end_text = time.strftime('%H:%M:%S', time.gmtime(end))
|
|
else:
|
|
end_text = time.strftime('%M:%S', time.gmtime(end))
|
|
else:
|
|
cur_text = ""
|
|
end_text = times
|
|
|
|
self._time2.SetText(cur_text+"-"+end_text)
|
|
|
|
|
|
self._title.NewCoord(90,167)
|
|
self._title.Draw()
|
|
|
|
self._time.NewCoord(90,140)
|
|
self._time.Draw()
|
|
|
|
self._time2.NewCoord(135,140)
|
|
self._time2.Draw()
|
|
|
|
if self._RollCanvas != None:
|
|
# self._RollCanvas.fill((111,22,33))
|
|
self._RollCanvas.fill(SkinManager().GiveColor('Black'))
|
|
if self._song_title._Width > self._RollW:
|
|
if (self._song_title._PosX + self._song_title._Width) > self._RollW and self._frames % 30 == 0:
|
|
self._song_title._PosX -= 1
|
|
elif (self._song_title._PosX + self._song_title._Width) <= self._RollW and self._frames % 30 == 0:
|
|
self._song_title._PosX = 0
|
|
else:
|
|
self._song_title._PosX = 0
|
|
|
|
self._song_title.Draw()
|
|
|
|
self._CanvasHWND.blit(self._RollCanvas,(135,165,self._RollW,self._RollH))
|
|
|
|
|
|
try:
|
|
spects = self._queue_data
|
|
if len(spects) == 0:
|
|
return
|
|
# print("spects:",spects)
|
|
|
|
step = int( round( len( spects ) / meterNum) )
|
|
# print(len(spects))
|
|
self._bbs = []
|
|
a = numpy.logspace(0, 1, num=meterNum,endpoint=True)
|
|
|
|
for i in range(0,meterNum):
|
|
index = int(a[i] * step)
|
|
total = 0
|
|
|
|
value = spects[index]
|
|
self._bbs.append(value)
|
|
|
|
if len(self._bby) < len(self._bbs):
|
|
self._bby = self._bbs
|
|
elif len(self._bby) == len(self._bbs):
|
|
for i in range(0,len(self._bbs)):
|
|
self._bby[i] = (self._bby[i]+self._bbs[i])/2
|
|
|
|
self.SgsSmooth()
|
|
|
|
for i in range(0,meterNum):
|
|
value = self._bby[ i ]
|
|
if math.isnan(value) or math.isinf(value):
|
|
value = 0
|
|
|
|
value = value/32768.0
|
|
value = value * 123
|
|
value = value % (self._Height-gap-margin_bottom)
|
|
|
|
if len(self._vis_values) < len(self._bby):
|
|
self._vis_values.append(value)
|
|
elif len(self._vis_values) == len(self._bby):
|
|
if self._vis_values[i] < value:
|
|
self._vis_values[i] = value
|
|
|
|
|
|
except Exception,e:
|
|
print(e)
|
|
return
|
|
else: # got line
|
|
if len(self._vis_values) == 0:
|
|
return
|
|
|
|
for i in range(0,meterNum):
|
|
value = self._vis_values[i]
|
|
|
|
if len(self._capYPositionArray) < round(meterNum):
|
|
self._capYPositionArray.append(value)
|
|
|
|
if value < self._capYPositionArray[i]:
|
|
self._capYPositionArray[i]-=0.5
|
|
else:
|
|
self._capYPositionArray[i] = value
|
|
|
|
pygame.draw.rect(self._CanvasHWND,SkinManager().GiveColor('White'),(i*(bw+gap)+margin_left,self._Height-gap-self._capYPositionArray[i]-margin_bottom,bw,gap),0)
|
|
|
|
pygame.draw.rect(self._CanvasHWND,SkinManager().GiveColor('White'),(i*(bw+gap)+margin_left,self._Height-value-gap-margin_bottom,bw,value+gap),0)
|
|
|
|
self._vis_values[i] -= 2
|
|
|
|
|
|
|