Add airplay volume synchronizer

Author: SteveMicroNova

amplipi/streams/airplay.py

@@ -7,6 +7,9 @@
 import time
 import os
 import io
+import sys
+import threading
+import json
 
 
 def write_sp_config_file(filename, config):
@@ -42,6 +45,29 @@ def __init__(self, name: str, ap2: bool, disabled: bool = False, mock: bool = Fa
     self._connect_time = 0.0
     self._coverart_dir = ''
     self._log_file: Optional[io.TextIOBase] = None
+    self.src_config_folder: Optional[str] = None
+    self.volume_watcher_process: Optional[threading.Thread] = None  # Populates the fifo that the vol sync script depends on
+    self.volume_sync_process: Optional[subprocess.Popen] = None
+    self._volume_fifo: Optional[str] = None
+
+  def watch_vol(self):
+    """Creates and supplies a FIFO with volume data for volume sync"""
+    while True:
+      try:
+        if self.src is not None:
+          if self._volume_fifo is None and self.src_config_folder is not None:
+            fifo_path = f"{self.src_config_folder}/vol"
+            if not os.path.isfile(fifo_path):
+              os.mkfifo(fifo_path)
+            self._volume_fifo = os.open(fifo_path, os.O_WRONLY, os.O_NONBLOCK)
+          data = json.dumps({
+            'zones': self.connected_zones,
+            'volume': self.volume,
+          })
+          os.write(self._volume_fifo, bytearray(f"{data}\r\n", encoding="utf8"))
+      except Exception as e:
+        logger.error(f"{self.name} volume thread ran into exception: {e}")
+      time.sleep(0.1)
 
   def reconfig(self, **kwargs):
     self.validate_stream(**kwargs)
@@ -71,9 +97,9 @@ def _activate(self, vsrc: int):
         logger.info(f'Another Airplay 2 stream is already in use, unable to start {self.name}, mocking connection')
         return
 
-    src_config_folder = f'{utils.get_folder("config")}/srcs/v{vsrc}'
+    self.src_config_folder = f'{utils.get_folder("config")}/srcs/v{vsrc}'
     try:
-      os.remove(f'{src_config_folder}/currentSong')
+      os.remove(f'{self.src_config_folder}/currentSong')
     except FileNotFoundError:
       pass
     self._connect_time = time.time()
@@ -86,9 +112,9 @@ def _activate(self, vsrc: int):
         'name': self.name,
         'port': 5100 + 100 * vsrc,  # Listen for service requests on this port
         'udp_port_base': 6101 + 100 * vsrc,  # start allocating UDP ports from this port number when needed
-        'drift': 2000,  # allow this number of frames of drift away from exact synchronisation before attempting to correct it
-        'resync_threshold': 0,  # a synchronisation error greater than this will cause resynchronisation; 0 disables it
-        'log_verbosity': 0,  # "0" means no debug verbosity, "3" is most verbose.
+        'drift_in_seconds': 2,  # allow this number of frames of drift away from exact synchronisation before attempting to correct it
+        'resync_threshold_in_seconds': 0,  # a synchronisation error greater than this will cause resynchronisation; 0 disables it
+        'log_verbosity': "diagnostics",  # "none" means no debug verbosity, "diagnostics" is most verbose.
         'mpris_service_bus': 'Session',
       },
       'metadata': {
@@ -99,7 +125,7 @@ def _activate(self, vsrc: int):
       'alsa': {
         'output_device': utils.virtual_output_device(vsrc),  # alsa output device
         # If set too small, buffer underflow occurs on low-powered machines. Too long and the response times with software mixer become annoying.
-        'audio_backend_buffer_desired_length': 11025
+        'audio_backend_buffer_desired_length': 11025,
       },
     }
 
@@ -109,10 +135,10 @@ def _activate(self, vsrc: int):
     except FileNotFoundError:
       pass
     os.makedirs(self._coverart_dir, exist_ok=True)
-    os.makedirs(src_config_folder, exist_ok=True)
-    config_file = f'{src_config_folder}/shairport.conf'
+    os.makedirs(self.src_config_folder, exist_ok=True)
+    config_file = f'{self.src_config_folder}/shairport.conf'
     write_sp_config_file(config_file, config)
-    self._log_file = open(f'{src_config_folder}/log', mode='w')
+    self._log_file = open(f'{self.src_config_folder}/log', mode='w')
     shairport_args = f"{utils.get_folder('streams')}/shairport-sync{'-ap2' if self.ap2 else ''} -c {config_file}".split(' ')
     logger.info(f'shairport_args: {shairport_args}')
 
@@ -125,7 +151,15 @@ def _activate(self, vsrc: int):
       # shairport sync only adds the pid to the mpris name if it cannot use the default name
       if len(os.popen("pgrep shairport-sync").read().strip().splitlines()) > 1:
         mpris_name += f".i{self.proc.pid}"
