esphome/esphomeyaml/helpers.py

from __future__ import print_function

import logging
import re
from collections import OrderedDict, deque

from esphomeyaml import core
from esphomeyaml.const import CONF_AVAILABILITY, CONF_COMMAND_TOPIC, CONF_DISCOVERY, \
    CONF_INVERTED, \
    CONF_MODE, CONF_NUMBER, CONF_PAYLOAD_AVAILABLE, CONF_PAYLOAD_NOT_AVAILABLE, CONF_PCF8574, \
    CONF_RETAIN, CONF_STATE_TOPIC, CONF_TOPIC
from esphomeyaml.core import ESPHomeYAMLError, HexInt, TimePeriodMicroseconds, \
    TimePeriodMilliseconds, TimePeriodSeconds

_LOGGER = logging.getLogger(__name__)


def ensure_unique_string(preferred_string, current_strings):
    test_string = preferred_string
    current_strings_set = set(current_strings)

    tries = 1

    while test_string in current_strings_set:
        tries += 1
        test_string = u"{}_{}".format(preferred_string, tries)

    return test_string


def indent_all_but_first_and_last(text, padding=u'  '):
    lines = text.splitlines(True)
    if len(lines) <= 2:
        return text
    return lines[0] + u''.join(padding + line for line in lines[1:-1]) + lines[-1]


def indent_list(text, padding=u'  '):
    return [padding + line for line in text.splitlines()]


def indent(text, padding=u'  '):
    return u'\n'.join(indent_list(text, padding))


class Expression(object):
    def __init__(self):
        self.requires = []
        self.required = False

    def __str__(self):
        raise NotImplementedError

    def require(self):
        self.required = True
        for require in self.requires:
            if require.required:
                continue
            require.require()


class RawExpression(Expression):
    def __init__(self, text):
        super(RawExpression, self).__init__()
        self.text = text

    def __str__(self):
        return self.text


# pylint: disable=redefined-builtin
class AssignmentExpression(Expression):
    def __init__(self, type, modifier, name, rhs, obj):
        super(AssignmentExpression, self).__init__()
        self.type = type
        self.modifier = modifier
        self.name = name
        self.rhs = safe_exp(rhs)
        self.requires.append(self.rhs)
        self.obj = obj

    def __str__(self):
        type_ = self.type
        if core.SIMPLIFY:
            type_ = u'auto'
        return u"{} {}{} = {}".format(type_, self.modifier, self.name, self.rhs)


class ExpressionList(Expression):
    def __init__(self, *args):
        super(ExpressionList, self).__init__()
        # Remove every None on end
        args = list(args)
        while args and args[-1] is None:
            args.pop()
        self.args = []
        for arg in args:
            exp = safe_exp(arg)
            self.requires.append(exp)
            self.args.append(exp)

    def __str__(self):
        text = u", ".join(unicode(x) for x in self.args)
        return indent_all_but_first_and_last(text)


class TemplateArguments(Expression):
    def __init__(self, *args):
        super(TemplateArguments, self).__init__()
        self.args = ExpressionList(*args)
        self.requires.append(self.args)

    def __str__(self):
        return u'<{}>'.format(self.args)


class CallExpression(Expression):
    def __init__(self, base, *args):
        super(CallExpression, self).__init__()
        self.base = base
        if args and isinstance(args[0], TemplateArguments):
            self.template_args = args[0]
            self.requires.append(self.template_args)
            args = args[1:]
        else:
            self.template_args = None
        self.args = ExpressionList(*args)
        self.requires.append(self.args)

    def __str__(self):
        if self.template_args is not None:
            return u'{}{}({})'.format(self.base, self.template_args, self.args)
        return u'{}({})'.format(self.base, self.args)


class StructInitializer(Expression):
    def __init__(self, base, *args):
        super(StructInitializer, self).__init__()
        self.base = base
        if not isinstance(args, OrderedDict):
            args = OrderedDict(args)
        self.args = OrderedDict()
        for key, value in args.iteritems():
            if value is None:
                continue
            exp = safe_exp(value)
            self.args[key] = exp
            self.requires.append(exp)

    def __str__(self):
        cpp = u'{}{{\n'.format(self.base)
        for key, value in self.args.iteritems():
            cpp += u'  .{} = {},\n'.format(key, value)
        cpp += u'}'
        return cpp


