<commit_before>"""Config flow for Apple TV integration."""
from ipaddress import ip_address
import logging
from random import randrange

from pyatv import exceptions, pair, scan
from pyatv.const import Protocol
from pyatv.convert import protocol_str
import voluptuous as vol

from homeassistant import config_entries
from homeassistant.components import zeroconf
from homeassistant.const import CONF_ADDRESS, CONF_NAME, CONF_PIN, CONF_PROTOCOL
from homeassistant.core import callback
from homeassistant.data_entry_flow import AbortFlow, FlowResult
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.aiohttp_client import async_get_clientsession

from .const import CONF_CREDENTIALS, CONF_IDENTIFIER, CONF_START_OFF, DOMAIN

_LOGGER = logging.getLogger(__name__)

DEVICE_INPUT = "device_input"

INPUT_PIN_SCHEMA = vol.Schema({vol.Required(CONF_PIN, default=None): int})

DEFAULT_START_OFF = False
PROTOCOL_PRIORITY = [Protocol.MRP, Protocol.DMAP, Protocol.AirPlay]


async def device_scan(identifier, loop, cache=None):
    """Scan for a specific device using identifier as filter."""

    def _filter_device(dev):
        if identifier is None:
            return True
        if identifier == str(dev.address):
            return True
        if identifier == dev.name:
            return True
        return any(service.identifier == identifier for service in dev.services)

    def _host_filter():
        try:
            return [ip_address(identifier)]
        except ValueError:
            return None

    if cache:
        matches = [atv for atv in cache if _filter_device(atv)]
        if matches:
            return cache, matches[0]

    for hosts in (_host_filter(), None):
        scan_result = await scan(loop, timeout=3, hosts=hosts)
        matches = [atv for atv in scan_result if _filter_device(atv)]

        if matches:
            return scan_result, matches[0]

    return scan_result, None


def is_valid_credentials(credentials):
    """Verify that credentials are valid for establishing a connection."""
    return (
        credentials.get(Protocol.MRP.value) is not None
        or credentials.get(Protocol.DMAP.value) is not None
    )


class AppleTVConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
    """Handle a config flow for Apple TV."""

    VERSION = 1

    @staticmethod
    @callback
    def async_get_options_flow(config_entry):
        """Get options flow for this handler."""
        return AppleTVOptionsFlow(config_entry)

    def __init__(self):
        """Initialize a new AppleTVConfigFlow."""
        self.target_device = None
        self.scan_result = None
        self.atv = None
        self.protocol = None
        self.pairing = None
        self.credentials = {}  # Protocol -> credentials

    async def async_step_reauth(self, info):
        """Handle initial step when updating invalid credentials."""
        await self.async_set_unique_id(info[CONF_IDENTIFIER])
        self.target_device = info[CONF_IDENTIFIER]

        self.context["title_placeholders"] = {"name": info[CONF_NAME]}
        self.context["identifier"] = self.unique_id
        return await self.async_step_reconfigure()

    async def async_step_reconfigure(self, user_input=None):
        """Inform user that reconfiguration is about to start."""
        if user_input is not None:
            return await self.async_find_device_wrapper(
                self.async_begin_pairing, allow_exist=True
            )

        return self.async_show_form(step_id="reconfigure")

    async def async_step_user(self, user_input=None):
        """Handle the initial step."""
        # Be helpful to the user and look for devices
        if self.scan_result is None:
            self.scan_result, _ = await device_scan(None, self.hass.loop)

        errors = {}
        default_suggestion = self._prefill_identifier()
        if user_input is not None:
            self.target_device = user_input[DEVICE_INPUT]
            try:
                await self.async_find_device()
            except DeviceNotFound:
                errors["base"] = "no_devices_found"
            except DeviceAlreadyConfigured:
                errors["base"] = "already_configured"
            except exceptions.NoServiceError:
                errors["base"] = "no_usable_service"
            except Exception:  # pylint: disable=broad-except
                _LOGGER.exception("Unexpected exception")
                errors["base"] = "unknown"
            else:
                await self.async_set_unique_id(
                    self.atv.identifier, raise_on_progress=False
                )
                return await self.async_step_confirm()

        return self.async_show_form(
            step_id="user",
            data_schema=vol.Schema(
                {vol.Required(DEVICE_INPUT, default=default_suggestion): str}
            ),
            errors=errors,
            description_placeholders={"devices": self._devices_str()},
        )

    async def async_step_zeroconf(
        self, discovery_info: zeroconf.ZeroconfServiceInfo
    ) -> FlowResult:
        """Handle device found via zeroconf."""
        service_type = discovery_info[zeroconf.ATTR_TYPE]
        properties = discovery_info[zeroconf.ATTR_PROPERTIES]

        if service_type == "_mediaremotetv._tcp.local.":
            identifier = properties["UniqueIdentifier"]
            name = properties["Name"]
        elif service_type == "_touch-able._tcp.local.":
            identifier = discovery_info[zeroconf.ATTR_NAME].split(".")[0]
            name = properties["CtlN"]
        else:
            return self.async_abort(reason="unknown")

        await self.async_set_unique_id(identifier)
        self._abort_if_unique_id_configured()

        self.context["identifier"] = self.unique_id
        self.context["title_placeholders"] = {"name": name}
        self.target_device = identifier
        return await self.async_find_device_wrapper(self.async_step_confirm)

    async def async_find_device_wrapper(self, next_func, allow_exist=False):
        """Find a specific device and call another function when done.

        This function will do error handling and bail out when an error
        occurs.
        """
        try:
            await self.async_find_device(allow_exist)
        except DeviceNotFound:
            return self.async_abort(reason="no_devices_found")
        except DeviceAlreadyConfigured:
            return self.async_abort(reason="already_configured")<commit_msg>Use dataclass properties in apple_tv discovery (#60557)<commit_after>"""Config flow for Apple TV integration."""
from ipaddress import ip_address
import logging
from random import randrange

from pyatv import exceptions, pair, scan
from pyatv.const import Protocol
from pyatv.convert import protocol_str
import voluptuous as vol

from homeassistant import config_entries
from homeassistant.components import zeroconf
from homeassistant.const import CONF_ADDRESS, CONF_NAME, CONF_PIN, CONF_PROTOCOL
from homeassistant.core import callback
from homeassistant.data_entry_flow import AbortFlow, FlowResult
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.aiohttp_client import async_get_clientsession

from .const import CONF_CREDENTIALS, CONF_IDENTIFIER, CONF_START_OFF, DOMAIN

_LOGGER = logging.getLogger(__name__)

DEVICE_INPUT = "device_input"

INPUT_PIN_SCHEMA = vol.Schema({vol.Required(CONF_PIN, default=None): int})

DEFAULT_START_OFF = False
PROTOCOL_PRIORITY = [Protocol.MRP, Protocol.DMAP, Protocol.AirPlay]


async def device_scan(identifier, loop, cache=None):
    """Scan for a specific device using identifier as filter."""

    def _filter_device(dev):
        if identifier is None:
            return True
        if identifier == str(dev.address):
            return True
        if identifier == dev.name:
            return True
        return any(service.identifier == identifier for service in dev.services)

    def _host_filter():
        try:
            return [ip_address(identifier)]
        except ValueError:
            return None

    if cache:
        matches = [atv for atv in cache if _filter_device(atv)]
        if matches:
            return cache, matches[0]

    for hosts in (_host_filter(), None):
        scan_result = await scan(loop, timeout=3, hosts=hosts)
        matches = [atv for atv in scan_result if _filter_device(atv)]

        if matches:
            return scan_result, matches[0]

    return scan_result, None


def is_valid_credentials(credentials):
    """Verify that credentials are valid for establishing a connection."""
    return (
        credentials.get(Protocol.MRP.value) is not None
        or credentials.get(Protocol.DMAP.value) is not None
    )


class AppleTVConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
    """Handle a config flow for Apple TV."""

    VERSION = 1

    @staticmethod
    @callback
    def async_get_options_flow(config_entry):
        """Get options flow for this handler."""
        return AppleTVOptionsFlow(config_entry)

    def __init__(self):
        """Initialize a new AppleTVConfigFlow."""
        self.target_device = None
        self.scan_result = None
        self.atv = None
        self.protocol = None
        self.pairing = None
        self.credentials = {}  # Protocol -> credentials

    async def async_step_reauth(self, info):
        """Handle initial step when updating invalid credentials."""
        await self.async_set_unique_id(info[CONF_IDENTIFIER])
        self.target_device = info[CONF_IDENTIFIER]

        self.context["title_placeholders"] = {"name": info[CONF_NAME]}
        self.context["identifier"] = self.unique_id
        return await self.async_step_reconfigure()

    async def async_step_reconfigure(self, user_input=None):
        """Inform user that reconfiguration is about to start."""
        if user_input is not None:
            return await self.async_find_device_wrapper(
                self.async_begin_pairing, allow_exist=True
            )

        return self.async_show_form(step_id="reconfigure")

    async def async_step_user(self, user_input=None):
        """Handle the initial step."""
        # Be helpful to the user and look for devices
        if self.scan_result is None:
            self.scan_result, _ = await device_scan(None, self.hass.loop)

        errors = {}
        default_suggestion = self._prefill_identifier()
        if user_input is not None:
            self.target_device = user_input[DEVICE_INPUT]
            try:
                await self.async_find_device()
            except DeviceNotFound:
                errors["base"] = "no_devices_found"
            except DeviceAlreadyConfigured:
                errors["base"] = "already_configured"
            except exceptions.NoServiceError:
                errors["base"] = "no_usable_service"
            except Exception:  # pylint: disable=broad-except
                _LOGGER.exception("Unexpected exception")
                errors["base"] = "unknown"
            else:
                await self.async_set_unique_id(
                    self.atv.identifier, raise_on_progress=False
                )
                return await self.async_step_confirm()

        return self.async_show_form(
            step_id="user",
            data_schema=vol.Schema(
                {vol.Required(DEVICE_INPUT, default=default_suggestion): str}
            ),
            errors=errors,
            description_placeholders={"devices": self._devices_str()},
        )

    async def async_step_zeroconf(
        self, discovery_info: zeroconf.ZeroconfServiceInfo
    ) -> FlowResult:
        """Handle device found via zeroconf."""
        service_type = discovery_info.type
        properties = discovery_info.properties

        if service_type == "_mediaremotetv._tcp.local.":
            identifier = properties["UniqueIdentifier"]
            name = properties["Name"]
        elif service_type == "_touch-able._tcp.local.":
            identifier = discovery_info.name.split(".")[0]
            name = properties["CtlN"]
        else:
            return self.async_abort(reason="unknown")

        await self.async_set_unique_id(identifier)
        self._abort_if_unique_id_configured()

        self.context["identifier"] = self.unique_id
        self.context["title_placeholders"] = {"name": name}
        self.target_device = identifier
        return await self.async_find_device_wrapper(self.async_step_confirm)

    async def async_find_device_wrapper(self, next_func, allow_exist=False):
        """Find a specific device and call another function when done.

        This function will do error handling and bail out when an error
        occurs.
        """
        try:
            await self.async_find_device(allow_exist)
        except DeviceNotFound:
            return self.async_abort(reason="no_devices_found")
        except DeviceAlreadyConfigured:
            return self.async_abort(reason="already_configured")
{'fraction_non_alphanumeric': 0.08223911541119558, 'fraction_numerical': 0.001151808339092375, 'mean_word_length': 2.7115987460815045, 'pattern_counts': {'":': 10, '<': 3, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, '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': '3393803', 'n_tokens_mistral': 3556, 'n_tokens_neox': 3492, 'n_words': 901}