<filename>music_assistant/providers/squeezebox/__init__.py
"""Squeezebox emulated player provider."""

import asyncio
import logging
from typing import List

from music_assistant.constants import CONF_CROSSFADE_DURATION
from music_assistant.helpers.typing import MusicAssistant
from music_assistant.helpers.util import callback, create_task
from music_assistant.models.config_entry import ConfigEntry
from music_assistant.models.player import DeviceInfo, Player, PlayerFeature, PlayerState
from music_assistant.models.player_queue import QueueItem
from music_assistant.models.provider import PlayerProvider

from .constants import PROV_ID, PROV_NAME
from .discovery import DiscoveryProtocol
from .socket_client import SqueezeEvent, SqueezeSocketClient

CONF_LAST_POWER = "last_power"
CONF_LAST_VOLUME = "last_volume"

LOGGER = logging.getLogger(PROV_ID)

CONFIG_ENTRIES = []  # we don't have any provider config entries (for now)
PLAYER_FEATURES = [PlayerFeature.QUEUE, PlayerFeature.CROSSFADE, PlayerFeature.GAPLESS]
PLAYER_CONFIG_ENTRIES = []  # we don't have any player config entries (for now)


async def setup(mass):
    """Perform async setup of this Plugin/Provider."""
    prov = PySqueezeProvider()
    await mass.register_provider(prov)


class PySqueezeProvider(PlayerProvider):
    """Python implementation of SlimProto server."""

    _tasks = []

    @property
    def id(self) -> str:
        """Return provider ID for this provider."""
        return PROV_ID

    @property
    def name(self) -> str:
        """Return provider Name for this provider."""
        return PROV_NAME

    @property
    def config_entries(self) -> List[ConfigEntry]:
        """Return Config Entries for this provider."""
        return CONFIG_ENTRIES

    async def on_start(self) -> bool:
        """Handle initialization of the provider. Called on startup."""
        # start slimproto server
        create_task(asyncio.start_server(self._client_connected, "0.0.0.0", 3483))

        # setup discovery
        create_task(self.start_discovery())

    async def start_discovery(self):
        """Start discovery for players."""
        transport, _ = await self.mass.loop.create_datagram_endpoint(
            lambda: DiscoveryProtocol(self.mass.web.port),
            local_addr=("0.0.0.0", 3483),
        )
        try:
            while True:
                await asyncio.sleep(60)  # serve forever
        finally:
            transport.close()

    async def _client_connected(self, reader, writer):
        """Handle a client connection on the socket."""
        addr = writer.get_extra_info("peername")
        LOGGER.debug("Socket client connected: %s", addr)
        socket_client = SqueezeSocketClient(self.mass, reader, writer)

        def handle_event(event: SqueezeEvent, socket_client: SqueezeSocketClient):
            player_id = socket_client.player_id
            if not player_id:
                return
            # always check if we already have this player as it might be reconnected
            player = self.mass.players.get_player(player_id)
            if not player:
                player = SqueezePlayer(self.mass, socket_client)
            player.set_socket_client(socket_client)
            # just update, the playermanager will take care of adding it if it's a new player
            player.handle_socket_client_event(event)

        socket_client.register_callback(handle_event)


