<commit_before>"""Provide functionality to interact with Cast devices on the network."""
from __future__ import annotations

import asyncio
from collections.abc import Callable
from contextlib import suppress
from datetime import datetime
import json
import logging

import pychromecast
from pychromecast.controllers.homeassistant import HomeAssistantController
from pychromecast.controllers.media import (
    MEDIA_PLAYER_ERROR_CODES,
    MEDIA_PLAYER_STATE_BUFFERING,
    MEDIA_PLAYER_STATE_PLAYING,
    MEDIA_PLAYER_STATE_UNKNOWN,
)
from pychromecast.controllers.multizone import MultizoneManager
from pychromecast.controllers.receiver import VOLUME_CONTROL_TYPE_FIXED
from pychromecast.quick_play import quick_play
from pychromecast.socket_client import (
    CONNECTION_STATUS_CONNECTED,
    CONNECTION_STATUS_DISCONNECTED,
)
import voluptuous as vol
import yarl

from homeassistant.components import media_source, zeroconf
from homeassistant.components.media_player import (
    BrowseError,
    BrowseMedia,
    MediaPlayerDeviceClass,
    MediaPlayerEntity,
    MediaPlayerEntityFeature,
    async_process_play_media_url,
)
from homeassistant.components.media_player.const import (
    ATTR_MEDIA_EXTRA,
    MEDIA_CLASS_DIRECTORY,
    MEDIA_TYPE_MOVIE,
    MEDIA_TYPE_MUSIC,
    MEDIA_TYPE_TVSHOW,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
    CAST_APP_ID_HOMEASSISTANT_LOVELACE,
    EVENT_HOMEASSISTANT_STOP,
    STATE_BUFFERING,
    STATE_IDLE,
    STATE_OFF,
    STATE_PAUSED,
    STATE_PLAYING,
)
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity import DeviceInfo
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.network import NoURLAvailableError, get_url, is_hass_url
import homeassistant.util.dt as dt_util
from homeassistant.util.logging import async_create_catching_coro

from .const import (
    ADDED_CAST_DEVICES_KEY,
    CAST_MULTIZONE_MANAGER_KEY,
    CONF_IGNORE_CEC,
    CONF_UUID,
    DOMAIN as CAST_DOMAIN,
    SIGNAL_CAST_DISCOVERED,
    SIGNAL_CAST_REMOVED,
    SIGNAL_HASS_CAST_SHOW_VIEW,
)
from .discovery import setup_internal_discovery
from .helpers import (
    CastStatusListener,
    ChromecastInfo,
    ChromeCastZeroconf,
    PlaylistError,
    PlaylistSupported,
    parse_playlist,
)

_LOGGER = logging.getLogger(__name__)

APP_IDS_UNRELIABLE_MEDIA_INFO = ("Netflix",)

CAST_SPLASH = "https://www.home-assistant.io/images/cast/splash.png"

ENTITY_SCHEMA = vol.All(
    vol.Schema(
        {
            vol.Optional(CONF_UUID): cv.string,
            vol.Optional(CONF_IGNORE_CEC): vol.All(cv.ensure_list, [cv.string]),
        }
    ),
)


@callback
def _async_create_cast_device(hass: HomeAssistant, info: ChromecastInfo):
    """Create a CastDevice entity or dynamic group from the chromecast object.

    Returns None if the cast device has already been added.
    """
    _LOGGER.debug("_async_create_cast_device: %s", info)
    if info.uuid is None:
        _LOGGER.error("_async_create_cast_device uuid none: %s", info)
        return None

    # Found a cast with UUID
    added_casts = hass.data[ADDED_CAST_DEVICES_KEY]
    if info.uuid in added_casts:
        # Already added this one, the entity will take care of moved hosts
        # itself
        return None
    # -> New cast device
    added_casts.add(info.uuid)

    if info.is_dynamic_group:
        # This is a dynamic group, do not add it but connect to the service.
        group = DynamicCastGroup(hass, info)
        group.async_setup()
        return None

    return CastMediaPlayerEntity(hass, info)