-      self.mpris = MPRIS(mpris_name, f'{src_config_folder}/metadata.txt')
+      self.mpris = MPRIS(mpris_name, f'{self.src_config_folder}/metadata.txt')
+
+      vol_sync = f"{utils.get_folder('streams')}/shairport_volume_handler.py"
+      vol_args = [sys.executable, vol_sync, mpris_name, f"{utils.get_folder('config')}/srcs/v{self.vsrc}"]
+
+      logger.info(f'{self.name}: starting vol synchronizer: {vol_args}')
+      self.volume_watcher_process = threading.Thread(target=self.watch_vol, daemon=True)
+      self.volume_watcher_process.start()
+      self.volume_sync_process = subprocess.Popen(args=vol_args, stdout=self._log_file, stderr=self._log_file)
     except Exception as exc:
       logger.exception(f'Error starting airplay MPRIS reader: {exc}')
 
@@ -135,12 +169,22 @@ def _deactivate(self):
     self.mpris = None
     if self._is_running():
       self.proc.stdin.close()
+
       logger.info('stopping shairport-sync')
       self.proc.terminate()
+      if self.volume_sync_process is not None:
+        self.volume_sync_process.terminate()
+
       if self.proc.wait(1) != 0:
         logger.info('killing shairport-sync')
         self.proc.kill()
       self.proc.communicate()
+
+      if self.volume_sync_process is not None:
+        if self.volume_sync_process.wait(1) != 0:
+          logger.info('killing shairport vol sync')
+          self.volume_sync_process.kill()
+
     if '_log_file' in self.__dir__() and self._log_file:
       self._log_file.close()
     if self.src:
@@ -149,7 +193,11 @@ def _deactivate(self):
       except Exception as e:
         logger.exception(f'Error removing airplay config files: {e}')
     self._disconnect()
+
     self.proc = None