class SqueezePlayer(Player):
    """Squeezebox player."""

    def __init__(self, mass: MusicAssistant, socket_client: SqueezeSocketClient):
        """Initialize."""
        super().__init__()
        self.mass = mass
        self._socket_client = socket_client

    @property
    def available(self) -> bool:
        """Return current availablity of player."""
        return self._socket_client.connected

    @property
    def should_poll(self) -> bool:
        """Return True if this player should be polled for state updates."""
        return False

    @property
    def socket_client(self):
        """Return the uinderluing socket client for the player."""
        return self._socket_client

    def set_socket_client(self, socket_client: SqueezeSocketClient):
        """Set a (new) socket client to this player."""
        self._socket_client = socket_client

    async def on_remove(self) -> None:
        """Call when player is removed from the player manager."""
        self.socket_client.disconnect()

    @property
    def player_id(self) -> str:
        """Return player id (=mac address) of the player."""
        return self.socket_client.player_id

    @property
    def provider_id(self) -> str:
        """Return provider id of this player."""
        return PROV_ID

    @property
    def name(self) -> str:
        """Return name of the player."""
        return self.socket_client.name

    @property
    def volume_level(self):
        """Return current volume level of player."""
        return self.socket_client.volume_level

    @property
    def powered(self):
        """Return current power state of player."""
        return self.socket_client.powered

    @property
    def muted(self):
        """Return current mute state of player."""
        return self.socket_client.muted

    @property
    def state(self):
        """Return current state of player."""
        return PlayerState(self.socket_client.state)

    @property
    def elapsed_time(self):
        """Return elapsed_time of current playing track in (fractions of) seconds."""
        return self.socket_client.elapsed_seconds

    @property
    def elapsed_milliseconds(self) -> int:
        """Return (realtime) elapsed time of current playing media in milliseconds."""
        return self.socket_client.elapsed_milliseconds

    @property
    def current_uri(self):
        """Return uri of currently loaded track."""
        return self.socket_client.current_uri

    @property
    def features(self) -> List[PlayerFeature]:
        """Return list of features this player supports."""
        return PLAYER_FEATURES

    @property
    def config_entries(self) -> List[ConfigEntry]:
        """Return player specific config entries (if any)."""
        return PLAYER_CONFIG_ENTRIES

    @property
    def device_info(self) -> DeviceInfo:
        """Return the device info for this player."""
        return DeviceInfo(
            model=self.socket_client.device_type,
            address=self.socket_client.device_address,
        )

    async def cmd_stop(self):
        """Send stop command to player."""
        return await self.socket_client.cmd_stop()

    async def cmd_play(self):
        """Send play (unpause) command to player."""
        return await self.socket_client.cmd_play()

    async def cmd_pause(self):
        """Send pause command to player."""
        return await self.socket_client.cmd_pause()

    async def cmd_power_on(self) -> None:
        """Send POWER ON command to player."""
        # save power and volume state in cache
        cache_str = f"squeezebox_player_{self.player_id}"
        await self.mass.cache.set(cache_str, (True, self.volume_level))
        return await self.socket_client.cmd_power(True)

    async def cmd_power_off(self) -> None:
        """Send POWER OFF command to player."""
        # save power and volume state in cache
        cache_str = f"squeezebox_player_{self.player_id}"
        await self.mass.cache.set(cache_str, (False, self.volume_level))
        return await self.socket_client.cmd_power(False)

    async def cmd_volume_set(self, volume_level: int):
        """Send new volume level command to player."""
        return await self.socket_client.cmd_volume_set(volume_level)

    async def cmd_mute(self, muted: bool = False):
        """Send mute command to player."""
        return await self.socket_client.cmd_mute(muted)

    async def cmd_play_uri(self, uri: str):
        """Request player to start playing a single uri."""
        crossfade = self.mass.config.player_settings[self.player_id][
            CONF_CROSSFADE_DURATION
        ]
        return await self.socket_client.play_uri(uri, crossfade_duration=crossfade)

    async def cmd_next(self):
        """Send NEXT TRACK command to player."""
        queue = self.mass.players.get_player_queue(self.player_id)
        if queue:
            new_track = queue.get_item(queue.cur_index + 1)
            if new_track:
                return await self.cmd_play_uri(new_track.stream_url)

    async def cmd_previous(self):
        """Send PREVIOUS TRACK command to player."""
        queue = self.mass.players.get_player_queue(self.player_id)
        if queue:
            new_track = queue.get_item(queue.cur_index - 1)
            if new_track:
                return await self.cmd_play_uri(new_track.stream_url)

    async def cmd_queue_play_index(self, index: int):
        """
        Play item at index X on player's queue.

            :param index: (int) index of the queue item that should start playing
        """
        queue = self.mass.players.get_player_queue(self.player_id)
        if queue:
            new_track = queue.get_item(index)
            if new_track:
                return await self.cmd_play_uri(new_track.stream_url)

    async def cmd_queue_load(self, queue_items: List[QueueItem], repeat: bool = False):
        """
        Load/overwrite given items in the player's queue implementation.

            :param queue_items: a list of QueueItems
        """
        if queue_items:
            await self.cmd_play_uri(queue_items[0].stream_url)
            return await self.cmd_play_uri(queue_items[0].stream_url)

    async def cmd_queue_insert(
        self, queue_items: List[QueueItem], insert_at_index: int
    ):
        """
        Insert new items at position X into existing queue.

        If insert_at_index 0 or None, will start playing newly added item(s)
            :param queue_items: a list of QueueItems
            :param insert_at_index: queue position to insert new items
        """
        # queue handled by built-in queue controller
        # we only check the start index
        queue = self.mass.players.get_player_queue(self.player_id)
        if queue and insert_at_index == queue.cur_index:
            return await self.cmd_queue_play_index(insert_at_index)

    async def cmd_queue_append(self, queue_items: List[QueueItem]):
        """
        Append new items at the end of the queue.

            :param queue_items: a list of QueueItems
        """
        # automagically handled by built-in queue controller

    async def cmd_queue_update(self, queue_items: List[QueueItem]):
        """
        Overwrite the existing items in the queue, used for reordering.

            :param queue_items: a list of QueueItems
        """
        # automagically handled by built-in queue controller

    async def cmd_queue_clear(self):
        """Clear the player's queue."""
        # queue is handled by built-in queue controller but send stop
        return await self.cmd_stop()

    async def restore_states(self):
        """Restore power/volume states."""
        cache_str = f"squeezebox_player_{self.player_id}"
        cache_data = await self.mass.cache.get(cache_str)
        last_power, last_volume = cache_data if cache_data else (False, 40)
        await self.socket_client.cmd_volume_set(last_volume)
        await self.socket_client.cmd_power(last_power)

    @callback
    def handle_socket_client_event(self, event: SqueezeEvent):
        """Process incoming event from the socket client."""
        if event == SqueezeEvent.CONNECTED:
            # restore previous power/volume
            create_task(self.restore_states())
        elif event == SqueezeEvent.DECODER_READY:
            # tell player to load next queue track
            queue = self.mass.players.get_player_queue(self.player_id)
            if queue:
                next_item = queue.next_item
                if next_item:
                    crossfade = self.mass.config.player_settings[self.player_id][
                        CONF_CROSSFADE_DURATION
                    ]
                    create_task(
                        self.socket_client.play_uri(
                            next_item.stream_url,
                            send_flush=False,
                            crossfade_duration=crossfade,
                        )
                    )
        self.update_state()

