webradio/software/webradio.py

#!/usr/bin/env python

from PIL import Image, ImageDraw, ImageFont
import asyncio
from mpd import MPDClient
import spidev

IO_DELAY = (1.0 / 20.0)
MPD_DELAY = 0.5

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

	def render(self):
		print("render NYI")

	def animate(self, ticks):
		print("animate NYI")

	def get_image(self):
		temp = self.img
		return temp.rotate(-90,expand=True).tobytes()

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)

	def update_text(self, title):
		self.title = title

		w,h = self.d.textsize(self.title, font=self.font_title)
		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
			self.title_scroll = self.title_scroll_max - 10
			# Prerender
			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)
			self.title_img.paste(self.title_img, (w,0))
			self.title_img = self.title_img.crop((0,0,fw,h-font_shift))
		else:
			self.title_scroll = 0
			self.title_scroll_max = -1

	def update_info(self, info):
		self.info = info
		self.bar_redraw = True

	def update_syms(self, syms):
		self.syms = syms
		self.bar_redraw = True

	def scroll_title(self, amount = 1):
		if (self.title_scroll >= 0):
			self.title_scroll = (self.title_scroll + amount) % self.title_scroll_max

	def animate(self, ticks):
		self.scroll_title(ticks)

	def _render_bar(self):
		# Clear framebuffer
		self.d.rectangle((0,self.height - 1 - 10,self.width,self.height), fill = self.fg)

		# Playback info
		w,h = self.d.textsize(self.info, font=self.font_info)
		self.d.text((self.width-w-2, self.height - 1 - 10 - 1), self.info, font=self.font_info, fill=not self.fg)

		# Status icons
		status = "".join([" " + icon + " " for icon in self.syms])
		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()

		# 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):
			return self.current_display.get_image()
		else:
			return []

	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:
	async def io_main(self):
		while True:
			self.disp_mgr.animate(1)
			self.disp_mgr.render()
			data = list(self.disp_mgr.get_image())
			self.spi.xfer(data)
			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:
				title = currentsong['title']
				try:
					title += " by " + currentsong['artist']
				except:
					pass
			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)
				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)
			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):
		# Connect to SPI
		self.spi = spidev.SpiDev()
		self.spi.open(0, 0)
		self.spi.max_speed_hz = 500000
		self.spi.mode = 0b00

		# Connect to MPD
		self.mpd = MPDClient()
		self.mpd.connect("yuki", 6600)

		# 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()

		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())

		# TODO: Wait for exit condition (shutdown command)
		await asyncio.sleep(60)

		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()