webradio/software/webradio.py

271 lines
7.2 KiB
Python
Raw Normal View History

2020-07-05 17:45:45 +02:00
#!/usr/bin/env python
from PIL import Image, ImageDraw, ImageFont
import asyncio
from mpd import MPDClient
2020-07-19 18:39:01 +02:00
import spidev
2020-07-05 17:45:45 +02:00
2020-08-08 16:49:31 +02:00
IO_DELAY = (1.0 / 18.0)
2020-07-05 20:06:20 +02:00
MPD_DELAY = 0.5
2020-08-08 16:49:31 +02:00
SPI_SPEED = 500000
2020-07-05 17:45:45 +02:00
class display(object):
def __init__(self):
self.fg = 1
self.width = 128
self.height = 32
self.font_title = ImageFont.truetype('/usr/share/fonts/noto-cjk/NotoSansCJK-Light.ttc', 18)
self.font_info = ImageFont.truetype('/usr/share/fonts/TTF/DejaVuSans.ttf', 10)
self.img = Image.new('1', (self.width, self.height), color = not self.fg)
self.d = ImageDraw.Draw(self.img)
self.bar_redraw = True
2020-07-05 17:45:45 +02:00
def render(self):
print("render NYI")
def animate(self, ticks):
print("animate NYI")
def get_image(self):
2020-07-19 18:39:01 +02:00
temp = self.img
return temp.rotate(-90,expand=True).tobytes()
2020-07-05 17:45:45 +02:00
class display_playback(display):
def __init__(self):
super().__init__()
self.title = ""
self.title_scroll = 0
self.title_scroll_max = -1
self.info = ""
self.syms = [];
self.title_img = Image.new('1', (0, 0), color = not self.fg)
2020-07-05 17:45:45 +02:00
def update_text(self, title):
self.title = title
w,h = self.d.textsize(self.title, font=self.font_title)
2020-07-13 18:10:48 +02:00
if (w > self.width): # Check the text is short enough to be static
self.title += "."
w,h = self.d.textsize(self.title, font=self.font_title)
w += 10 # Spacing between repetitions
self.title_scroll_max = w
2020-07-05 17:45:45 +02:00
self.title_scroll = self.title_scroll_max - 10
# Prerender
2020-07-13 18:10:48 +02:00
fw = w + self.width
self.title_img = Image.new('1', (fw, h), color = not self.fg)
font_shift = 5
ImageDraw.Draw(self.title_img).text((0,-font_shift), self.title, font=self.font_title, fill=self.fg)
2020-07-13 18:10:48 +02:00
self.title_img.paste(self.title_img, (w,0))
self.title_img = self.title_img.crop((0,0,fw,h-font_shift))
2020-07-05 17:45:45 +02:00
else:
self.title_scroll = 0
self.title_scroll_max = -1
def update_info(self, info):
self.info = info
self.bar_redraw = True
2020-07-05 17:45:45 +02:00
def update_syms(self, syms):
self.syms = syms
self.bar_redraw = True
2020-07-05 17:45:45 +02:00
def scroll_title(self, amount = 1):
if (self.title_scroll >= 0):
2020-07-05 17:45:45 +02:00
self.title_scroll = (self.title_scroll + amount) % self.title_scroll_max
def animate(self, ticks):
self.scroll_title(ticks)
def _render_bar(self):
2020-07-05 17:45:45 +02:00
# Clear framebuffer
self.d.rectangle((0,self.height - 1 - 10,self.width,self.height), fill = self.fg)
2020-07-05 17:45:45 +02:00
# Playback info
w,h = self.d.textsize(self.info, font=self.font_info)
2020-07-05 19:42:30 +02:00
self.d.text((self.width-w-2, self.height - 1 - 10 - 1), self.info, font=self.font_info, fill=not self.fg)
2020-07-05 17:45:45 +02:00
# Status icons
status = "".join([" " + icon + " " for icon in self.syms])
2020-07-05 19:42:30 +02:00
self.d.text((-2, self.height - 1 - 10 - 1), status, font=self.font_info, fill=not self.fg)
def render(self):
# Title (prerendered bitmap to shift)
self.img.paste(self.title_img, (-self.title_scroll,0))
# Other things were pre-rendered to the framebuffer directly
if (self.bar_redraw):
self.bar_redraw = False
self._render_bar()
2020-07-05 17:45:45 +02:00
# TODO: Maybe a progress bar?
#self.d.line((0,15, 127,15), fill=self.fg)
class display_manager(object):
def __init__(self):
self.tick = 0
self.current_display = None
def set_display(self, disp):
self.current_display = disp
def render(self):
if (self.current_display):
self.current_display.render()
def get_image(self):
if (self.current_display):
2020-07-05 20:09:31 +02:00
return self.current_display.get_image()
2020-07-05 17:45:45 +02:00
else:
2020-07-05 20:09:31 +02:00
return []
2020-07-05 17:45:45 +02:00
def animate(self, ticks):
old_tick = self.tick
self.tick += ticks
if (self.current_display):
if (not hasattr(self.current_display, 'ltick')):
self.current_display.ltick = old_tick
self.current_display.animate(self.tick - self.current_display.ltick)
self.current_display.ltick = self.tick
# TODO: Consider also animating inactive screens... Might resolve some glitches when updating text while a screen is in the background (or pass `tick` to update funcs)
class radioserv:
def handle_rx(self, rdata):
# Handle key presses
key_id = rdata[0] & 0x0F
key_press_short = ((rdata[0] & 0xC0) == 128)
if (key_press_short):
print("Press key: {}".format(key_id))
if (key_id == 0): # Tune-
pass
elif (key_id == 1): # Tune+
self.mpd.previous()
elif (key_id == 2): # Power
self.kill_event.set()
elif (key_id == 4): # Wheel
self.mpd.pause()
elif (key_id == 5): # Source
pass
elif (key_id == 6): # Info / Menu
self.mpd.next()
elif (key_id == 7): # PRESET
pass
# Handle rotary input
rotary_delta = int(rdata[1])# - 128
if (rotary_delta > 0):
print("Rotary: {}".format(rdata[2]))
# self.mpd.volume(rotary_delta)
#print("Battery: {}".format(rdata[2]))
2020-07-05 17:45:45 +02:00
async def io_main(self):
while True:
self.disp_mgr.animate(1)
self.disp_mgr.render()
rdata = self.spi.xfer([100, 101, 102], 50000, 1000)
self.handle_rx(rdata)
2020-07-19 18:39:01 +02:00
data = list(self.disp_mgr.get_image())
rdata = self.spi.xfer(data, SPI_SPEED, 10000)
2020-07-05 17:45:45 +02:00
await self.io_event.wait()
self.io_event.clear()
async def io_timer_main(self):
while True:
await asyncio.sleep(IO_DELAY)
self.io_event.set()
async def mpd_main(self):
last_title = ""
last_state = ""
last_info = ""
play_icon_map = {"play" : "▶️", "pause" : "ll", "stop" : ""}
while True:
# Query MPD
status = self.mpd.status()
currentsong = self.mpd.currentsong()
# Song title
try:
2020-07-19 18:37:48 +02:00
title = currentsong['title']
try:
title += " by " + currentsong['artist']
except:
pass
2020-07-05 17:45:45 +02:00
except:
try:
title = currentsong['file'];
except:
title = ""
if (title != last_title):
self.disp_playback.update_text(title)
last_title = title
# Playback state (icons)
if (status['state'] != last_state):
icons = []
icons.append(play_icon_map.get(status["state"], "?"))
self.disp_playback.update_syms(icons)
last_state = status["state"]
# Playback info
try:
cm, cs = divmod(round(float(status['elapsed'])), 60)
2020-07-19 18:38:06 +02:00
if "duration" in status:
dm, ds = divmod(round(float(status['duration'])), 60)
info = "#%d/%d %d:%02d/%d:%02d" % (int(status['song']) + 1, int(status['playlistlength']), cm, cs, dm, ds)
else:
info = "Radio %02d:%02d" % (cm, cs)
2020-07-05 17:45:45 +02:00
except:
info = "Stopped"
if (info != last_info):
self.disp_playback.update_info(info)
# Sadly I can't use mpd.idle() here
await asyncio.sleep(MPD_DELAY)
async def main(self):
2020-07-19 18:39:01 +02:00
# Connect to SPI
self.spi = spidev.SpiDev()
self.spi.open(0, 0)
self.spi.max_speed_hz = SPI_SPEED
2020-07-19 18:39:01 +02:00
self.spi.mode = 0b00
2020-07-05 17:45:45 +02:00
# Connect to MPD
self.mpd = MPDClient()
2020-08-08 16:53:11 +02:00
self.mpd.connect("localhost", 6600)
2020-07-05 17:45:45 +02:00
# Set up displays
self.disp_playback = display_playback()
self.disp_playback.update_text("Loading...")
self.disp_mgr = display_manager()
self.disp_mgr.set_display(self.disp_playback)
# Create main tasks
self.io_event = asyncio.Event()
self.kill_event = asyncio.Event()
2020-07-05 17:45:45 +02:00
io_task = asyncio.create_task(self.io_main())
io_timer_task = asyncio.create_task(self.io_timer_main())
mpd_task = asyncio.create_task(self.mpd_main())
await self.kill_event.wait()
self.kill_event.clear()
2020-07-05 17:45:45 +02:00
io_timer_task.cancel()
io_task.cancel()
mpd_task.cancel()
self.mpd.close()
self.mpd.disconnect()
def __init__(self):
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
try:
loop.run_until_complete(self.main())
finally:
loop.run_until_complete(loop.shutdown_asyncgens())
loop.close()
r = radioserv()