Update airplay sync to work again post-rebase

Author: SteveMicroNova

amplipi/streams/airplay.py

@@ -8,6 +8,8 @@
 import os
 import io
 import sys
+import threading
+import json
 
 
 def write_sp_config_file(filename, config):
@@ -43,7 +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.volume_process: Optional[subprocess.Popen] = 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)
@@ -73,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()
@@ -101,7 +125,8 @@ 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,
+        'software_mixer': 'yes'
       },
     }
 
@@ -111,10 +136,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}')
 
@@ -127,12 +152,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, str(self.id)]
+      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_process = subprocess.Popen(args=vol_args, stdout=self._log_file, stderr=self._log_file)
+      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}')
 
@@ -142,18 +170,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_process is not None:
-      self.volume_process.terminate()
-      if self.volume_process.wait(1) != 0:
-        logger.info('killing shairport vol sync')
-        self.volume_process.kill()
+      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:
@@ -162,8 +194,11 @@ def _deactivate(self):
       except Exception as e:
         logger.exception(f'Error removing airplay config files: {e}')
     self._disconnect()
+
     self.proc = None
-    self.volume_process = 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

@@ -1,39 +1,37 @@
 """Script for synchronizing AmpliPi and Spotify volumes"""
 import argparse
-import logging
-import sys
 from time import sleep
-
-from volume_synchronizer import VolumeSynchronizer, StreamData
-
 from dasbus.connection import SessionMessageBus
+from dasbus.typing import Variant
+from volume_synchronizer import VolSyncDispatcher, StreamWatcher, VolEvents
 
 
-logger = logging.getLogger(__name__)
-logger.setLevel(logging.DEBUG)
-sh = logging.StreamHandler(sys.stdout)
-logger.addHandler(sh)
-
-
-class ShairportData(StreamData):
+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"
     )
-    super().__init__()
+
+    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 is not None and self.volume != self.mpris.Volume:
+        if self.volume != self.mpris.Volume:
           self.logger.debug(f"Airplay volume changed from {self.volume} to {self.mpris.Volume}")
-          self.callback("stream_volume_changed")
-        self.volume = float(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}")
@@ -54,8 +52,15 @@ def set_vol(self, amplipi_volume: float, vol_set_point: float) -> float:  # This
         # 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 synchronisers, you can check if self.mpris.CanControl is true and then potentially directly set self.mpris.Volume
-      # Apple just disallows externally sourced volume changes due to their walled garden
+      # 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:
@@ -67,10 +72,9 @@ def set_vol(self, amplipi_volume: float, vol_set_point: float) -> float:  # This
   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("stream_id", help="The stream's amplipi side stream_id", type=int)
+  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 = VolumeSynchronizer(ShairportData(service_suffix=args.service_suffix), args.stream_id, args.debug)
-  handler.watcher_loop()
+  handler = VolSyncDispatcher(ShairportWatcher(service_suffix=args.service_suffix), args.config_dir, args.debug)

streams/volume_synchronizer.py

@@ -74,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] = []
@@ -88,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"""