<commit_before>"""Offer reusable conditions."""
from datetime import timedelta
import logging
import sys

from homeassistant.helpers.typing import ConfigType

from homeassistant.core import HomeAssistant
from homeassistant.components import (
    zone as zone_cmp, sun as sun_cmp)
from homeassistant.const import (
    ATTR_GPS_ACCURACY, ATTR_LATITUDE, ATTR_LONGITUDE,
    CONF_ENTITY_ID, CONF_VALUE_TEMPLATE, CONF_CONDITION,
    WEEKDAYS, CONF_STATE, CONF_ZONE, CONF_BEFORE,
    CONF_AFTER, CONF_WEEKDAY, SUN_EVENT_SUNRISE, SUN_EVENT_SUNSET,
    CONF_BELOW, CONF_ABOVE)
from homeassistant.exceptions import TemplateError, HomeAssistantError
import homeassistant.helpers.config_validation as cv
import homeassistant.util.dt as dt_util
from homeassistant.util.async import run_callback_threadsafe

FROM_CONFIG_FORMAT = '{}_from_config'

_LOGGER = logging.getLogger(__name__)


def from_config(config: ConfigType, config_validation: bool=True):
    """Turn a condition configuration into a method."""
    factory = getattr(
        sys.modules[__name__],
        FROM_CONFIG_FORMAT.format(config.get(CONF_CONDITION)), None)

    if factory is None:
        raise HomeAssistantError('Invalid condition "{}" specified {}'.format(
            config.get(CONF_CONDITION), config))

    return factory(config, config_validation)


def and_from_config(config: ConfigType, config_validation: bool=True):
    """Create multi condition matcher using 'AND'."""
    if config_validation:
        config = cv.AND_CONDITION_SCHEMA(config)
    checks = [from_config(entry, False) for entry in config['conditions']]

    def if_and_condition(hass: HomeAssistant,
                         variables=None) -> bool:
        """Test and condition."""
        for check in checks:
            try:
                if not check(hass, variables):
                    return False
            except Exception as ex:  # pylint: disable=broad-except
                _LOGGER.warning('Error during and-condition: %s', ex)
                return False

        return True

    return if_and_condition


def or_from_config(config: ConfigType, config_validation: bool=True):
    """Create multi condition matcher using 'OR'."""
    if config_validation:
        config = cv.OR_CONDITION_SCHEMA(config)
    checks = [from_config(entry, False) for entry in config['conditions']]

    def if_or_condition(hass: HomeAssistant,
                        variables=None) -> bool:
        """Test and condition."""
        for check in checks:
            try:
                if check(hass, variables):
                    return True
            except Exception as ex:  # pylint: disable=broad-except
                _LOGGER.warning('Error during or-condition: %s', ex)

        return False

    return if_or_condition


# pylint: disable=too-many-arguments
def numeric_state(hass: HomeAssistant, entity, below=None, above=None,
                  value_template=None, variables=None):
    """Test a numeric state condition."""
    return run_callback_threadsafe(
        hass.loop, async_numeric_state, hass, entity, below, above,
        value_template, variables,
    ).result()


def async_numeric_state(hass: HomeAssistant, entity, below=None, above=None,
                        value_template=None, variables=None):
    """Test a numeric state condition."""
    if isinstance(entity, str):
        entity = hass.states.get(entity)

    if entity is None:
        return False

    if value_template is None:
        value = entity.state
    else:
        variables = dict(variables or {})
        variables['state'] = entity
        try:
            value = value_template.async_render(variables)
        except TemplateError as ex:
            _LOGGER.error("Template error: %s", ex)
            return False

    try:
        value = float(value)
    except ValueError:
        _LOGGER.warning("Value cannot be processed as a number: %s", value)
        return False

    if below is not None and value > below:
        return False

    if above is not None and value < above:
        return False

    return True


