/
/
/
1"""Demo Player Provider implementation."""
2
3from __future__ import annotations
4
5from typing import TYPE_CHECKING, cast
6
7from music_assistant_models.config_entries import ConfigEntry
8from music_assistant_models.enums import ConfigEntryType
9from zeroconf import ServiceStateChange
10
11from music_assistant.helpers.util import get_primary_ip_address_from_zeroconf
12from music_assistant.models.player_provider import PlayerProvider
13
14from .constants import CONF_NUMBER_OF_PLAYERS
15from .player import DemoPlayer
16
17if TYPE_CHECKING:
18 from zeroconf.asyncio import AsyncServiceInfo
19
20
21class DemoPlayerprovider(PlayerProvider):
22 """
23 Example/demo Player provider.
24
25 Note that this is always subclassed from PlayerProvider,
26 which in turn is a subclass of the generic Provider model.
27
28 The base implementation already takes care of some convenience methods,
29 such as the mass object and the logger. Take a look at the base class
30 for more information on what is available.
31
32 Just like with any other subclass, make sure that if you override
33 any of the default methods (such as __init__), you call the super() method.
34 In most cases its not needed to override any of the builtin methods and you only
35 implement the abc methods with your actual implementation.
36 """
37
38 async def get_config_entries(self) -> tuple[ConfigEntry, ...]:
39 """
40 Return the (options) config entries for this (existing) provider instance.
41
42 Return an empty tuple when the provider has no options. Interactive setup input
43 (if any) is collected by a ``setup_flow.py`` module; one-shot buttons are declared
44 here as ``ConfigEntryType.ACTION`` entries and handled in ``handle_config_action``.
45 """
46 return (
47 # example of a ConfigEntry for the number of players to create
48 ConfigEntry(
49 key=CONF_NUMBER_OF_PLAYERS,
50 type=ConfigEntryType.INTEGER,
51 label="Number of Players",
52 required=True,
53 default_value=2,
54 description="Number of demo players to create.",
55 ),
56 )
57
58 async def handle_async_init(self) -> None:
59 """Handle async initialization of the provider."""
60 # OPTIONAL
61 # this is an optional method that you can implement if
62 # relevant or leave out completely if not needed.
63 # it will be called when the provider is initialized in Music Assistant.
64 # you can use this to do any async initialization of the provider,
65 # such as loading configuration, setting up connections, etc.
66 self.logger.info("Initializing DemoPlayerProvider with config: %s", self.config)
67
68 async def loaded_in_mass(self) -> None:
69 """Call after the provider has been loaded."""
70 # OPTIONAL
71 # this is an optional method that you can implement if
72 # relevant or leave out completely if not needed.
73 # it will be called after the provider has been fully loaded into Music Assistant.
74 self.logger.info("DemoPlayerProvider loaded")
75
76 async def unload(self, is_removed: bool = False) -> None:
77 """
78 Handle unload/close of the provider.
79
80 Called when provider is deregistered (e.g. MA exiting or config reloading).
81 is_removed will be set to True when the provider is removed from the configuration.
82 """
83 # OPTIONAL
84 # this is an optional method that you can implement if
85 # relevant or leave out completely if not needed.
86 # it will be called when the provider is unloaded from Music Assistant.
87 # this means also when the provider is getting reloaded
88 for player in self.players:
89 # if you have any cleanup logic for the players, you can do that here.
90 # e.g. disconnecting from the player, closing connections, etc.
91 self.logger.debug("Unloading player %s", player.name)
92 await self.mass.players.unregister(player.player_id)
93
94 def on_player_enabled(self, player_id: str) -> None:
95 """Call (by config manager) when a player gets enabled."""
96 # OPTIONAL
97 # this is an optional method that you can implement if
98 # you want to do something special when a player is enabled.
99 super().on_player_enabled(player_id)
100
101 def on_player_disabled(self, player_id: str) -> None:
102 """Call (by config manager) when a player gets disabled."""
103 # OPTIONAL
104 # this is an optional method that you can implement if
105 # you want to do something special when a player is disabled.
106 # e.g. you can stop polling the player or disconnect from it.
107 super().on_player_disabled(player_id)
108
109 async def remove_player(self, player_id: str) -> None:
110 """Remove a player from this provider."""
111 # OPTIONAL - required only if you specified ProviderFeature.REMOVE_PLAYER
112 # this is used to actually remove a player.
113
114 async def on_mdns_service_state_change(
115 self, name: str, state_change: ServiceStateChange, info: AsyncServiceInfo | None
116 ) -> None:
117 """Handle MDNS service state callback."""
118 # MANDATORY IF YOU WANT TO USE MDNS DISCOVERY
119 # OPTIONAL if you dont use mdns for discovery of players
120 # If you specify a mdns service type in the manifest.json, this method will be called
121 # automatically on mdns changes for the specified service type.
122
123 # If no mdns service type is specified, this method is omitted and you
124 # can completely remove it from your provider implementation.
125
126 if not info:
127 return # guard
128
129 # NOTE: If you do not use mdns for discovery of players on the network,
130 # you must implement your own discovery mechanism and logic to add new players
131 # and update them on state changes when needed.
132 # Below is a bit of example implementation but we advise to look at existing
133 # player providers for more inspiration.
134 name = name.split("@", 1)[1] if "@" in name else name
135 player_id = info.decoded_properties["uuid"] # this is just an example!
136 if not player_id:
137 return # guard, we need a player_id to work with
138
139 # handle removed player
140 if state_change == ServiceStateChange.Removed:
141 # check if the player manager has an existing entry for this player
142 if mass_player := self.mass.players.get_player(player_id):
143 # the player has become unavailable
144 self.logger.debug("Player offline: %s", mass_player.display_name)
145 await self.mass.players.unregister(player_id)
146 return
147 # handle update for existing device
148 # (state change is either updated or added)
149 # check if we have an existing player in the player manager
150 # note that you can use this point to update the player connection info
151 # if that changed (e.g. ip address)
152 if mass_player := self.mass.players.get_player(player_id):
153 # existing player found in the player manager,
154 # this is an existing player that has been updated/reconnected
155 # or simply a re-announcement on mdns.
156 cur_address = get_primary_ip_address_from_zeroconf(info)
157 if cur_address and cur_address != mass_player.device_info.ip_address:
158 self.logger.debug(
159 "Address updated to %s for player %s", cur_address, mass_player.display_name
160 )
161 # inform the player manager of any changes to the player object
162 # note that you would normally call this from some other callback from
163 # the player's native api/library which informs you of changes in the player state.
164 # as a last resort you can also choose to let the player manager
165 # poll the player for state changes
166 mass_player.update_state()
167 return
168 # handle new player
169 self.logger.debug("Discovered device %s on %s", name, cur_address)
170 # your own connection logic will probably be implemented here where
171 # you connect to the player etc. using your device/provider specific library.
172
173 async def discover_players(self) -> None:
174 """Discover players for this provider."""
175 # This is an optional method that you can implement if
176 # you want to (manually) discover players on the
177 # network and you do not use mdns discovery.
178 number_of_players = cast("int", self.config.get_value(CONF_NUMBER_OF_PLAYERS, 0))
179 self.logger.info(
180 "Discovering %s demo players",
181 number_of_players,
182 )
183 for i in range(number_of_players):
184 player = DemoPlayer(
185 provider=self,
186 player_id=f"demo_{i}",
187 )
188 # register the player with the player manager
189 await self.mass.players.register(player)
190 # once the player is registered, you can either instruct the player manager to
191 # poll the player for state changes or you can implement your own logic to
192 # listen for state changes from the player and update the player object accordingly.
193 # if the player state needs to be updated, you can call the update method on the player:
194 # player.update_state()
195