#!/usr/bin/env python from PIL import Image, ImageDraw, ImageFont import asyncio from mpd import MPDClient import spidev IO_DELAY = (1.0 / 18.0) MPD_DELAY = 0.5 SPI_SPEED = 500000 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) 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) self.bar_redraw = True def update_text(self, title, center=False): self.title = title font_shift = 5 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) 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: if (center): x_shift = (self.width - w) / 2 else: x_shift = 0; w = self.width self.title_img = Image.new('1', (w, h), color = not self.fg) ImageDraw.Draw(self.title_img).text((x_shift,-font_shift), self.title, font=self.font_title, fill=self.fg) 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, live=-1): self.current_display = disp self.current_display.live = live def display_is_dead(self): if (self.current_display): return (self.current_display.live == 0) return 0 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 if (self.current_display.live > 0): self.current_display.live -= ticks # 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]) if (rotary_delta > 127): rotary_delta -= 256 if (rotary_delta != 0): print("Rotary: {}".format(rotary_delta)) nvol = int(self.mpd.status()['volume']) + (rotary_delta * 4) if (nvol < 0): nvol = 0; elif (nvol > 100): nvol = 100 self.disp_info.update_text("{}%".format(nvol), True) self.disp_info.update_info("Volume") self.disp_mgr.set_display(self.disp_info, 40) self.mpd.setvol(nvol) #print("Battery: {}".format(rdata[2])) async def io_main(self): while True: if self.disp_mgr.display_is_dead(): self.disp_mgr.set_display(self.disp_playback) self.disp_mgr.animate(1) self.disp_mgr.render() rdata = self.spi.xfer([100, 101, 102], 50000, 1000) self.handle_rx(rdata) data = list(self.disp_mgr.get_image()) rdata = self.spi.xfer(data, SPI_SPEED, 10000) 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 = SPI_SPEED self.spi.mode = 0b00 # Connect to MPD self.mpd = MPDClient() self.mpd.connect("localhost", 6600) # Set up displays self.disp_playback = display_playback() self.disp_playback.update_text("Loading...") self.disp_info = display_playback() self.disp_info.update_text("---"); 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() 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() 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()