def numeric_state_from_config(config, config_validation=True):
    """Wrap action method with state based condition."""
    if config_validation:
        config = cv.NUMERIC_STATE_CONDITION_SCHEMA(config)
    entity_id = config.get(CONF_ENTITY_ID)
    below = config.get(CONF_BELOW)
    above = config.get(CONF_ABOVE)
    value_template = config.get(CONF_VALUE_TEMPLATE)

    def if_numeric_state(hass, variables=None):
        """Test numeric state condition."""
        if value_template is not None:
            value_template.hass = hass

        return numeric_state(hass, entity_id, below, above, value_template,
                             variables)

    return if_numeric_state


def state(hass, entity, req_state, for_period=None):
    """Test if state matches requirements."""
    if isinstance(entity, str):
        entity = hass.states.get(entity)

    if entity is None:
        return False

    is_state = entity.state == req_state

    if for_period is None or not is_state:
        return is_state

    return dt_util.utcnow() - for_period > entity.last_changed


def state_from_config(config, config_validation=True):
    """Wrap action method with state based condition."""
    if config_validation:
        config = cv.STATE_CONDITION_SCHEMA(config)
    entity_id = config.get(CONF_ENTITY_ID)
    req_state = config.get(CONF_STATE)
    for_period = config.get('for')

    def if_state(hass, variables=None):
        """Test if condition."""
        return state(hass, entity_id, req_state, for_period)

    return if_state


def sun(hass, before=None, after=None, before_offset=None, after_offset=None):
    """Test if current time matches sun requirements."""
    now = dt_util.now().time()
    before_offset = before_offset or timedelta(0)
    after_offset = after_offset or timedelta(0)

    if before == SUN_EVENT_SUNRISE and now > (sun_cmp.next_rising(hass) +
                                              before_offset).time():
        return False

    elif before == SUN_EVENT_SUNSET and now > (sun_cmp.next_setting(hass) +
                                               before_offset).time():
        return False

    if after == SUN_EVENT_SUNRISE and now < (sun_cmp.next_rising(hass) +
                                             after_offset).time():
        return False

    elif after == SUN_EVENT_SUNSET and now < (sun_cmp.next_setting(hass) +
                                              after_offset).time():
        return False

    return True


def sun_from_config(config, config_validation=True):
    """Wrap action method with sun based condition."""
    if config_validation:
        config = cv.SUN_CONDITION_SCHEMA(config)
    before = config.get('before')
    after = config.get('after')
    before_offset = config.get('before_offset')
    after_offset = config.get('after_offset')

    def time_if(hass, variables=None):
        """Validate time based if-condition."""
        return sun(hass, before, after, before_offset, after_offset)

    return time_if


def template(hass, value_template, variables=None):
    """Test if template condition matches."""
    return run_callback_threadsafe(
        hass.loop, async_template, hass, value_template, variables,
    ).result()


def async_template(hass, value_template, variables=None):
    """Test if template condition matches."""
    try:
        value = value_template.async_render(variables)
    except TemplateError as ex:
        _LOGGER.error('Error duriong template condition: %s', ex)
        return False

    return value.lower() == 'true'


def template_from_config(config, config_validation=True):
    """Wrap action method with state based condition."""
    if config_validation:
        config = cv.TEMPLATE_CONDITION_SCHEMA(config)
    value_template = config.get(CONF_VALUE_TEMPLATE)

    def template_if(hass, variables=None):
        """Validate template based if-condition."""
        value_template.hass = hass

        return template(hass, value_template, variables)

    return template_if


