<commit_before>"""Bluetooth support for esphome."""
from __future__ import annotations

import logging

from aioesphomeapi import APIClient

from homeassistant.components.bluetooth import (
    HaBluetoothConnector,
    async_get_advertisement_callback,
    async_register_scanner,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import (
    CALLBACK_TYPE,
    HomeAssistant,
    async_get_hass,
    callback as hass_callback,
)

from ..domain_data import DomainData
from ..entry_data import RuntimeEntryData
from .client import ESPHomeClient
from .scanner import ESPHomeScanner

_LOGGER = logging.getLogger(__name__)


@hass_callback
def async_can_connect(source: str) -> bool:
    """Check if a given source can make another connection."""
    domain_data = DomainData.get(async_get_hass())
    entry = domain_data.get_by_unique_id(source)
    entry_data = domain_data.get_entry_data(entry)
    _LOGGER.debug(
        "Checking if %s can connect, available=%s, ble_connections_free=%s",
        source,
        entry_data.available,
        entry_data.ble_connections_free,
    )
    return bool(entry_data.available and entry_data.ble_connections_free)


async def async_connect_scanner(
    hass: HomeAssistant,
    entry: ConfigEntry,
    cli: APIClient,
    entry_data: RuntimeEntryData,
) -> CALLBACK_TYPE:
    """Connect scanner."""
    assert entry.unique_id is not None
    source = str(entry.unique_id)
    new_info_callback = async_get_advertisement_callback(hass)
    assert entry_data.device_info is not None
    version = entry_data.device_info.bluetooth_proxy_version
    connectable = version >= 2
    _LOGGER.debug(
        "Connecting scanner for %s, version=%s, connectable=%s",
        source,
        version,
        connectable,
    )
    connector = HaBluetoothConnector(
        client=ESPHomeClient,
        source=source,
        can_connect=lambda: async_can_connect(source),
    )
    scanner = ESPHomeScanner(hass, source, new_info_callback, connector, connectable)
    unload_callbacks = [
        async_register_scanner(hass, scanner, connectable),
        scanner.async_setup(),
    ]
    await cli.subscribe_bluetooth_le_advertisements(scanner.async_on_advertisement)
    if connectable:<commit_msg>Fix Bluetooth failover when esphome device unexpectedly disconnects (#79769)<commit_after>"""Bluetooth support for esphome."""
from __future__ import annotations

from collections.abc import Callable
import logging

from aioesphomeapi import APIClient

from homeassistant.components.bluetooth import (
    HaBluetoothConnector,
    async_get_advertisement_callback,
    async_register_scanner,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import CALLBACK_TYPE, HomeAssistant, callback as hass_callback

from ..entry_data import RuntimeEntryData
from .client import ESPHomeClient
from .scanner import ESPHomeScanner

_LOGGER = logging.getLogger(__name__)


@hass_callback
def _async_can_connect_factory(
    entry_data: RuntimeEntryData, source: str
) -> Callable[[], bool]:
    """Create a can_connect function for a specific RuntimeEntryData instance."""

    @hass_callback
    def _async_can_connect() -> bool:
        """Check if a given source can make another connection."""
        _LOGGER.debug(
            "Checking if %s can connect, available=%s, ble_connections_free=%s",
            source,
            entry_data.available,
            entry_data.ble_connections_free,
        )
        return bool(entry_data.available and entry_data.ble_connections_free)

    return _async_can_connect


async def async_connect_scanner(
    hass: HomeAssistant,
    entry: ConfigEntry,
    cli: APIClient,
    entry_data: RuntimeEntryData,
) -> CALLBACK_TYPE:
    """Connect scanner."""
    assert entry.unique_id is not None
    source = str(entry.unique_id)
    new_info_callback = async_get_advertisement_callback(hass)
    assert entry_data.device_info is not None
    version = entry_data.device_info.bluetooth_proxy_version
    connectable = version >= 2
    _LOGGER.debug(
        "Connecting scanner for %s, version=%s, connectable=%s",
        source,
        version,
        connectable,
    )
    connector = HaBluetoothConnector(
        client=ESPHomeClient,
        source=source,
        can_connect=_async_can_connect_factory(entry_data, source),
    )
    scanner = ESPHomeScanner(hass, source, new_info_callback, connector, connectable)
    unload_callbacks = [
        async_register_scanner(hass, scanner, connectable),
        scanner.async_setup(),
    ]
    await cli.subscribe_bluetooth_le_advertisements(scanner.async_on_advertisement)
    if connectable:
{'fraction_non_alphanumeric': 0.07432721059376335, 'fraction_numerical': 0.00149508756941478, 'mean_word_length': 3.9191176470588234, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 10, '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': '5708823', 'n_tokens_mistral': 1417, 'n_tokens_neox': 1383, 'n_words': 319}