#!/usr/bin/env python from PIL import Image, ImageDraw, ImageFont import asyncio from mpd import MPDClient import spidev IO_DELAY = (1.0 / 25.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()