#!/usr/bin/env python from PIL import Image, ImageDraw, ImageFont import asyncio from mpd import MPDClient IO_DELAY = (1.0 / 20.0) MPD_DELAY = 1 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.dbg = 0 def render(self): print("render NYI") def get_image(self): print("get_image NYI") def animate(self, ticks): print("animate NYI") def get_image(self): #self.img.save('pil_text_font.png') #self.img.save("/tmp/wr/frame%d.png" % self.dbg) print(str(len(self.img.tobytes()))) self.dbg += 1 class display_playback(display): def __init__(self): super().__init__() self.title = "" self.title_scroll = 0 self.title_scroll_max = -1 self.info = "" self.syms = []; def update_text(self, title): self.title = title w,h = self.d.textsize(self.title, font=self.font_title) if (w > self.width): self.title += ". " # Wraparound for scrolling (pt1) self.title_scroll_max,h = self.d.textsize(self.title, font=self.font_title) self.title_scroll = self.title_scroll_max - 10 self.title += self.title # Wraparound for scrolling (pt2) else: self.title_scroll = 0 self.title_scroll_max = -1 def update_info(self, info): self.info = info def update_syms(self, syms): self.syms = syms; 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(self): # Clear framebuffer self.d.rectangle((0,0,self.width,self.height), fill = not self.fg) # Title self.d.text((-self.title_scroll,-5), self.title, font=self.font_title, fill=self.fg) # Playback info w,h = self.d.textsize(self.info, font=self.font_info) self.d.text((self.width-w, self.height - 1 - 10 - 1), self.info, font=self.font_info, fill=self.fg) # Status icons status = "" status = status.join([" " + icon + " " for icon in self.syms]) self.d.text((-1, self.height - 1 - 10 - 1), status, font=self.font_info, fill=self.fg) # TODO: Maybe a progress bar? #self.d.line((0,15, 127,15), fill=self.fg) # Serialize # TODO: Serialize 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): self.current_display.get_image() else: print("TODO: return empty image") 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() self.disp_mgr.get_image() print("x") 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'] + " by " + currentsong['artist'] 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) 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) 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 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(10) 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()