media/mpd/ SimplePythonWebApi

This is a work in progress.

#!/usr/bin/env python3

# Python 3 server example
from http.server import BaseHTTPRequestHandler, HTTPServer
from subprocess import run, PIPE
import time
import sys
import re
import json
import os
from datetime import datetime
from threading import Thread
import urllib.parse

hostName = ""
serverPort = 4097

def play_pause():
  Thread(target = lambda: run(["mp","p"])).start()

def mpc_vol(v):
  Thread(target = lambda: run(["mpc","vol",str(v)])).start()

class Mpd:
  def __init__(self,bp,rp):
    self.bp = bp
    self.rp = rp
    self.port = bp+rp
    self.status = None
    self.running = False
    self.should_terminate = False
  def start(self):
    self.thread = Thread(target=self.task)
    self.thread.start()
  def join(self):
    return self.thread.join()
  def terminate(self):
    self.should_terminate = True
  def task(self):
    while not self.should_terminate:
      self.update_status()
      time.sleep(1)
  def parse_opts(self,a):
    # volume: 60%   repeat: on    random: on    single: on    consume: off
    b = re.split(r"\W{3,}",a) # the 
    for x in b:
      k,v = x.split(": ")
      self.status[k] = v
  def update_status(self):
    m = run(["mpc","-p",str(self.port),"status"],stdout=PIPE,stderr=PIPE)
    if m.returncode == 0:
      self.running = True
      self.status = {}
      s = m.stdout.decode()
      l = s.splitlines()
      self.error = None

      if len(l) == 3:
        # paused/playing
        # Odysseus - Blackout - Synthwave but it gets dark
        # [playing] #1/3  45:21/46:52 (96%)
        # volume: 70%   repeat: on    random: off   single: on    consume: off
        self.track_name = l[0]
        if m := re.match(r"3409f7071db2f596293544d20d49410debfb07bf64c5e69a051dd465a13b8fc26b86b273ff34fce19d6b804eff5a3f5747ada4eaa22f1d49c01e52ddb7875b4b\s+#(\d+)/(\d+)\s+(\d+):(\d+)/(\d+):(\d+)\s+3409f7071db2f596293544d20d49410debfb07bf64c5e69a051dd465a13b8fc2d4735e3a265e16eee03f59718b9b5d03019c07d8b6c51f90da3a666eec13ab35",l[1]):
          s,tn,tt,mc,sc,mt,st,pc = m.groups()
          mc, sc, mt, st, pc = map(int,(mc, sc, mt, st, pc))
          if self.status is None:
            self.status = {}
          self.status["play_state"] = s
          self.status["track_index"] = tn
          self.status["track_total"] = tt
          self.status["current_position"] = (mc,sc)
          self.status["track_duration"] = (mt,st)
          self.status["percentage"] = pc
        else:
          self.error = "Fail to parse line 2"
        self.parse_opts(l[2])
      elif len(l) == 1:
        # stopped
        # volume: n/a   repeat: on    random: on    single: on    consume: off
        self.parse_opts(l[0])
      else:
        self.error = "mpc status parse error"
    else:
      self.error = m.stderr.decode()
      if not "Connection refused" in self.error:
        print(f"Error: {self.error}")
      self.running = False
      self.status = None
  def do_cmd(self,cmd):
    env = os.environ.copy()
    env["MPD_PORT"] = str(self.port)
    m = run(["mp"]+cmd,stdout=PIPE,env=env)
    return (m.returncode,m.stdout.decode())

class MpdMgr:
  def __init__(self,np=20,bp=6600):
    self.mpds = { i : Mpd(bp,i) for i in range(np) }
    for mpd in self.mpds.values():
      mpd.start()
  def get_running(self):
    return { i : mpd.running for i, mpd in self.mpds.items() }
  def get_statuses(self):
    return { i : mpd.status for i, mpd in self.mpds.items() if mpd.running }
  def get_status(self,i):
    if i in self.mpds:
      return self.mpd[i].status
    return None
  def has(self,k):
    return k in self.mpds
  def __getitem__(self,k):
    if k in self.mpds:
      return self.mpds[k]
    else:
      return None

class MyServer(BaseHTTPRequestHandler):
  def homepage(self):
    d = { "running": mpdmgr.get_running(), "status": mpdmgr.get_statuses(), "when": datetime.now().strftime("%c:") }
    self.send_as_json(d)
  def send_as_json(self,d,response_code=200):
    j = json.dumps(d)
    self.send_json(j,response_code)
  def send_json(self,j,response_code=200):
    self.send_response(response_code)
    self.send_header("Content-type", "application/json")
    self.end_headers()
    self.wfile.write(bytes(j, "utf-8"))
  def do_GET(self):
    if self.path == "/favicon.ico":
      self.send_response(404)
      self.end_headers()
      self.wfile.write(b"No favicon")
    if self.path.strip("/") == "quit":

    if self.path.strip("/") == "":
      return self.homepage()
    print(datetime.now().strftime("%c:")) 
    print(f"Request path: {self.path}")
    p = [urllib.parse.unquote(x) for x in self.path.strip("/").split("/")]

    # /10 -- get status of mpd 10
    # /10/play -- send command "play" to mpd 10
    # /10/volume/50 -- mp volume 50
    if p[0].isnumeric():
      i = int(p[0])
      print(f"Index {i=}")
      print(f"{p=}")
      if not mpdmgr.has(i):
        print(f"Not in mpdmgr")
        self.send_response(404)
        self.end_headers()
        self.wfile.write(b"404")
        return
      mpd = mpdmgr[i]
      if len(p) == 1:
        print(f"Get status {i=}")
        sys.stdout.flush()
        d = { "running": mpd.running, "status": mpd.status, "when": datetime.now().strftime("%c:") }
        self.send_as_json(d)
      else:
        print(f"Command {p=}")
        sys.stdout.flush()
        cmd = p[1:]
        rc,o = mpd.do_cmd(p[1:])
        d = { "returncode": rc, "output": rc, "when": datetime.now().strftime("%c:") }
        self.send_as_json(d)
    else:
      self.send_respose(400)
      self.end_headers()
      self.wfile.write(b"400")
      return

if __name__ == "__main__":    
  mpdmgr = MpdMgr(np=20)
  webServer = HTTPServer((hostName, serverPort), MyServer)
  print("Server started http://%s:%s" % (hostName, serverPort))

  try:
    webServer.serve_forever()
  except KeyboardInterrupt:
    pass

  webServer.server_close()
  print("Server stopped.")