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