def time(before=None, after=None, weekday=None):
    """Test if local time condition matches.
<commit_msg>Make helpers.condition.* async<commit_after>"""Offer reusable conditions."""
from datetime import timedelta
import logging
import sys

from homeassistant.helpers.typing import ConfigType

from homeassistant.core import HomeAssistant
from homeassistant.components import (
    zone as zone_cmp, sun as sun_cmp)
from homeassistant.const import (
    ATTR_GPS_ACCURACY, ATTR_LATITUDE, ATTR_LONGITUDE,
    CONF_ENTITY_ID, CONF_VALUE_TEMPLATE, CONF_CONDITION,
    WEEKDAYS, CONF_STATE, CONF_ZONE, CONF_BEFORE,
    CONF_AFTER, CONF_WEEKDAY, SUN_EVENT_SUNRISE, SUN_EVENT_SUNSET,
    CONF_BELOW, CONF_ABOVE)
from homeassistant.exceptions import TemplateError, HomeAssistantError
import homeassistant.helpers.config_validation as cv
import homeassistant.util.dt as dt_util
from homeassistant.util.async import run_callback_threadsafe

FROM_CONFIG_FORMAT = '{}_from_config'
ASYNC_FROM_CONFIG_FORMAT = 'async_{}_from_config'

_LOGGER = logging.getLogger(__name__)

# PyLint does not like the use of _threaded_factory
# pylint: disable=invalid-name


def _threaded_factory(async_factory):
    """Helper method to create threaded versions of async factories."""
    def factory(config, config_validation=True):
        """Threaded factory."""
        async_check = async_factory(config, config_validation)

        def condition_if(hass, variables=None):
            """Validate condition."""
            return run_callback_threadsafe(
                hass.loop, async_check, hass, variables,
            ).result()

        return condition_if

    return factory


def async_from_config(config: ConfigType, config_validation: bool=True):
    """Turn a condition configuration into a method.

    Should be run on the event loop.
    """
    for fmt in (ASYNC_FROM_CONFIG_FORMAT, FROM_CONFIG_FORMAT):
        factory = getattr(
            sys.modules[__name__],
            fmt.format(config.get(CONF_CONDITION)), None)

        if factory:
            break

    if factory is None:
        raise HomeAssistantError('Invalid condition "{}" specified {}'.format(
            config.get(CONF_CONDITION), config))

    return factory(config, config_validation)


from_config = _threaded_factory(async_from_config)


def async_and_from_config(config: ConfigType, config_validation: bool=True):
    """Create multi condition matcher using 'AND'."""
    if config_validation:
        config = cv.AND_CONDITION_SCHEMA(config)
    checks = None

    def if_and_condition(hass: HomeAssistant,
                         variables=None) -> bool:
        """Test and condition."""
        nonlocal checks

        if checks is None:
            checks = [async_from_config(entry, False) for entry
                      in config['conditions']]

        try:
            for check in checks:
                if not check(hass, variables):
                    return False
        except Exception as ex:  # pylint: disable=broad-except
            _LOGGER.warning('Error during and-condition: %s', ex)
            return False

        return True

    return if_and_condition


and_from_config = _threaded_factory(async_and_from_config)


def async_or_from_config(config: ConfigType, config_validation: bool=True):
    """Create multi condition matcher using 'OR'."""
    if config_validation:
        config = cv.OR_CONDITION_SCHEMA(config)
    checks = None

    def if_or_condition(hass: HomeAssistant,
                        variables=None) -> bool:
        """Test and condition."""
        nonlocal checks

        if checks is None:
            checks = [async_from_config(entry, False) for entry
                      in config['conditions']]

        try:
            for check in checks:
                if check(hass, variables):
                    return True
        except Exception as ex:  # pylint: disable=broad-except
            _LOGGER.warning('Error during or-condition: %s', ex)

        return False

    return if_or_condition


or_from_config = _threaded_factory(async_or_from_config)


# pylint: disable=too-many-arguments
def numeric_state(hass: HomeAssistant, entity, below=None, above=None,
                  value_template=None, variables=None):
    """Test a numeric state condition."""
    return run_callback_threadsafe(
        hass.loop, async_numeric_state, hass, entity, below, above,
        value_template, variables,
    ).result()


def async_numeric_state(hass: HomeAssistant, entity, below=None, above=None,
                        value_template=None, variables=None):
    """Test a numeric state condition."""
    if isinstance(entity, str):
        entity = hass.states.get(entity)

    if entity is None:
        return False

    if value_template is None:
        value = entity.state
    else:
        variables = dict(variables or {})
        variables['state'] = entity
        try:
            value = value_template.async_render(variables)
        except TemplateError as ex:
            _LOGGER.error("Template error: %s", ex)
            return False

    try:
        value = float(value)
    except ValueError:
        _LOGGER.warning("Value cannot be processed as a number: %s", value)
        return False

    if below is not None and value > below:
        return False

    if above is not None and value < above:
        return False

    return True


def async_numeric_state_from_config(config, config_validation=True):
    """Wrap action method with state based condition."""
    if config_validation:
        config = cv.NUMERIC_STATE_CONDITION_SCHEMA(config)
    entity_id = config.get(CONF_ENTITY_ID)
    below = config.get(CONF_BELOW)
    above = config.get(CONF_ABOVE)
    value_template = config.get(CONF_VALUE_TEMPLATE)

    def if_numeric_state(hass, variables=None):
        """Test numeric state condition."""
        if value_template is not None:
            value_template.hass = hass

        return async_numeric_state(
            hass, entity_id, below, above, value_template, variables)

    return if_numeric_state


numeric_state_from_config = _threaded_factory(async_numeric_state_from_config)


def state(hass, entity, req_state, for_period=None):
    """Test if state matches requirements."""
    if isinstance(entity, str):
        entity = hass.states.get(entity)

    if entity is None:
        return False

    is_state = entity.state == req_state

    if for_period is None or not is_state:
        return is_state

    return dt_util.utcnow() - for_period > entity.last_changed


def state_from_config(config, config_validation=True):
    """Wrap action method with state based condition."""
    if config_validation:
        config = cv.STATE_CONDITION_SCHEMA(config)
    entity_id = config.get(CONF_ENTITY_ID)
    req_state = config.get(CONF_STATE)
    for_period = config.get('for')

    def if_state(hass, variables=None):
        """Test if condition."""
        return state(hass, entity_id, req_state, for_period)

    return if_state


def sun(hass, before=None, after=None, before_offset=None, after_offset=None):
    """Test if current time matches sun requirements."""
    now = dt_util.now().time()
    before_offset = before_offset or timedelta(0)
    after_offset = after_offset or timedelta(0)

    if before == SUN_EVENT_SUNRISE and now > (sun_cmp.next_rising(hass) +
                                              before_offset).time():
        return False

    elif before == SUN_EVENT_SUNSET and now > (sun_cmp.next_setting(hass) +
                                               before_offset).time():
        return False

    if after == SUN_EVENT_SUNRISE and now < (sun_cmp.next_rising(hass) +
                                             after_offset).time():
        return False

    elif after == SUN_EVENT_SUNSET and now < (sun_cmp.next_setting(hass) +
                                              after_offset).time():
        return False

    return True


def sun_from_config(config, config_validation=True):
    """Wrap action method with sun based condition."""
    if config_validation:
        config = cv.SUN_CONDITION_SCHEMA(config)
    before = config.get('before')
    after = config.get('after')
    before_offset = config.get('before_offset')
    after_offset = config.get('after_offset')

    def time_if(hass, variables=None):
        """Validate time based if-condition."""
        return sun(hass, before, after, before_offset, after_offset)

    return time_if


def template(hass, value_template, variables=None):
    """Test if template condition matches."""
    return run_callback_threadsafe(
        hass.loop, async_template, hass, value_template, variables,
    ).result()


def async_template(hass, value_template, variables=None):
    """Test if template condition matches."""
    try:
        value = value_template.async_render(variables)
    except TemplateError as ex:
        _LOGGER.error('Error duriong template condition: %s', ex)
        return False

    return value.lower() == 'true'


def async_template_from_config(config, config_validation=True):
    """Wrap action method with state based condition."""
    if config_validation:
        config = cv.TEMPLATE_CONDITION_SCHEMA(config)
    value_template = config.get(CONF_VALUE_TEMPLATE)

    def template_if(hass, variables=None):
        """Validate template based if-condition."""
        value_template.hass = hass

        return async_template(hass, value_template, variables)

    return template_if


template_from_config = _threaded_factory(async_template_from_config)


def time(before=None, after=None, weekday=None):
    """Test if local time condition matches.

{'fraction_non_alphanumeric': 0.08781200809534517, 'fraction_numerical': 0.00022487069934787497, 'mean_word_length': 2.880671902268761, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 15, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 16, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3378797', 'n_tokens_mistral': 5117, 'n_tokens_neox': 4957, 'n_words': 1339}