<filename>music_assistant/providers/squeezebox/__init__.py
"""Squeezebox emulated player provider."""

import asyncio
import logging
from typing import List

from music_assistant.constants import CONF_CROSSFADE_DURATION
from music_assistant.helpers.typing import MusicAssistant
from music_assistant.helpers.util import callback, create_task
from music_assistant.models.config_entry import ConfigEntry
from music_assistant.models.player import DeviceInfo, Player, PlayerFeature, PlayerState
from music_assistant.models.player_queue import QueueItem
from music_assistant.models.provider import PlayerProvider

from .constants import PROV_ID, PROV_NAME
from .discovery import DiscoveryProtocol
from .socket_client import SqueezeEvent, SqueezeSocketClient

CONF_LAST_POWER = "last_power"
CONF_LAST_VOLUME = "last_volume"

LOGGER = logging.getLogger(PROV_ID)

CONFIG_ENTRIES = []  # we don't have any provider config entries (for now)
PLAYER_FEATURES = [PlayerFeature.QUEUE, PlayerFeature.CROSSFADE, PlayerFeature.GAPLESS]
PLAYER_CONFIG_ENTRIES = []  # we don't have any player config entries (for now)


async def setup(mass):
    """Perform async setup of this Plugin/Provider."""
    prov = PySqueezeProvider()
    await mass.register_provider(prov)