class ArrayInitializer(Expression):
    def __init__(self, *args, **kwargs):
        super(ArrayInitializer, self).__init__()
        self.multiline = kwargs.get('multiline', True)
        self.args = []
        for arg in args:
            if arg is None:
                continue
            exp = safe_exp(arg)
            self.args.append(exp)
            self.requires.append(exp)

    def __str__(self):
        if not self.args:
            return u'{}'
        if self.multiline:
            cpp = u'{\n'
            for arg in self.args:
                cpp += u'  {},\n'.format(arg)
            cpp += u'}'
        else:
            cpp = u'{' + u', '.join(str(arg) for arg in self.args) + u'}'
        return cpp


class Literal(Expression):
    def __str__(self):
        raise NotImplementedError


# From https://stackoverflow.com/a/14945195/8924614
def cpp_string_escape(string, encoding='utf-8'):
    if isinstance(string, unicode):
        string = string.encode(encoding)
    result = ''
    for character in string:
        if not (32 <= ord(character) < 127) or character in ('\\', '"'):
            result += '\\%03o' % ord(character)
        else:
            result += character
    return '"' + result + '"'


class StringLiteral(Literal):
    def __init__(self, string):
        super(StringLiteral, self).__init__()
        self.string = string

    def __str__(self):
        return u'{}'.format(cpp_string_escape(self.string))


class IntLiteral(Literal):
    def __init__(self, i):
        super(IntLiteral, self).__init__()
        self.i = i

    def __str__(self):
        return unicode(self.i)


class BoolLiteral(Literal):
    def __init__(self, binary):
        super(BoolLiteral, self).__init__()
        self.binary = binary

    def __str__(self):
        return u"true" if self.binary else u"false"


class HexIntLiteral(Literal):
    def __init__(self, i):
        super(HexIntLiteral, self).__init__()
        self.i = HexInt(i)

    def __str__(self):
        return str(self.i)


class FloatLiteral(Literal):
    def __init__(self, float_):
        super(FloatLiteral, self).__init__()
        self.float_ = float_

    def __str__(self):
        return u"{:f}f".format(self.float_)


def safe_exp(obj):
    if isinstance(obj, Expression):
        return obj
    elif isinstance(obj, bool):
        return BoolLiteral(obj)
    elif isinstance(obj, (str, unicode)):
        return StringLiteral(obj)
    elif isinstance(obj, HexInt):
        return HexIntLiteral(obj)
    elif isinstance(obj, (int, long)):
        return IntLiteral(obj)
    elif isinstance(obj, float):
        return FloatLiteral(obj)
    elif isinstance(obj, TimePeriodMicroseconds):
        return IntLiteral(int(obj.total_microseconds))
    elif isinstance(obj, TimePeriodMilliseconds):
        return IntLiteral(int(obj.total_milliseconds))
    elif isinstance(obj, TimePeriodSeconds):
        return IntLiteral(int(obj.total_seconds))
    raise ValueError(u"Object is not an expression", obj)


class Statement(object):
    def __init__(self):
        pass

    def __str__(self):
        raise NotImplementedError


class RawStatement(Statement):
    def __init__(self, text):
        super(RawStatement, self).__init__()
        self.text = text

    def __str__(self):
        return self.text


class ExpressionStatement(Statement):
    def __init__(self, expression):
        super(ExpressionStatement, self).__init__()
        self.expression = safe_exp(expression)

    def __str__(self):
        return u"{};".format(self.expression)


def statement(expression):
    if isinstance(expression, Statement):
        return expression
    return ExpressionStatement(expression)


# pylint: disable=redefined-builtin, invalid-name
def variable(type, id, rhs):
    rhs = safe_exp(rhs)
    obj = MockObj(id, u'.')
    assignment = AssignmentExpression(type, '', id, rhs, obj)
    add(assignment)
    _VARIABLES[id] = obj, type
    obj.requires.append(assignment)
    return obj


def Pvariable(type, id, rhs):
    rhs = safe_exp(rhs)
    obj = MockObj(id, u'->')
    assignment = AssignmentExpression(type, '*', id, rhs, obj)
    add(assignment)
    _VARIABLES[id] = obj, type
    obj.requires.append(assignment)
    return obj


_QUEUE = deque()
_VARIABLES = {}
_EXPRESSIONS = []


def get_variable(id, type=None):
    result = None
    while _QUEUE:
        if id is not None:
            if id in _VARIABLES:
                result = _VARIABLES[id][0]
                break
        elif type is not None:
            result = next((x[0] for x in _VARIABLES.itervalues() if x[1] == type), None)
            if result is not None:
                break
        func, config = _QUEUE.popleft()
        func(config)
    if id is None and type is None:
        return None
    if result is None:
        if id is not None:
            if id in _VARIABLES:
                result = _VARIABLES[id][0]
        elif type is not None:
            result = next((x[0] for x in _VARIABLES.itervalues() if x[1] == type), None)

        if result is None:
            raise ESPHomeYAMLError(u"Couldn't find ID '{}' with type {}".format(id, type))
    return result


