Home Assistant AppDaemon plugin to make dumb amplifier smart (SMSL Q5Pro)

amplifier_manager.py

import appdaemon.plugins.hass.hassapi as hass
from time import sleep
from threading import Lock

#
# SMSL Q5 Pro (Amplifier Manager)
# Amplifiers like the SMSL Q5 Pro can with this App be controlled with sliders for
# volume, bass and treble and an input_select for the input source
#
# Expected (configuration.yaml): 
# 
# input_select:
#   amp_source:
#     name: Amplifier Source
#     options:
#       - Analog
#       - USB
#       - Optical
#       - Coax
#
# input_number:
#   amp_volume:
#     name: Amplifier Volume
#     min: 1
#     max: 30
#   amp_bass:
#     name: Amplifier Bass
#     min: -9
#     max: 9
#   amp_treble:
#     name: Amplifier Treble
#     min: -9
#     max: 9
#
# Args:
#
# TODO: Make plugi more generic
# TODO: Remove duplication

class AmplifierManager(hass.Hass):

  def initialize(self):
    self.log("Amplifier Manager has started.")
    self.listen_state(self.onAmplifierSwitchChange, "switch.stereo_amp_switch")
    self.listen_state(self.onVolumeChange, "input_number.amp_volume")
    self.listen_state(self.onBassChange, "input_number.amp_bass")
    self.listen_state(self.onTrebleChange, "input_number.amp_treble")
    self.listen_state(self.onSourceChange, "input_select.amp_source")
    self.sources = ["USB", "Optical", "Coax", "Analog"] # Order of these are important!
    self.lock = Lock()


  # Volume on this amplifier only saves volume level between 5 and 20
  # So it will reset to either on poweron so this automation will reflect this strange behaviour on the input_number
  def onAmplifierSwitchChange(self, entity, attribute, old, new, kwargs):
    if old == new:
      return
    if old == 'off' and new == 'on':
      # TODO: Restore volume on poweron?
      return
    current_volume = float(self.entities.input_number.amp_volume.state)
    self.log("Amplifier powered OFF")
    if current_volume < 5:
      self.setVolume(5)
    elif current_volume > 20:
      self.setVolume(20)

  def onSourceChange(self, entity, attribute, old, new, kwargs):
    if old == new:
      return
    if old == "":
      old = "Optical"
      return
    steps = self.calculateSteps(old, new)
    self.log("Old source: " + old + " changed to: " + new)
    self.lock.acquire()
    for _ in range(0, steps):
      self.sendInput()
      sleep(1)
    self.lock.release()

  def onVolumeChange(self, entity, attribute, old, new, kwargs):
    if old == new:
      return
    if old == "":
      old = 5
      return
    difference = abs(int((float(old)) - (float(new))))
    increase = int(float(new)) > int(float(old))
    self.log("Old volume: " + old + " changed to: " + new)
    self.lock.acquire()
    for _ in range(0, difference):
      if increase:
         self.sendUp()
      else:
         self.sendDown()
      sleep(0.25)
    self.lock.release()

  def onBassChange(self, entity, attribute, old, new, kwargs):
    if old == new:
      return
    if old == "":
      old = 0
      return
    difference = abs(int((float(old)) - (float(new))))
    increase = int(float(new)) > int(float(old))
    self.log("Old bass level: " + old + " changed to: " + new)
    self.lock.acquire()
    self.sendTone()
    for _ in range(0, difference):
      if increase:
          self.sendUp()
      else:
          self.sendDown()
      sleep(0.25)
    self.sendTone()
    self.sendTone()
    self.lock.release()

  def onTrebleChange(self, entity, attribute, old, new, kwargs):
    if old == new:
      return
    if old == "":
      old = 0
      return
    difference = abs(int((float(old)) - (float(new))))
    increase = int(float(new)) > int(float(old))
    self.log("Old treble level: " + old + " changed to: " + new)
    self.lock.acquire()
    self.sendTone()
    self.sendTone()
    for _ in range(0, difference):
      if increase:
          self.sendUp()
      else:
          self.sendDown()
      sleep(0.25)
    self.sendTone()
    self.lock.release()

  def setVolume(self, volume):
    self.call_service("input_number/set_value", entity_id = "input_number.amp_volume", value = volume)

  def sendTone(self):
    self.call_service("remote/send_command", entity_id = "remote.amp", command = "tone")

  def sendInput(self):
    self.call_service("remote/send_command", entity_id = "remote.amp", command = "input")

  def sendUp(self):
    self.call_service("remote/send_command", entity_id = "remote.amp", command = "up")

  def sendDown(self):
    self.call_service("remote/send_command", entity_id = "remote.amp", command = "down")

  # Calculates the steps (indexes) needed to get to the wanted index
  def calculateSteps(self, current, wanted):
    currentIndex = self.sources.index(current) + 1
    wantedIndex = self.sources.index(wanted) + 1

    if (wantedIndex < currentIndex):
      return ((len(self.sources) - currentIndex) + wantedIndex)
    elif (wantedIndex > currentIndex):
      return (wantedIndex - currentIndex)
    
    return 0

Hey, I just found this and think it could be adapted to work with my own amplifier and custom remote :D

How exactly do I set the AppDeamon? And could you share a bit more info about how the daemon interfaces with the remote?

Hey, I just found this and think it could be adapted to work with my own amplifier and custom remote :D

How exactly do I set the AppDeamon? And could you share a bit more info about how the daemon interfaces with the remote?

I switched to scripts instead. I hope this helps: https://gist.github.com/tillbaks/4ece10575ce017e0ba68e725361472ab