class PySqueezeProvider(PlayerProvider):
    """Python implementation of SlimProto server."""

    _tasks = []

    @property
    def id(self) -> str:
        """Return provider ID for this provider."""
        return PROV_ID

    @property
    def name(self) -> str:
        """Return provider Name for this provider."""
        return PROV_NAME

    @property
    def config_entries(self) -> List[ConfigEntry]:
        """Return Config Entries for this provider."""
        return CONFIG_ENTRIES

    async def on_start(self) -> bool:
        """Handle initialization of the provider. Called on startup."""
        # start slimproto server
        create_task(asyncio.start_server(self._client_connected, "0.0.0.0", 3483))

        # setup discovery
        create_task(self.start_discovery())

    async def start_discovery(self):
        """Start discovery for players."""
        transport, _ = await self.mass.loop.create_datagram_endpoint(
            lambda: DiscoveryProtocol(self.mass.web.port),
            local_addr=("0.0.0.0", 3483),
        )
        try:
            while True:
                await asyncio.sleep(60)  # serve forever
        finally:
            transport.close()

    async def _client_connected(self, reader, writer):
        """Handle a client connection on the socket."""
        addr = writer.get_extra_info("peername")
        LOGGER.debug("Socket client connected: %s", addr)
        socket_client = SqueezeSocketClient(self.mass, reader, writer)

        def handle_event(event: SqueezeEvent, socket_client: SqueezeSocketClient):
            player_id = socket_client.player_id
            if not player_id:
                return
            # always check if we already have this player as it might be reconnected
            player = self.mass.players.get_player(player_id)
            if not player:
                player = SqueezePlayer(self.mass, socket_client)
            player.set_socket_client(socket_client)
            # just update, the playermanager will take care of adding it if it's a new player
            player.handle_socket_client_event(event)

        socket_client.register_callback(handle_event)


