import appdaemon.plugins.hass.hassapi as hass from ad_toolbox.smartobject import SmartObject import ad_toolbox.smartcondition as SmartCondition import time from ad_toolbox.virtualsensors import ValueSelector from ad_toolbox.eventhandler import EventHandler MIN_TIME_BETWEEN_GROUP_COMMANDS = 5 class PlayerDataSet: class VolumeSlot: def __init__(self,context,volume): self.context = list() self.context.extend(context) self.volume = volume def __init__(self): self.volume_slots = list() def get_volume_slots_as_string(self): def get_context_as_string(context): string = "" for item in context: string += f"{item[0]}={item[1]}, " return string[:-2] string = "\n" for slot in self.volume_slots: string = string + f"[{get_context_as_string(slot.context)}] {slot.volume}\n" if string == "\n": string = "No slot saved" else : string = string[:-1] return string def get_volume(self,controller,entity_list, logger = None): context = list() for entity in entity_list: context.append((entity,controller.get_state(entity))) if logger: logger(f"context = {context}") #search for a similar context for slot in self.volume_slots: if slot.context == context: if logger: logger(f"Mathing context = {slot.context}") logger(f"Mathing volume = {slot.volume}") return slot.volume return -1.0 def save_volume(self,controller,entity_list, volume): context = list() for entity in entity_list: context.append((entity,controller.get_state(entity))) #search for a similar context for slot in self.volume_slots: if slot.context == context: slot.volume = volume return self.volume_slots.append(PlayerDataSet.VolumeSlot(context,volume)) def clean(self,entity_list): for slot in self.volume_slots: if slot.volume == None: self.volume_slots.remove(slot) self.clean(entity_list) return if len(entity_list) != len(slot.context): self.volume_slots.remove(slot) self.clean(entity_list) return for i in range(0,len(slot.context)): if slot.context[i][0] != entity_list[i]: self.volume_slots.remove(slot) self.clean(entity_list) return class ControllerDataSet: def __init__(self): self.players_dataset = dict() class SonosPlayer: class VolumeTransition(): def __init__(self,current_volume,target_volume,duration): assert current_volume != target_volume self.start_time = time.time() self.end_time = self.start_time + duration self.start_volume = current_volume self.target_volume = target_volume def __str__(self): return "[" + str(self.start_volume) + " => " + str(self.target_volume) + "] step = " + str(self.get_transition_step()) def get_transition_volume(self): if time.time() > self.end_time: return self.target_volume else: time_elapse = time.time() - self.start_time percent = time_elapse / (self.end_time - self.start_time) return self.start_volume + percent*(self.target_volume - self.start_volume) def get_transition_step(self): ideal_number_of_step = abs(self.target_volume - self.start_volume) / 0.01 step = (self.end_time - self.start_time) / ideal_number_of_step if ideal_number_of_step > 0 else 0 return max(1.0,round(step,2)) def should_stop_transition(self,old,new): if old and new: if self.target_volume > self.start_volume: if new >= self.target_volume: return True if new < self.start_volume: return True if old > new : return True #we are decreasing volume while the transition is increasing elif self.target_volume < self.start_volume: if new <= self.target_volume: return True if new > self.start_volume: return True if old < new : return True #we are increasing volume while the transition is decreasing return False class VolumeContextData(): def __init__(self,entity,transition = 0,force_volume = -1): self.entity = entity self.force_volume = force_volume self.transition = transition def __init__(self,controller, player_name, yaml_conf,dataset : PlayerDataSet): self.controller = controller self.player_name = player_name self.group_master = None self.dataset = dataset if "entity" in yaml_conf: self.entity_id = yaml_conf["entity"] self.log(f"Initializing {player_name} with entity {self.entity_id}") else: self.entity_id = None self.log_error(f"Can't initialize {player_name}, not entity was configured",True) if "status_light" in yaml_conf: self.status_light_entity = yaml_conf["status_light"]["entity"] self.light_entity_conditions_evaluator = SmartCondition.Evaluator(self.controller,yaml_conf['status_light']["conditions"],constants = { 'self' : self.status_light_entity, 'super' : self.entity_id},condition_name = 'status_light',on_change_cb = self.on_change_status_light_conditions) if "audio_delay" in yaml_conf: # self.audio_delay_entity = yaml_conf["status_light"]["audio_delay"] self.log(f"Creating a ValueSelector for the audio_delay") self.audio_delay_selector = ValueSelector(self.controller,self.controller,sensor_name = yaml_conf["audio_delay"]["entity"],super_entity_id = self.entity_id,yaml_block = yaml_conf["audio_delay"]['value'],self_initialize = True) self.is_playing_podcast_sensor = f'binary_sensor.{player_name}_is_playing_podcast' is_playing_podcast_sensor_entity = self.controller.get_entity(self.is_playing_podcast_sensor) if not is_playing_podcast_sensor_entity.exists(): is_playing_podcast_sensor_entity.add() self.controller.listen_state(self.on_state_change,self.entity_id) self.controller.listen_state(self.on_update_media_artist,self.entity_id,attribute = "media_artist") self.update_is_playing_podcast_sensor() if "leave_conditions" in yaml_conf: self.leave_conditions_evaluator = SmartCondition.Evaluator(self.controller,yaml_conf["leave_conditions"],constants = { 'self' : self.entity_id},condition_name = player_name,on_update_cb = self.on_update_leave_condition, log_callback_trigger_reason = False ,extra_entities_to_listen = "sensor.ad_number_of_music_player_active") self.on_update_leave_condition() else: self.leave_conditions_evaluator = None if "join_conditions" in yaml_conf: self.join_conditions_evaluator = SmartCondition.Evaluator(self.controller,yaml_conf["join_conditions"],constants = { 'self' : self.entity_id},condition_name = player_name,on_update_cb = self.on_update_join_condition, log_callback_trigger_reason = False, extra_entities_to_listen = "sensor.ad_number_of_music_player_active") self.on_update_join_condition() else: self.join_conditions_evaluator = None self.off_conditions_evaluator = SmartCondition.Evaluator(self.controller,yaml_conf["off_conditions"],constants = { 'self' : self.entity_id},condition_name = player_name,on_update_cb = self.on_update_off_conditions, log_callback_trigger_reason = False) self.on_update_off_conditions() #Volume controller stuffs self.volume_context_data = list() self.volume_context_entities = list() self.volume_transition = None if 'volume_controller' in yaml_conf: volume_controller_conf = yaml_conf['volume_controller'] if "context_entities" in volume_controller_conf: for element in volume_controller_conf["context_entities"]: if "entity" in element: entity = element["entity"] else: self.log_error("Not entity defined") if "transition" in element: transition = element["transition"] else: transition = 0 if "force_volume" in element: force_volume = element["force_volume"] else : force_volume = -1 self.log(str(entity) + " : transition = " + str(transition) + ", force_volume = " + str(force_volume)) if force_volume < 0: self.volume_context_entities.append(entity) self.volume_context_data.append(SonosPlayer.VolumeContextData(entity,transition,force_volume)) if len(self.volume_context_entities) > 0: #I don't want the context save to be dependent of the order of the yaml file self.volume_context_entities.sort() for element in self.volume_context_data: self.log(f"adding context for {element.entity}") self.controller.listen_state(self.on_volume_context_change,element.entity) self.dataset.clean(self.volume_context_entities) self.log(f"Volumes slots: {self.dataset.get_volume_slots_as_string()}") self.controller.listen_state(self.on_volume_change,self.entity_id,attribute = "volume_level") #self.controller.listen_state(self.on_state_change,self.entity_id,new = 'playing') #self.controller.listen_state(self.on_state_change,self.entity_id,old = 'playing') if self.is_playing(): self.update_volume() else: self.log("Not context entities defined") # --- Playback event handlers --- self._current_radio_index = -1 self._current_album_index = -1 if 'play_stop_events' in yaml_conf: self._play_stop_event_handler = EventHandler(self.controller, yaml_conf['play_stop_events'], self.on_play_stop_event) if 'cycle_radio_events' in yaml_conf: self._cycle_radio_event_handler = EventHandler(self.controller, yaml_conf['cycle_radio_events'], self.on_cycle_radio_event) if 'cycle_album_events' in yaml_conf: self._cycle_album_event_handler = EventHandler(self.controller, yaml_conf['cycle_album_events'], self.on_cycle_album_event) def get_player_id(self): return self.entity_id def set_group_master(self,player_id): self.group_master = player_id def log(self,message,**kwargs): self.controller.log(f"[{self.player_name}]{message}",**kwargs) def stop_playback(self): # if self.get_number_of_sonos_player_active() > 1: # self.leave_group() # else: self.log(f"{self.entity_id} is pausing the playback") self.controller.call_service("media_player/media_pause", entity_id = self.entity_id) # ----------------------------------------------------------------------- # Playback event handlers # ----------------------------------------------------------------------- def _browse_media(self, media_content_type, media_content_id): """Call browse_media and return the children list, or [] on error.""" try: result = self.controller.call_service( "media_player/browse_media", entity_id=self.entity_id, media_content_type=media_content_type, media_content_id=media_content_id, ) if result: browse_data = (result.get("result") or {}).get("response") or {} entity_data = browse_data.get(self.entity_id) or {} return entity_data.get("children") or [] except Exception as e: self.log(f"_browse_media({media_content_type}) error: {e}", level="WARNING") return [] def _fetch_sonos_favorites(self): """Browse the Sonos favorites tree and return a flat list of individual favorites. Each item is tagged with '_is_radio' (True/False). The Sonos integration organises favorites into typed folders (media_content_type='favorites_folder'). We browse into each folder and tag its children: - folders whose media_content_id contains 'audioBroadcast' → radio - all other folders → album/playlist If the response is already a flat list (no folders), we fall back to tagging by can_expand (False → radio, True → album). """ folders = self._browse_media("favorites", "") items = [] for folder in folders: if folder.get("media_content_type") == "favorites_folder": folder_id = folder.get("media_content_id", "") is_radio = "audioBroadcast" in folder_id or "audioItem" in folder_id children = self._browse_media(folder["media_content_type"], folder_id) for child in children: tagged = dict(child) tagged["_is_radio"] = is_radio items.append(tagged) else: tagged = dict(folder) tagged["_is_radio"] = not folder.get("can_expand", False) items.append(tagged) self.log(f"_fetch_sonos_favorites: {len(items)} items " f"({sum(1 for f in items if f['_is_radio'])} radio, " f"{sum(1 for f in items if not f['_is_radio'])} albums)") return items def _play_favorite(self, fav): """Play a Sonos favourite by name via select_source.""" title = fav.get("title", "") self.log(f"playing favourite '{title}' via select_source") self.controller.call_service( "media_player/select_source", entity_id=self.entity_id, source=title, ) def on_play_stop_event(self, event_name, event_data, context): """Toggle play/pause. If the player is idle/off, start playback from the first Sonos favourite fetched via browse_media.""" player_state = self.controller.get_state(self.entity_id) self.log(f"play_stop_event: player_state = {player_state}") if player_state == 'playing': self.log("play_stop_event: pausing playback") self.controller.call_service("media_player/media_pause", entity_id=self.entity_id) elif player_state == 'paused': self.log("play_stop_event: resuming playback") self.controller.call_service("media_player/media_play", entity_id=self.entity_id) else: favs = self._fetch_sonos_favorites() if favs: self._play_favorite(favs[0]) else: self.log("play_stop_event: no favourites available", level="WARNING") def on_cycle_radio_event(self, event_name, event_data, context): """Cycle through Sonos radio favourites (folders tagged as audioBroadcast).""" radios = [f for f in self._fetch_sonos_favorites() if f.get("_is_radio")] if not radios: self.log("cycle_radio_event: no radio favourites found", level="WARNING") return self._current_radio_index = (self._current_radio_index + 1) % len(radios) fav = radios[self._current_radio_index] self.log(f"cycle_radio_event: '{fav.get('title')}' (index {self._current_radio_index}/{len(radios)})") self._play_favorite(fav) def on_cycle_album_event(self, event_name, event_data, context): """Cycle through Sonos album/playlist favourites (all non-radio folders).""" albums = [f for f in self._fetch_sonos_favorites() if not f.get("_is_radio")] if not albums: self.log("cycle_album_event: no album/playlist favourites found", level="WARNING") return self._current_album_index = (self._current_album_index + 1) % len(albums) fav = albums[self._current_album_index] self.log(f"cycle_album_event: '{fav.get('title')}' (index {self._current_album_index}/{len(albums)})") self._play_favorite(fav) def on_change_status_light_conditions(self,prev_result,result): if result == SmartCondition.Result.Succeeded: self.controller.turn_on(self.status_light_entity) else: self.controller.turn_off(self.status_light_entity) def is_playing(self): player_state = self.controller.get_state(self.entity_id) #self.log(f"Current state = {player_state}") return player_state == 'playing' def on_update_join_condition(self,*args): if self.join_conditions_evaluator.evaluate(False) == SmartCondition.Result.Succeeded: if self.controller.get_number_of_players_active() >= 1 and not self.is_playing() and self.group_master == None: self.join_conditions_evaluator.log_callback_trigger_reason() self.join_conditions_evaluator.log_evaluation_result() if self.controller.can_send_group_command(): self.controller.join_group(self.entity_id) else: self.log(f"A group command has been send recently, trying again in 2s") self.controller.run_in(self.on_update_join_condition,2) def on_update_leave_condition(self,*args): #*args is needed because the function is also use as a run_in cb if self.leave_conditions_evaluator.evaluate(False) == SmartCondition.Result.Succeeded: if self.group_master != None: self.leave_conditions_evaluator.log_callback_trigger_reason() self.leave_conditions_evaluator.log_evaluation_result() if self.controller.can_send_group_command(): self.controller.leave_group(self.entity_id) else: self.log(f"A group command has been send recently, trying again in 2s") self.controller.run_in(self.on_update_leave_condition,2) def on_state_change(self, entity, attribute, old, new, *kwargs): self.log(f"on_state_change, entity = {entity}, attribute = {attribute}, old = {old}, new = {new}") if old != new and self.is_playing(): transition_time = 5 if self.get_desired_volume() > self.controller.get_state(self.entity_id, attribute="volume_level") else 0 self.update_volume(transition_time) self.update_is_playing_podcast_sensor() def on_update_media_artist(self, entity, attribute, old, new, *kwargs): #self.log(f"on_update_media_artist, entity = {entity}, attribute = {attribute}, old = {old}, new = {new}") self.update_is_playing_podcast_sensor() def update_is_playing_podcast_sensor(self): if self.is_playing_podcast_sensor and self.is_playing(): prev_state = self.controller.get_state(self.is_playing_podcast_sensor) if self.controller.get_state(self.entity_id,attribute = "media_artist") == None: if prev_state != 'on': self.log(f"setting {self.is_playing_podcast_sensor} to on") self.controller.set_state(self.is_playing_podcast_sensor,state = "on") else: if prev_state != 'off': self.log(f"setting {self.is_playing_podcast_sensor} to off") self.controller.set_state(self.is_playing_podcast_sensor,state = "off") def on_update_off_conditions(self): if self.off_conditions_evaluator.evaluate(False) == SmartCondition.Result.Succeeded: if self.is_playing(): self.off_conditions_evaluator.log_callback_trigger_reason() self.off_conditions_evaluator.log_evaluation_result() self.stop_playback() def is_volume_forced(self): for element in self.volume_context_data: if element.force_volume >= 0 and self.controller.get_state(element.entity) != "off": return True return False def get_force_volume(self): for element in self.volume_context_data: if element.force_volume >= 0 and self.controller.get_state(element.entity) != "off": return element.force_volume return -1.0 def is_volume_transitioning(self): return self.volume_transition != None #and round(float(self.volume_transition.target_volume),2) != round(float(self.controller.get_state(self.entity_id, attribute="volume_level")),2) def cancel_volume_transition(self): self.volume_transition = None def on_update_volume_transition(self, kwargs): if self.volume_transition: transition_step = self.volume_transition.get_transition_step() if self.is_playing() == False: self.cancel_volume_transition() else: self.controller.call_service("media_player/volume_set", entity_id = self.entity_id, volume_level = self.volume_transition.get_transition_volume()) self.controller.run_in(self.on_update_volume_transition, transition_step) def on_volume_context_change(self, entity, attribute, old, new, *kwargs): if old != new: if self.is_playing(): self.log(str(entity) + " state changed from " + str(old) + " to " + str(new) + ", attribute = " + str(attribute)) transition_time = 0.0 for element in self.volume_context_data: if element.entity == entity: transition_time = element.transition self.update_volume(transition_time) def on_volume_change(self, entity, attribute, old, new, *kwargs): if new != old and new != None and self.is_playing(): self.log("Volume changed from " + str(old) + " to " + str(new)) if self.is_volume_transitioning(): self.log("Volume Transition active: " + str(self.volume_transition)) if self.volume_transition.should_stop_transition(old,new): self.log("Volume Transition Stopped") self.cancel_volume_transition() if not self.is_volume_forced(): self.dataset.save_volume(self.controller,self.volume_context_entities,new) elif not self.is_volume_forced(): self.dataset.save_volume(self.controller,self.volume_context_entities,new) def get_desired_volume(self): desired_volume = self.get_force_volume() if desired_volume < 0: desired_volume = self.dataset.get_volume(self.controller,self.volume_context_entities,self.log) else: self.log(f"volume has been forced too {desired_volume}") return desired_volume def update_volume(self, transition_time = 0.0): desired_volume = self.get_desired_volume() current_volume = self.controller.get_state(self.entity_id, attribute="volume_level") if desired_volume < 0: #no context saved, return current volume self.log("No context was found, creating a new one",level = "WARNING") if current_volume != None and not self.is_volume_forced(): self.dataset.save_volume(self.controller,self.volume_context_entities,current_volume) elif desired_volume != current_volume: if transition_time > 0: self.volume_transition = SonosPlayer.VolumeTransition(current_volume,desired_volume,transition_time) self.log("Request transition to " + str(desired_volume)) self.controller.run_in(self.on_update_volume_transition, 1) else: self.log("Request Volume changed to " + str(desired_volume)) self.cancel_volume_transition() self.controller.call_service("media_player/volume_set", entity_id = self.entity_id, volume_level = desired_volume) elif self.volume_transition: #the current_volume is equal to the desired_volume and we are in the middle of a transition self.cancel_volume_transition() class SonosController(SmartObject): GROUP_ATTRIBUTE = "group_members" #@SmartCondition.catch_smartcondition_exception(lambda self, message: self.log_error(message,stop_app = True)) def on_initialize_smart_object(self): #super().initialize() self.players = dict() self.players_groups = dict() self.group_command_time = 0 #self.depends_on_module("virtualsensors") #Creating a new dataset if nothing has been deserialized from initialize if self.dataset == None: self.dataset = ControllerDataSet() if "players" in self.args: for player in self.args['players']: self.log(f"Creating player {player}") player_id = self.args['players'][player]["entity"] if not player_id in self.dataset.players_dataset: self.log(f"Creating dataset for {player_id}") self.dataset.players_dataset[player_id] = PlayerDataSet() self.players[player_id] = SonosPlayer(self,player,self.args['players'][player],self.dataset.players_dataset[player_id]) self.listen_state(self.on_sonos_group_changed,player_id,attribute = self.GROUP_ATTRIBUTE) self.update_player_group(player_id) self.log_groups() else: self.log(f"No players found",level ="ERROR") def update_player_group(self,player_id): self.log(f"[{player_id}]updating players group") if player_id != None: players_list = self.get_state(player_id, attribute=self.GROUP_ATTRIBUTE) if players_list: self.add_player_to_group(player_id,players_list[0]) def add_player_to_group(self,player_id,group_id): group_changed = False group_master = None #let's confirm the group_master know me players_list = self.get_state(group_id, attribute=self.GROUP_ATTRIBUTE) if player_id != group_id and not player_id in players_list: # sometimes, some player think they belong to a group, but the group master doesn't know them # in such case I send them back to their own group self.log(f"{group_id} doesn't have {player_id} in it's group, although {player_id} seems to think otherwise") group_id = player_id else: if players_list != None and len(players_list) > 1: group_master = players_list[0] self.get_player(player_id).set_group_master(group_master) #self.set_state(player_id,attributes = {"group_master" : group_master}) empty_group = list() for current_group_id in self.players_groups: if current_group_id != group_id and player_id in self.players_groups[current_group_id]: self.log(f"Removing {player_id} from it's previous group {current_group_id}") group_changed = True self.players_groups[current_group_id].remove(player_id) if len(self.players_groups[current_group_id]) == 0: empty_group.append(current_group_id) for group in empty_group: del self.players_groups[group] if group_id in self.players_groups: if not player_id in self.players_groups[group_id]: self.log(f"Adding {player_id} to group {group_id}") group_changed = True self.players_groups[group_id].append(player_id) else: self.log(f"Creating group {group_id} and adding {player_id}") group_changed = True self.players_groups[group_id] = [player_id] return group_changed def get_player(self,player_id : str): return self.players[player_id] def try_get_player(self,player_id : str): try: return self.players[player_id] except KeyError: return None def log_groups(self): solo_players = list() for group in self.players_groups: if len(self.players_groups[group]) > 1: self.log(f"Group[{group}] : {self.players_groups[group]}") else: solo_players.extend(self.players_groups[group]) if len(solo_players) >= 1: self.log(f"Solo players : {solo_players}") def on_sonos_group_changed(self, entity, attribute, old, new, kwargs): self.log(f"on_sonos_group_changed, entity = {entity}, attribute = {attribute}, old = {old}, new = {new}") if new and new != old: if self.add_player_to_group(entity,new[0]): self.log_groups() def join_group(self,player_id): if self.can_send_group_command(): best_group = None best_group_size = 0 for group_master,group_members in self.players_groups.items(): if self.players[group_master].is_playing(): #TODO : ignore here stand-alone player group_size = len(group_members) if group_size > best_group_size: best_group_size = group_size best_group = group_master if best_group: self.log(f"Join group of {best_group}") self.call_service("media_player/join", entity_id = best_group, group_members = [ player_id ] ) self.group_command_time = time.time() return self.log(f"Couldn't find any group to join",level = "WARNING") else: self.log(f"A previous group command has been sent less than {MIN_TIME_BETWEEN_GROUP_COMMANDS}s ago, please wait",level = "WARNING") def leave_group(self,player_id): if self.can_send_group_command(): group_master = self.get_player(player_id).group_master self.log(f"{player_id} is leaving the group of {group_master}") self.call_service("media_player/unjoin", entity_id = player_id) self.group_command_time = time.time() self.get_player(player_id).set_group_master(None) else: self.log(f"A previous group command has been sent less than {MIN_TIME_BETWEEN_GROUP_COMMANDS}s ago, please wait",level = "WARNING") def can_send_group_command(self): return time.time() - self.group_command_time > MIN_TIME_BETWEEN_GROUP_COMMANDS def get_number_of_players_active(self): number_of_players_active = 0 for player_id in self.players: if self.players[player_id].is_playing(): number_of_players_active += 1 return number_of_players_active