async def async_setup_entry(
    hass: HomeAssistant,
    config_entry: ConfigEntry,
    async_add_entities: AddEntitiesCallback,
) -> None:
    """Set up Cast from a config entry."""
    hass.data.setdefault(ADDED_CAST_DEVICES_KEY, set())

    # Import CEC IGNORE attributes
    pychromecast.IGNORE_CEC += config_entry.data.get(CONF_IGNORE_CEC) or []

    wanted_uuids = config_entry.data.get(CONF_UUID) or None

    @callback
    def async_cast_discovered(discover: ChromecastInfo) -> None:
        """Handle discovery of a new chromecast."""
        # If wanted_uuids is set, we're only accepting specific cast devices identified
        # by UUID
        if wanted_uuids is not None and str(discover.uuid) not in wanted_uuids:
            # UUID not matching, ignore.
            return

        cast_device = _async_create_cast_device(hass, discover)
        if cast_device is not None:
            async_add_entities([cast_device])

    async_dispatcher_connect(hass, SIGNAL_CAST_DISCOVERED, async_cast_discovered)
    ChromeCastZeroconf.set_zeroconf(await zeroconf.async_get_instance(hass))
    hass.async_add_executor_job(setup_internal_discovery, hass, config_entry)


class CastDevice:
    """Representation of a Cast device or dynamic group on the network.

    This class is the holder of the pychromecast.Chromecast object and its
    socket client. It therefore handles all reconnects and audio groups changing
    "elected leader" itself.
    """

    _mz_only: bool

    def __init__(self, hass: HomeAssistant, cast_info: ChromecastInfo) -> None:
        """Initialize the cast device."""

        self.hass: HomeAssistant = hass
        self._cast_info = cast_info
        self._chromecast: pychromecast.Chromecast | None = None
        self.mz_mgr = None
        self._status_listener: CastStatusListener | None = None
        self._add_remove_handler: Callable[[], None] | None = None
        self._del_remove_handler: Callable[[], None] | None = None
        self._name: str | None = None

    def _async_setup(self, name: str) -> None:
        """Create chromecast object."""
        self._name = name
        self._add_remove_handler = async_dispatcher_connect(
            self.hass, SIGNAL_CAST_DISCOVERED, self._async_cast_discovered
        )
        self._del_remove_handler = async_dispatcher_connect(
            self.hass, SIGNAL_CAST_REMOVED, self._async_cast_removed
        )
        self.hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, self._async_stop)
        # asyncio.create_task is used to avoid delaying startup wrapup if the device
        # is discovered already during startup but then fails to respond
        asyncio.create_task(
            async_create_catching_coro(self._async_connect_to_chromecast())
        )

    async def _async_tear_down(self) -> None:
        """Disconnect chromecast object and remove listeners."""
        await self._async_disconnect()
        if self._cast_info.uuid is not None:
            # Remove the entity from the added casts so that it can dynamically
            # be re-added again.
            self.hass.data[ADDED_CAST_DEVICES_KEY].remove(self._cast_info.uuid)
        if self._add_remove_handler:
            self._add_remove_handler()
            self._add_remove_handler = None
        if self._del_remove_handler:
            self._del_remove_handler()
            self._del_remove_handler = None

    async def _async_connect_to_chromecast(self):
        """Set up the chromecast object."""
        _LOGGER.debug(
            "[%s %s] Connecting to cast device by service %s",
            self._name,
            self._cast_info.friendly_name,
            self._cast_info.cast_info.services,
        )
        chromecast = await self.hass.async_add_executor_job(
            pychromecast.get_chromecast_from_cast_info,
            self._cast_info.cast_info,
            ChromeCastZeroconf.get_zeroconf(),
        )
        self._chromecast = chromecast

        if CAST_MULTIZONE_MANAGER_KEY not in self.hass.data:
            self.hass.data[CAST_MULTIZONE_MANAGER_KEY] = MultizoneManager()

        self.mz_mgr = self.hass.data[CAST_MULTIZONE_MANAGER_KEY]

        self._status_listener = CastStatusListener(
            self, chromecast, self.mz_mgr, self._mz_only
        )
        self._chromecast.start()

    async def _async_disconnect(self):
        """Disconnect Chromecast object if it is set."""
        if self._chromecast is not None:
            _LOGGER.debug(
                "[%s %s] Disconnecting from chromecast socket",
                self._name,
                self._cast_info.friendly_name,
            )
            await self.hass.async_add_executor_job(self._chromecast.disconnect)

        self._invalidate()

    def _invalidate(self):
        """Invalidate some attributes."""
        self._chromecast = None
        self.mz_mgr = None
        if self._status_listener is not None:
            self._status_listener.invalidate()
            self._status_listener = None

    async def _async_cast_discovered(self, discover: ChromecastInfo):
        """Handle discovery of new Chromecast."""
        if self._cast_info.uuid != discover.uuid:
            # Discovered is not our device.
            return

        _LOGGER.debug("Discovered chromecast with same UUID: %s", discover)
        self._cast_info = discover

    async def _async_cast_removed(self, discover: ChromecastInfo):
        """Handle removal of Chromecast."""

    async def _async_stop(self, event):
        """Disconnect socket on Home Assistant stop."""
        await self._async_disconnect()