class SqueezePlayer(Player):
    """Squeezebox player."""

    def __init__(self, mass: MusicAssistant, socket_client: SqueezeSocketClient):
        """Initialize."""
        super().__init__()
        self.mass = mass
        self._socket_client = socket_client

    @property
    def available(self) -> bool:
        """Return current availablity of player."""
        return self._socket_client.connected

    @property
    def should_poll(self) -> bool:
        """Return True if this player should be polled for state updates."""
        return False

    @property
    def socket_client(self):
        """Return the uinderluing socket client for the player."""
        return self._socket_client

    def set_socket_client(self, socket_client: SqueezeSocketClient):
        """Set a (new) socket client to this player."""
        self._socket_client = socket_client

    async def on_remove(self) -> None:
        """Call when player is removed from the player manager."""
        self.socket_client.disconnect()

    @property
    def player_id(self) -> str:
        """Return player id (=mac address) of the player."""
        return self.socket_client.player_id

    @property
    def provider_id(self) -> str:
        """Return provider id of this player."""
        return PROV_ID

    @property
    def name(self) -> str:
        """Return name of the player."""
        return self.socket_client.name

    @property
    def volume_level(self):
        """Return current volume level of player."""
        return self.socket_client.volume_level

    @property
    def powered(self):
        """Return current power state of player."""
        return self.socket_client.powered

    @property
    def muted(self):
        """Return current mute state of player."""
        return self.socket_client.muted

    @property
    def state(self):
        """Return current state of player."""
        return PlayerState(self.socket_client.state)

    @property
    def elapsed_time(self):
        """Return elapsed_time of current playing track in (fractions of) seconds."""
        return self.socket_client.elapsed_seconds

    @property
    def elapsed_milliseconds(self) -> int:
        """Return (realtime) elapsed time of current playing media in milliseconds."""
        return self.socket_client.elapsed_milliseconds

    @property
    def current_uri(self):
        """Return uri of currently loaded track."""
        return self.socket_client.current_uri

    @property
    def features(self) -> List[PlayerFeature]:
        """Return list of features this player supports."""
        return PLAYER_FEATURES

    @property
    def config_entries(self) -> List[ConfigEntry]:
        """Return player specific config entries (if any)."""
        return PLAYER_CONFIG_ENTRIES

    @property
    def device_info(self) -> DeviceInfo:
        """Return the device info for this player."""
        return DeviceInfo(
            model=self.socket_client.device_type,
            address=self.socket_client.device_address,
        )

    async def cmd_stop(self):
        """Send stop command to player."""
        return await self.socket_client.cmd_stop()

    async def cmd_play(self):
        """Send play (unpause) command to player."""
        return await self.socket_client.cmd_play()

    async def cmd_pause(self):
        """Send pause command to player."""
        return await self.socket_client.cmd_pause()

    async def cmd_power_on(self) -> None:
        """Send POWER ON command to player."""
        # save power and volume state in cache
        cache_str = f"squeezebox_player_{self.player_id}"
        await self.mass.cache.set(cache_str, (True, self.volume_level))
        return await self.socket_client.cmd_power(True)

    async def cmd_power_off(self) -> None:
        """Send POWER OFF command to player."""
        # save power and volume state in cache
        cache_str = f"squeezebox_player_{self.player_id}"
        await self.mass.cache.set(cache_str, (False, self.volume_level))
        return await self.socket_client.cmd_power(False)

    async def cmd_volume_set(self, volume_level: int):
        """Send new volume level command to player."""
        return await self.socket_client.cmd_volume_set(volume_level)

    async def cmd_mute(self, muted: bool = False):
        """Send mute command to player."""
        return await self.socket_client.cmd_mute(muted)

    async def cmd_play_uri(self, uri: str):
        """Request player to start playing a single uri."""
        crossfade = self.mass.config.player_settings[self.player_id][
            CONF_CROSSFADE_DURATION
        ]
        return await self.socket_client.play_uri(uri, crossfade_duration=crossfade)

    async def cmd_next(self):
        """Send NEXT TRACK command to player."""
        queue = self.mass.players.get_player_queue(self.player_id)
        if queue:
            new_track = queue.get_item(queue.cur_index + 1)
            if new_track:
                return await self.cmd_play_uri(new_track.stream_url)

    async def cmd_previous(self):
        """Send PREVIOUS TRACK command to player."""
        queue = self.mass.players.get_player_queue(self.player_id)
        if queue:
            new_track = queue.get_item(queue.cur_index - 1)
            if new_track:
                return await self.cmd_play_uri(new_track.stream_url)

    async def cmd_queue_play_index(self, index: int):
        """
        Play item at index X on player's queue.

            :param index: (int) index of the queue item that should start playing
        """
        queue = self.mass.players.get_player_queue(self.player_id)
        if queue:
            new_track = queue.get_item(index)
            if new_track:
                return await self.cmd_play_uri(new_track.stream_url)

    async def cmd_queue_load(self, queue_items: List[QueueItem], repeat: bool = False):
        """
        Load/overwrite given items in the player's queue implementation.

            :param queue_items: a list of QueueItems
        """
        if queue_items:
            await self.cmd_play_uri(queue_items[0].stream_url)
            return await self.cmd_play_uri(queue_items[0].stream_url)

    async def cmd_queue_insert(
        self, queue_items: List[QueueItem], insert_at_index: int
    ):
        """
        Insert new items at position X into existing queue.

        If insert_at_index 0 or None, will start playing newly added item(s)
            :param queue_items: a list of QueueItems
            :param insert_at_index: queue position to insert new items
        """
        # queue handled by built-in queue controller
        # we only check the start index
        queue = self.mass.players.get_player_queue(self.player_id)
        if queue and insert_at_index == queue.cur_index:
            return await self.cmd_queue_play_index(insert_at_index)

    async def cmd_queue_append(self, queue_items: List[QueueItem]):
        """
        Append new items at the end of the queue.

            :param queue_items: a list of QueueItems
        """
        # automagically handled by built-in queue controller

    async def cmd_queue_update(self, queue_items: List[QueueItem]):
        """
        Overwrite the existing items in the queue, used for reordering.

            :param queue_items: a list of QueueItems
        """
        # automagically handled by built-in queue controller

    async def cmd_queue_clear(self):
        """Clear the player's queue."""
        # queue is handled by built-in queue controller but send stop
        return await self.cmd_stop()

    async def restore_states(self):
        """Restore power/volume states."""
        cache_str = f"squeezebox_player_{self.player_id}"
        cache_data = await self.mass.cache.get(cache_str)
        last_power, last_volume = cache_data if cache_data else (False, 40)
        await self.socket_client.cmd_volume_set(last_volume)
        await self.socket_client.cmd_power(last_power)

    @callback
    def handle_socket_client_event(self, event: SqueezeEvent):
        """Process incoming event from the socket client."""
        if event == SqueezeEvent.CONNECTED:
            # restore previous power/volume
            create_task(self.restore_states())
        elif event == SqueezeEvent.DECODER_READY:
            # tell player to load next queue track
            queue = self.mass.players.get_player_queue(self.player_id)
            if queue:
                next_item = queue.next_item
                if next_item:
                    crossfade = self.mass.config.player_settings[self.player_id][
                        CONF_CROSSFADE_DURATION
                    ]
                    create_task(
                        self.socket_client.play_uri(
                            next_item.stream_url,
                            send_flush=False,
                            crossfade_duration=crossfade,
                        )
                    )
        self.update_state()

