#!/usr/bin/env python from PIL import Image, ImageDraw, ImageFont import asyncio from mpd import MPDClient import spidev import math import traceback from aiohttp import web import datetime IO_DELAY = (1.0 / 18.0) MPD_DELAY = 0.5 SPI_SPEED = 500000 IO_SPEED = 50000 LIVE_SELECT = 160 LIVE_VOLUME = 40 LIKE_FILE = "/var/db/wr_likes" 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) h = 20 # Force standard height to avoid artifacting 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) h = 26 # Force standard height to avoid artifacting 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_selection(display): def __init__(self): super().__init__() self.entries = [] self.page = 0 def animate(self, ticks): pass def render(self): #self.img.paste(self.img_pre, (0,0)) self.img = self.img_pre def set_entries(self, entries, left_only = False): self.left_only = left_only self.entries = entries self.set_page(0) def set_page(self, page): self.page = page font_shift = 0 x = 0 y = 0 self.img_pre = Image.new('1', (self.width, self.height), color = not self.fg) mult = 3 if self.left_only else 6 for entry in self.entries[page*mult:]: ImageDraw.Draw(self.img_pre).text((x,y-font_shift), entry, font=self.font_info, fill=self.fg) y += self.height / 3 if (y >= self.height): y = 0 x += self.width / 2 if (x >= self.width) or (self.left_only): break def set_page_relative(self, delta): self.page += delta max_page = math.ceil(len(self.entries) / 6) - 1 if (self.page < 0): self.page = 0 elif (self.page > max_page): self.page = max_page self.set_page(self.page) def get_entry(self, offset, page = -1): if (page == -1): page = self.page try: mult = 3 if self.left_only else 6 return self.entries[page * mult + offset] except: return "" 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 set_live(self, live = -1): self.current_display.live = live def kill_display(self): self.current_display.live = 0 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 load_liked(self): try: f = open(LIKE_FILE, "r") text = f.read() if text != "": self.liked = text.split("\n"); else: self.liked = [] f.close(); except: self.liked = [] def like_song(self): self.liked.append("[{}] {}".format(datetime.datetime.now(), self.get_formatted_title())) self.disp_info.update_text("♥", True) self.disp_info.update_info("Liked song.") self.disp_mgr.set_display(self.disp_info, LIVE_VOLUME) self.sync_liked() def unlike_song(self, id): self.liked.pop(id) self.sync_liked() def sync_liked(self): f = open(LIKE_FILE, "w") f.write("\n".join(self.liked)) f.close() def handle_volume(self, rotary_delta, bat): if (rotary_delta != 0): nvol = int(self.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("Battery: {}%".format(bat)) self.disp_mgr.set_display(self.disp_info, LIVE_VOLUME) self.mpd.setvol(nvol) def open_playlist_screen(self): playlists = self.mpd.listplaylists() entries = [] for entry in playlists: entries.append(entry['playlist']) self.disp_select.mode = "playlist" self.disp_select.set_entries(entries) self.disp_mgr.set_display(self.disp_select, LIVE_SELECT) def open_directory_screen(self): dirs = self.mpd.listfiles("") entries = [] for entry in dirs: try: entries.append(entry['directory']) except: pass self.disp_select.mode = "directory" self.disp_select.set_entries(entries, True) self.disp_mgr.set_display(self.disp_select, LIVE_SELECT) def get_formatted_title(self): try: title = self.currentsong['title'] try: title += " by " + self.currentsong['artist'] except: pass except: try: title = self.currentsong['file'] except: title = " " return title def handle_rx(self, rdata): # Handle battery voltage bat = int((int(rdata[2]) - 175) / 45 * 100) if (bat < 0): bat = 0 elif (bat > 100): bat = 100 if (bat == 0): print("Battery depleted. Shutting down.") self.kill_event.set() # Handle key presses key_id = rdata[0] & 0x0F key_press_short = ((rdata[0] & 0xC0) == 128) key_press_long = ((rdata[0] & 0xC0) == 192) if key_press_short: print("Press key: {}".format(key_id)) elif key_press_long: print("Long press key: {}".format(key_id)) # Handle rotary input rotary_delta = int(rdata[1] & 0x1F) - 16 if (self.disp_mgr.current_display == self.disp_playback): if (key_press_short): if (key_id == 0): # Tune- self.like_song() elif (key_id == 1): # Tune+ self.mpd.previous() elif (key_id == 2): # Power self.kill_event.set() elif (key_id == 4): # Wheel if (self.status['state'] == "stop"): self.mpd.play(0) else: self.mpd.pause() elif (key_id == 5): # Source self.open_playlist_screen() elif (key_id == 6): # Info / Menu self.mpd.next() elif (key_id == 7): # PRESET self.mpd.random(int(not int(self.status['random']))) if (key_press_long): if (key_id == 5): # Source self.open_directory_screen() self.handle_volume(rotary_delta, bat) elif (self.disp_mgr.current_display == self.disp_info): self.handle_volume(rotary_delta, bat) if (key_press_short): self.disp_mgr.kill_display() elif (self.disp_mgr.current_display == self.disp_select): if (key_press_short) or (key_press_long): sel = -1 if (key_id == 0): # Tune- sel = 2 elif (key_id == 1): # Tune+ sel = 1 elif (key_id == 2): # Power sel = 0 elif (key_id == 4): # Wheel sel = 100 elif (key_id == 5): # Source sel = 3 elif (key_id == 6): # Info / Menu sel = 4 elif (key_id == 7): # PRESET sel = 5 if (sel == -1): pass elif (sel == 100): self.disp_mgr.kill_display() else: print("Load {} {}.".format(self.disp_select.mode, self.disp_select.get_entry(sel))) if key_press_short: self.mpd.clear() if (self.disp_select.mode == "playlist"): self.mpd.load(self.disp_select.get_entry(sel)) else: self.mpd.add(self.disp_select.get_entry(sel)) if key_press_short: self.mpd.play(0) self.disp_mgr.kill_display() if (rotary_delta != 0): self.disp_mgr.set_live(LIVE_SELECT) self.disp_select.set_page_relative(rotary_delta) async def io_main(self): while True: try: 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], IO_SPEED, 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() except: print("Error in io_main") traceback.print_exc() async def io_timer_main(self): while True: await asyncio.sleep(IO_DELAY) self.io_event.set() async def mpd_main(self): last_title = "Loading..." last_state = "" last_info = "" last_random = -1 play_icon_map = {"play" : "▶️", "pause" : "ll", "stop" : "□"} while True: # Query MPD status = self.mpd.status() currentsong = self.mpd.currentsong() self.status = status self.currentsong = currentsong # Song title title = self.get_formatted_title() if (title != last_title): self.disp_playback.update_text(title) last_title = title # Playback state (icons) if (status['state'] != last_state or status['random'] != last_random): icons = [] icons.append(play_icon_map.get(status["state"], "?")) if (status['random'] == '1'): icons.append("Z") self.disp_playback.update_syms(icons) last_state = status["state"] last_random = status["random"] # 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 webserver_main(self): print("ws main") routes = web.RouteTableDef() @routes.get('/') @routes.post('/') async def hello(request): data = await request.post() if (len(data) > 0): for key in data: print(key[4:]) self.unlike_song(int(key[4:])) slist = "" i = 0 for entry in self.liked: slist += '
  • {}
  • '.format(entry, i) i += 1 if slist == "": slist = "There are no liked songs. Push the like-button to add them to the list." text = """ Web Radio: Liked Songs

    Web Radio: Liked Songs

    {}

    Refresh

    """.format(slist); return web.Response(text=text, headers=({'CONTENT-TYPE': "text/html"})) app = web.Application() app.add_routes(routes) await web._run_app(app, host="*", port=80) 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() retry = 15 while retry > 0: retry -= 1 try: self.mpd.connect("localhost", 6600) break except: print("Error connecting to MPD. Retrying in 1s.") await asyncio.sleep(1) try: self.mpd.status() except: print("Could not connect to server. Exiting.") return -1 # 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_select = display_selection() self.disp_mgr = display_manager() self.disp_mgr.set_display(self.disp_playback) # Flush I/O messages rdata = self.spi.xfer([0, 0, 0], IO_SPEED, 1000) # 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()) ws_task = asyncio.create_task(self.webserver_main()) await self.kill_event.wait() self.kill_event.clear() # Show shutdown screen self.disp_info.update_text("Shutting down", True); self.disp_info.update_info(" ") self.disp_mgr.set_display(self.disp_info) await asyncio.sleep(0.1) io_timer_task.cancel() io_task.cancel() mpd_task.cancel() self.mpd.close() self.mpd.disconnect() def __init__(self): self.loop = asyncio.new_event_loop() asyncio.set_event_loop(self.loop) self.load_liked() try: self.loop.run_until_complete(self.main()) finally: self.loop.run_until_complete(self.loop.shutdown_asyncgens()) self.loop.close() r = radioserv()