class CastMediaPlayerEntity(CastDevice, MediaPlayerEntity):
    """Representation of a Cast device on the network."""

    _attr_should_poll = False
    _attr_media_image_remotely_accessible = True
    _mz_only = False

    def __init__(self, hass: HomeAssistant, cast_info: ChromecastInfo) -> None:
        """Initialize the cast device."""

        CastDevice.__init__(self, hass, cast_info)

        self.cast_status = None
        self.media_status = None
        self.media_status_received = None
        self.mz_media_status: dict[str, pychromecast.controllers.media.MediaStatus] = {}
        self.mz_media_status_received: dict[str, datetime] = {}
        self._attr_available = False
        self._hass_cast_controller: HomeAssistantController | None = None

        self._cast_view_remove_handler = None
        self._attr_unique_id = str(cast_info.uuid)
        self._attr_name = cast_info.friendly_name
        self._attr_device_info = DeviceInfo(
            identifiers={(CAST_DOMAIN, str(cast_info.uuid).replace("-", ""))},
            manufacturer=str(cast_info.cast_info.manufacturer),
            model=cast_info.cast_info.model_name,
            name=str(cast_info.friendly_name),
        )

        if cast_info.cast_info.cast_type in [
            pychromecast.const.CAST_TYPE_AUDIO,
            pychromecast.const.CAST_TYPE_GROUP,
        ]:
            self._attr_device_class = MediaPlayerDeviceClass.SPEAKER

    async def async_added_to_hass(self):
        """Create chromecast object when added to hass."""
        self._async_setup(self.entity_id)

        self._cast_view_remove_handler = async_dispatcher_connect(
            self.hass, SIGNAL_HASS_CAST_SHOW_VIEW, self._handle_signal_show_view
        )

    async def async_will_remove_from_hass(self) -> None:<commit_msg>Migrate Cast to new entity naming style (#74901)<commit_after>"""Provide functionality to interact with Cast devices on the network."""
from __future__ import annotations

import asyncio
from collections.abc import Callable
from contextlib import suppress
from datetime import datetime
import json
import logging

import pychromecast
from pychromecast.controllers.homeassistant import HomeAssistantController
from pychromecast.controllers.media import (
    MEDIA_PLAYER_ERROR_CODES,
    MEDIA_PLAYER_STATE_BUFFERING,
    MEDIA_PLAYER_STATE_PLAYING,
    MEDIA_PLAYER_STATE_UNKNOWN,
)
from pychromecast.controllers.multizone import MultizoneManager
from pychromecast.controllers.receiver import VOLUME_CONTROL_TYPE_FIXED
from pychromecast.quick_play import quick_play
from pychromecast.socket_client import (
    CONNECTION_STATUS_CONNECTED,
    CONNECTION_STATUS_DISCONNECTED,
)
import voluptuous as vol
import yarl

from homeassistant.components import media_source, zeroconf
from homeassistant.components.media_player import (
    BrowseError,
    BrowseMedia,
    MediaPlayerDeviceClass,
    MediaPlayerEntity,
    MediaPlayerEntityFeature,
    async_process_play_media_url,
)
from homeassistant.components.media_player.const import (
    ATTR_MEDIA_EXTRA,
    MEDIA_CLASS_DIRECTORY,
    MEDIA_TYPE_MOVIE,
    MEDIA_TYPE_MUSIC,
    MEDIA_TYPE_TVSHOW,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
    CAST_APP_ID_HOMEASSISTANT_LOVELACE,
    EVENT_HOMEASSISTANT_STOP,
    STATE_BUFFERING,
    STATE_IDLE,
    STATE_OFF,
    STATE_PAUSED,
    STATE_PLAYING,
)
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity import DeviceInfo
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.network import NoURLAvailableError, get_url, is_hass_url
import homeassistant.util.dt as dt_util
from homeassistant.util.logging import async_create_catching_coro

from .const import (
    ADDED_CAST_DEVICES_KEY,
    CAST_MULTIZONE_MANAGER_KEY,
    CONF_IGNORE_CEC,
    CONF_UUID,
    DOMAIN as CAST_DOMAIN,
    SIGNAL_CAST_DISCOVERED,
    SIGNAL_CAST_REMOVED,
    SIGNAL_HASS_CAST_SHOW_VIEW,
)
from .discovery import setup_internal_discovery
from .helpers import (
    CastStatusListener,
    ChromecastInfo,
    ChromeCastZeroconf,
    PlaylistError,
    PlaylistSupported,
    parse_playlist,
)

_LOGGER = logging.getLogger(__name__)

APP_IDS_UNRELIABLE_MEDIA_INFO = ("Netflix",)

CAST_SPLASH = "https://www.home-assistant.io/images/cast/splash.png"

ENTITY_SCHEMA = vol.All(
    vol.Schema(
        {
            vol.Optional(CONF_UUID): cv.string,
            vol.Optional(CONF_IGNORE_CEC): vol.All(cv.ensure_list, [cv.string]),
        }
    ),
)


@callback
def _async_create_cast_device(hass: HomeAssistant, info: ChromecastInfo):
    """Create a CastDevice entity or dynamic group from the chromecast object.

    Returns None if the cast device has already been added.
    """
    _LOGGER.debug("_async_create_cast_device: %s", info)
    if info.uuid is None:
        _LOGGER.error("_async_create_cast_device uuid none: %s", info)
        return None

    # Found a cast with UUID
    added_casts = hass.data[ADDED_CAST_DEVICES_KEY]
    if info.uuid in added_casts:
        # Already added this one, the entity will take care of moved hosts
        # itself
        return None
    # -> New cast device
    added_casts.add(info.uuid)

    if info.is_dynamic_group:
        # This is a dynamic group, do not add it but connect to the service.
        group = DynamicCastGroup(hass, info)
        group.async_setup()
        return None

    return CastMediaPlayerEntity(hass, info)


async def async_setup_entry(
    hass: HomeAssistant,
    config_entry: ConfigEntry,
    async_add_entities: AddEntitiesCallback,
) -> None:
    """Set up Cast from a config entry."""
    hass.data.setdefault(ADDED_CAST_DEVICES_KEY, set())

    # Import CEC IGNORE attributes
    pychromecast.IGNORE_CEC += config_entry.data.get(CONF_IGNORE_CEC) or []

    wanted_uuids = config_entry.data.get(CONF_UUID) or None

    @callback
    def async_cast_discovered(discover: ChromecastInfo) -> None:
        """Handle discovery of a new chromecast."""
        # If wanted_uuids is set, we're only accepting specific cast devices identified
        # by UUID
        if wanted_uuids is not None and str(discover.uuid) not in wanted_uuids:
            # UUID not matching, ignore.
            return

        cast_device = _async_create_cast_device(hass, discover)
        if cast_device is not None:
            async_add_entities([cast_device])

    async_dispatcher_connect(hass, SIGNAL_CAST_DISCOVERED, async_cast_discovered)
    ChromeCastZeroconf.set_zeroconf(await zeroconf.async_get_instance(hass))
    hass.async_add_executor_job(setup_internal_discovery, hass, config_entry)


class CastDevice:
    """Representation of a Cast device or dynamic group on the network.

    This class is the holder of the pychromecast.Chromecast object and its
    socket client. It therefore handles all reconnects and audio groups changing
    "elected leader" itself.
    """

    _mz_only: bool

    def __init__(self, hass: HomeAssistant, cast_info: ChromecastInfo) -> None:
        """Initialize the cast device."""

        self.hass: HomeAssistant = hass
        self._cast_info = cast_info
        self._chromecast: pychromecast.Chromecast | None = None
        self.mz_mgr = None
        self._status_listener: CastStatusListener | None = None
        self._add_remove_handler: Callable[[], None] | None = None
        self._del_remove_handler: Callable[[], None] | None = None
        self._name: str | None = None

    def _async_setup(self, name: str) -> None:
        """Create chromecast object."""
        self._name = name
        self._add_remove_handler = async_dispatcher_connect(
            self.hass, SIGNAL_CAST_DISCOVERED, self._async_cast_discovered
        )
        self._del_remove_handler = async_dispatcher_connect(
            self.hass, SIGNAL_CAST_REMOVED, self._async_cast_removed
        )
        self.hass.bus.async_listen_once(EVENT_HOMEASSISTANT_STOP, self._async_stop)
        # asyncio.create_task is used to avoid delaying startup wrapup if the device
        # is discovered already during startup but then fails to respond
        asyncio.create_task(
            async_create_catching_coro(self._async_connect_to_chromecast())
        )

    async def _async_tear_down(self) -> None:
        """Disconnect chromecast object and remove listeners."""
        await self._async_disconnect()
        if self._cast_info.uuid is not None:
            # Remove the entity from the added casts so that it can dynamically
            # be re-added again.
            self.hass.data[ADDED_CAST_DEVICES_KEY].remove(self._cast_info.uuid)
        if self._add_remove_handler:
            self._add_remove_handler()
            self._add_remove_handler = None
        if self._del_remove_handler:
            self._del_remove_handler()
            self._del_remove_handler = None

    async def _async_connect_to_chromecast(self):
        """Set up the chromecast object."""
        _LOGGER.debug(
            "[%s %s] Connecting to cast device by service %s",
            self._name,
            self._cast_info.friendly_name,
            self._cast_info.cast_info.services,
        )
        chromecast = await self.hass.async_add_executor_job(
            pychromecast.get_chromecast_from_cast_info,
            self._cast_info.cast_info,
            ChromeCastZeroconf.get_zeroconf(),
        )
        self._chromecast = chromecast

        if CAST_MULTIZONE_MANAGER_KEY not in self.hass.data:
            self.hass.data[CAST_MULTIZONE_MANAGER_KEY] = MultizoneManager()

        self.mz_mgr = self.hass.data[CAST_MULTIZONE_MANAGER_KEY]

        self._status_listener = CastStatusListener(
            self, chromecast, self.mz_mgr, self._mz_only
        )
        self._chromecast.start()

    async def _async_disconnect(self):
        """Disconnect Chromecast object if it is set."""
        if self._chromecast is not None:
            _LOGGER.debug(
                "[%s %s] Disconnecting from chromecast socket",
                self._name,
                self._cast_info.friendly_name,
            )
            await self.hass.async_add_executor_job(self._chromecast.disconnect)

        self._invalidate()

    def _invalidate(self):
        """Invalidate some attributes."""
        self._chromecast = None
        self.mz_mgr = None
        if self._status_listener is not None:
            self._status_listener.invalidate()
            self._status_listener = None

    async def _async_cast_discovered(self, discover: ChromecastInfo):
        """Handle discovery of new Chromecast."""
        if self._cast_info.uuid != discover.uuid:
            # Discovered is not our device.
            return

        _LOGGER.debug("Discovered chromecast with same UUID: %s", discover)
        self._cast_info = discover

    async def _async_cast_removed(self, discover: ChromecastInfo):
        """Handle removal of Chromecast."""

    async def _async_stop(self, event):
        """Disconnect socket on Home Assistant stop."""
        await self._async_disconnect()


class CastMediaPlayerEntity(CastDevice, MediaPlayerEntity):
    """Representation of a Cast device on the network."""

    _attr_has_entity_name = True
    _attr_should_poll = False
    _attr_media_image_remotely_accessible = True
    _mz_only = False

    def __init__(self, hass: HomeAssistant, cast_info: ChromecastInfo) -> None:
        """Initialize the cast device."""

        CastDevice.__init__(self, hass, cast_info)

        self.cast_status = None
        self.media_status = None
        self.media_status_received = None
        self.mz_media_status: dict[str, pychromecast.controllers.media.MediaStatus] = {}
        self.mz_media_status_received: dict[str, datetime] = {}
        self._attr_available = False
        self._hass_cast_controller: HomeAssistantController | None = None

        self._cast_view_remove_handler = None
        self._attr_unique_id = str(cast_info.uuid)
        self._attr_device_info = DeviceInfo(
            identifiers={(CAST_DOMAIN, str(cast_info.uuid).replace("-", ""))},
            manufacturer=str(cast_info.cast_info.manufacturer),
            model=cast_info.cast_info.model_name,
            name=str(cast_info.friendly_name),
        )

        if cast_info.cast_info.cast_type in [
            pychromecast.const.CAST_TYPE_AUDIO,
            pychromecast.const.CAST_TYPE_GROUP,
        ]:
            self._attr_device_class = MediaPlayerDeviceClass.SPEAKER

    async def async_added_to_hass(self):
        """Create chromecast object when added to hass."""
        self._async_setup(self.entity_id)

        self._cast_view_remove_handler = async_dispatcher_connect(
            self.hass, SIGNAL_HASS_CAST_SHOW_VIEW, self._handle_signal_show_view
        )

    async def async_will_remove_from_hass(self) -> None:
{'fraction_non_alphanumeric': 0.07289059706828745, 'fraction_numerical': 0.0002234537003932785, 'mean_word_length': 3.125553097345133, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 19, 'https://': 2, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5821943', 'n_tokens_mistral': 6947, 'n_tokens_neox': 6813, 'n_words': 1522}