Squeezebox emulated player provider. # we don't have any provider config entries (for now) # we don't have any player config entries (for now) Perform async setup of this Plugin/Provider. Python implementation of SlimProto server. Return provider ID for this provider. Return provider Name for this provider. Return Config Entries for this provider. Handle initialization of the provider. Called on startup. # start slimproto server # setup discovery Start discovery for players. # serve forever Handle a client connection on the socket. # always check if we already have this player as it might be reconnected # just update, the playermanager will take care of adding it if it's a new player Squeezebox player. Initialize. Return current availablity of player. Return True if this player should be polled for state updates. Return the uinderluing socket client for the player. Set a (new) socket client to this player. Call when player is removed from the player manager. Return player id (=mac address) of the player. Return provider id of this player. Return name of the player. Return current volume level of player. Return current power state of player. Return current mute state of player. Return current state of player. Return elapsed_time of current playing track in (fractions of) seconds. Return (realtime) elapsed time of current playing media in milliseconds. Return uri of currently loaded track. Return list of features this player supports. Return player specific config entries (if any). Return the device info for this player. Send stop command to player. Send play (unpause) command to player. Send pause command to player. Send POWER ON command to player. # save power and volume state in cache Send POWER OFF command to player. # save power and volume state in cache Send new volume level command to player. Send mute command to player. Request player to start playing a single uri. Send NEXT TRACK command to player. Send PREVIOUS TRACK command to player. Play item at index X on player's queue.              :param index: (int) index of the queue item that should start playing Load/overwrite given items in the player's queue implementation.              :param queue_items: a list of QueueItems Insert new items at position X into existing queue.          If insert_at_index 0 or None, will start playing newly added item(s)             :param queue_items: a list of QueueItems             :param insert_at_index: queue position to insert new items # queue handled by built-in queue controller # we only check the start index Append new items at the end of the queue.              :param queue_items: a list of QueueItems # automagically handled by built-in queue controller Overwrite the existing items in the queue, used for reordering.              :param queue_items: a list of QueueItems # automagically handled by built-in queue controller Clear the player's queue. # queue is handled by built-in queue controller but send stop Restore power/volume states. Process incoming event from the socket client. # restore previous power/volume # tell player to load next queue track