+    self.volume_sync_process = None
+    self.volume_watcher_process = None
+    self._volume_fifo = None
 
   def info(self) -> models.SourceInfo:
     source = models.SourceInfo(

streams/shairport_volume_handler.py

@@ -0,0 +1,80 @@
+"""Script for synchronizing AmpliPi and Spotify volumes"""
+import argparse
+from time import sleep
+from dasbus.connection import SessionMessageBus
+from dasbus.typing import Variant
+from volume_synchronizer import VolSyncDispatcher, StreamWatcher, VolEvents
+
+
+class ShairportWatcher(StreamWatcher):
+  """A class that watches and tracks changes to airplay-side volume"""
+
+  def __init__(self, service_suffix: str):
+    super().__init__()
+    self.mpris = SessionMessageBus().get_proxy(
+      service_name=f"org.mpris.MediaPlayer2.{service_suffix}",
+      object_path="/org/mpris/MediaPlayer2",
+      interface_name="org.mpris.MediaPlayer2.Player"
+    )
+
+    self.dbus = SessionMessageBus().get_proxy(
+      service_name=f"org.mpris.MediaPlayer2.{service_suffix}",
+      object_path="/org/mpris/MediaPlayer2",
+      interface_name="org.freedesktop.DBus.Properties"
+    )
+
+  async def watch_vol(self):
+    """Watch the shairport mpris stream for volume changes and update amplipi volume info accordingly"""
+    while True:
+      try:
+        if self.volume != self.mpris.Volume:
+          self.logger.debug(f"Airplay volume changed from {self.volume} to {self.mpris.Volume}")
+          self.volume = float(self.mpris.Volume)
+          self.schedule_event(VolEvents.CHANGE_AMPLIPI)
+          # self.delta = self.mpris.Volume - self.volume
+
+      except Exception as e:
+        self.logger.exception(f"Error: {e}")
+        return
+      sleep(0.1)
+
+  def set_vol(self, amplipi_volume: float, vol_set_point: float) -> float:  # This has unused variable vol_set_point to keep up with the underlying StreamData.set_vol function schema
+    """Update Airplay's volume slider to match AmpliPi"""
+    try:
+      # Airplay does not allow external devices to set the volume of a users system
+
+      # Airplay is a fully authoritative volume source, meaning it forces amplipi volume to equal its volume now. If that ever changes, this will be relevant:
+        # There are two values this could realistically be returned and become the new vol_set_point, and they each have their own drawbacks:
+
+        # amplipi_volume: If amplipi_volume is the new set point, any changes to airplay volume will send the volume to an odd
+        # spot as it just sets the vol average of amplipi to be the same as the value of airplay's vol
+
+        # vol_set_point: if vol_set_point is retained as the set point, any changes to amplipi will reflect for 1-2 seconds at most and then
+        # bounce back to where it had been, resulting in a glitchy front end interface
+
+      # In any future MPRIS based volume synchronizers, you can check if self.mpris.CanControl is true and then potentially directly set self.mpris.Volume
+      # Note that we cannot do this due to this line: <property name='Volume' type='d' access='read'/>
+      # That exists in the MPRIS config xml at https://github.com/mikebrady/shairport-sync/blob/master/org.mpris.MediaPlayer2.xml
+
+      # self.dbus.Set(
+      #   'org.mpris.MediaPlayer2',
+      #   'Volume',
+      #   Variant("d", amplipi_volume)
+      # )
+
+      return amplipi_volume
+    except Exception as e:
+      self.logger.exception(f"Exception: {e}")
+
+
+if __name__ == "__main__":
+
+  parser = argparse.ArgumentParser(description="Read metadata from a given URL and write it to a file.")
+
+  parser.add_argument("service_suffix", help="Name of mpris instance", type=str)
+  parser.add_argument("config_dir", help="The directory of the vsrc config", type=str)
+  parser.add_argument("--debug", action="store_true", help="Change log level from WARNING to DEBUG")
+
+  args = parser.parse_args()
+
+  handler = VolSyncDispatcher(ShairportWatcher(service_suffix=args.service_suffix), args.config_dir, args.debug)

streams/spotify_volume_handler.py

@@ -22,13 +22,13 @@ class SpotifyWatcher(StreamWatcher):
 
   def __init__(self, api_port: int):
     super().__init__()
-
     self.api_port: int = api_port
     """What port is go-librespot running on? Typically set to 3678 + vsrc."""
 
   async def watch_vol(self):
     """Watch the go-librespot websocket endpoint for volume change events and update AmpliPi volume info accordingly"""
     try:
+      # Connect to the websocket and listen for state changes
       # pylint: disable=E1101
       # E1101: Module 'websockets' has no 'connect' member (no-member)
       async with websockets.connect(f"ws://localhost:{self.api_port}/events", open_timeout=5) as websocket:

streams/volume_synchronizer.py

@@ -32,6 +32,7 @@ def __init__(self):
     self._volume: float = None
     """Value between 0 and 1, or None if not yet initialized by the upstream"""
 
+    self.delta: Optional[float] = None
     self.logger: logging.Logger
     """logging.Logger instance provided by VolSyncDispatcher"""
 
@@ -73,7 +74,7 @@ def __init__(self, config_dir: str, schedule_event: Callable, logger: logging.Lo
     """Event scheduler function provided by VolSyncDispatcher, has limited valid inputs that can be seen in the VolEvents enum"""
 
     self.logger: logging.Logger = logger
-    self.volume: float = None
+    self.volume: Optional[float] = None
     self.config_dir: str = config_dir
 
     self.connected_zones: List[int] = []
@@ -87,13 +88,16 @@ def get_vol(self):
       Read the volume FIFO from .config/amplipi/srcs/v{vsrc}/vol to load the currently connected zones and the averaged volume of them
       If the read volume is different than the previous volume, send a volume change event to the stream
     """
-    with open(f'{self.config_dir}/vol', 'r') as fifo:
-      while True:
-        data = json.loads(fifo.readline().strip())
-        if self.volume != data["volume"]:
-          self.volume = data["volume"]
-          self.schedule_event(VolEvents.CHANGE_STREAM)
-        self.connected_zones = data["zones"]
+    try:
+      with open(f'{self.config_dir}/vol', 'r') as fifo:
+        while True:
+          data = json.loads(fifo.readline().strip())
+          if self.volume != data["volume"]:
+            self.volume = data["volume"]
+            self.schedule_event(VolEvents.CHANGE_STREAM)
+          self.connected_zones = data["zones"]
+    except Exception as e:
+      self.logger.exception(f"Error while getting writing to {self.config_dir}/vol fifo: {e}")
 
   def set_vol(self, stream_volume: float, vol_set_point: float):
     """Update AmpliPi's volume to match the stream volume"""
@@ -106,6 +110,13 @@ def set_vol(self, stream_volume: float, vol_set_point: float):
         return vol_set_point
 
       delta = float(stream_volume - self.volume)
+      return self.set_vol_delta(delta)
+    except Exception as e:
+      self.logger.exception(f"Exception: {e}")
+
+  def set_vol_delta(self, delta: float):
+    """Update AmpliPi's volume by delta"""
+    try:
       expected_volume = self.volume + delta
       self.logger.debug(f"Setting AmpliPi volume to {expected_volume} from {self.volume}")
       requests.patch(
@@ -185,7 +196,13 @@ def event_loop(self):
 
         event = self.event_queue.get()
         if event == VolEvents.CHANGE_AMPLIPI:
-          self.vol_set_point = self.amplipi.set_vol(self.stream.volume, self.vol_set_point)
+          if self.stream.delta is not None:
+            # Reduce race condition potential by decoupling the value from the variable
+            delta = float(self.stream.delta)
+            self.vol_set_point = self.amplipi.set_vol_delta(delta)
+            self.stream.delta -= delta
+          else:
+            self.vol_set_point = self.amplipi.set_vol(self.stream.volume, self.vol_set_point)
         elif event == VolEvents.CHANGE_STREAM:
           self.vol_set_point = self.stream.set_vol(self.amplipi.volume, self.vol_set_point)
       except queue.Empty: