Program that tracks Dreambox's BER or SNR and emits MIDI notes

sat-align.py

#!/usr/bin/python
import alsaseq, alsamidi, requests, time, math
from lxml import etree

INTERVAL = 70e-3
URL = "http://root:dreamboxx@10.139.205.104/xml/streaminfo"
alsaseq.client('sat-align', 0, 1, True)

map = lambda x, s, e: s + x * (e-s)
clamp = lambda x, s, e: max(min(x, e), s)

def process(info, note):
  if info["sync"] and info["lock"]:
    #n = map(3*info["snr"]-2, 40, 80)
    n = map(-math.log10(info["ber"]+1)/5+1, 40, 80)
    n = clamp(n, 20, 100)
    note(n)

old_notes = []
new_notes = []
session = requests.Session()
alsaseq.start()
while True:
  for note in old_notes:
    if note not in new_notes:
      alsaseq.output(alsamidi.noteoffevent(note[0], note[1], note[2]))
  for note in new_notes:
    if note not in old_notes:
      alsaseq.output(alsamidi.noteonevent(note[0], note[1], note[2]))
  old_notes = new_notes
  new_notes = []
  time.sleep(INTERVAL)

  r = session.get(URL)
  if r.status_code != 200:
    print "Error, server replied", r.status_code
    continue
  root = etree.fromstring(r.content)
  def percentage(x):
    assert x.endswith("%")
    return int(x[:-1]) / 100.0
  boolean = lambda x: { "Yes":True, "No":False }[x]
  info = {
    "snr": percentage(root.xpath("snr")[0].text),
    "agc": percentage(root.xpath("agc")[0].text),
    "ber": int(root.xpath("ber")[0].text),
    "lock": boolean(root.xpath("lock")[0].text),
    "sync": boolean(root.xpath("sync")[0].text),
  }
  print info

  def note(n, v=127, p=0):
    new_notes.append((p,int(round(n)),v))
  process(info, note)