def add_task(func, config):
    _QUEUE.append((func, config))


def add(expression, require=True):
    if require and isinstance(expression, Expression):
        expression.require()
    _EXPRESSIONS.append(expression)
    return expression


class MockObj(Expression):
    def __init__(self, base, op=u'.'):
        self.base = base
        self.op = op
        super(MockObj, self).__init__()

    def __getattr__(self, attr):
        next_op = u'.'
        if attr.startswith(u'P'):
            attr = attr[1:]
            next_op = u'->'
        op = self.op
        obj = MockObj(u'{}{}{}'.format(self.base, op, attr), next_op)
        obj.requires.append(self)
        return obj

    def __call__(self, *args, **kwargs):
        call = CallExpression(self.base, *args)
        obj = MockObj(call, self.op)
        obj.requires.append(self)
        obj.requires.append(call)
        return obj

    def __str__(self):
        return unicode(self.base)

    def require(self):
        self.required = True
        for require in self.requires:
            if require.required:
                continue
            require.require()


App = MockObj(u'App')

GPIOPin = MockObj(u'GPIOPin')
GPIOOutputPin = MockObj(u'GPIOOutputPin')
GPIOInputPin = MockObj(u'GPIOInputPin')


def get_gpio_pin_number(conf):
    if isinstance(conf, int):
        return conf
    return conf[CONF_NUMBER]


def generic_gpio_pin_expression_(conf, mock_obj, default_mode):
    if conf is None:
        return None
    number = conf[CONF_NUMBER]
    inverted = conf.get(CONF_INVERTED)
    if CONF_PCF8574 in conf:
        hub = get_variable(conf[CONF_PCF8574], 'io::PCF8574Component')
        if default_mode == u'INPUT':
            mode = conf.get(CONF_MODE, u'INPUT')
            return hub.make_input_pin(number,
                                      RawExpression('PCF8574_' + mode),
                                      inverted)
        elif default_mode == u'OUTPUT':
            return hub.make_output_pin(number, inverted)
        else:
            raise ESPHomeYAMLError(u"Unknown default mode {}".format(default_mode))
    if len(conf) == 1:
        return IntLiteral(number)
    mode = RawExpression(conf.get(CONF_MODE, default_mode))
    return mock_obj(number, mode, inverted)


def gpio_output_pin_expression(conf):
    return generic_gpio_pin_expression_(conf, GPIOOutputPin, 'OUTPUT')


def gpio_input_pin_expression(conf):
    return generic_gpio_pin_expression_(conf, GPIOInputPin, 'INPUT')


def setup_mqtt_component(obj, config):
    if CONF_RETAIN in config:
        add(obj.set_retain(config[CONF_RETAIN]))
    if not config.get(CONF_DISCOVERY, True):
        add(obj.disable_discovery())
    if CONF_STATE_TOPIC in config:
        add(obj.set_custom_state_topic(config[CONF_STATE_TOPIC]))
    if CONF_COMMAND_TOPIC in config:
        add(obj.set_custom_command_topic(config[CONF_COMMAND_TOPIC]))
    if CONF_AVAILABILITY in config:
        availability = config[CONF_AVAILABILITY]
        add(obj.set_availability(availability[CONF_TOPIC], availability[CONF_PAYLOAD_AVAILABLE],
                                 availability[CONF_PAYLOAD_NOT_AVAILABLE]))


def exp_empty_optional(type):
    return RawExpression(u'Optional<{}>()'.format(type))


def exp_optional(type, value):
    if value is None:
        return exp_empty_optional(type)
    return value


# shlex's quote for Python 2.7
_find_unsafe = re.compile(r'[^\w@%+=:,./-]').search


def quote(s):
    """Return a shell-escaped version of the string *s*."""
    if not s:
        return u"''"
    if _find_unsafe(s) is None:
        return s

    # use single quotes, and put single quotes into double quotes
    # the string $'b is then quoted as '$'"'"'b'
    return u"'" + s.replace(u"'", u"'\"'\"'") + u"'"


def color(the_color, message='', reset=None):
    """Color helper."""
    from colorlog.escape_codes import escape_codes, parse_colors
    if not message:
        return parse_colors(the_color)
    return parse_colors(the_color) + message + escape_codes[reset or 'reset']