esphome/script/api_protobuf/api_protobuf.py

"""Python 3 script to automatically generate C++ classes for ESPHome's native API.

It's pretty crappy spaghetti code, but it works.

you need to install protobuf-compiler:
running protc --version should return
libprotoc 3.6.1

then run this script with python3 and the files

    esphome/components/api/api_pb2_service.h
    esphome/components/api/api_pb2_service.cpp
    esphome/components/api/api_pb2.h
    esphome/components/api/api_pb2.cpp

will be generated, they still need to be formatted
"""

import re
from pathlib import Path
from textwrap import dedent
from subprocess import call

# Generate with
# protoc --python_out=script/api_protobuf -I esphome/components/api/ api_options.proto

import api_options_pb2 as pb
import google.protobuf.descriptor_pb2 as descriptor

file_header = "// This file was automatically generated with a tool.\n"
file_header += "// See scripts/api_protobuf/api_protobuf.py\n"

cwd = Path(__file__).resolve().parent
root = cwd.parent.parent / "esphome" / "components" / "api"
prot = root / "api.protoc"
call(["protoc", "-o", str(prot), "-I", str(root), "api.proto"])
content = prot.read_bytes()

d = descriptor.FileDescriptorSet.FromString(content)


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


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


def camel_to_snake(name):
    # https://stackoverflow.com/a/1176023
    s1 = re.sub("(.)([A-Z][a-z]+)", r"\1_\2", name)
    return re.sub("([a-z0-9])([A-Z])", r"\1_\2", s1).lower()


class TypeInfo:
    def __init__(self, field):
        self._field = field

    @property
    def default_value(self):
        return ""

    @property
    def name(self):
        return self._field.name

    @property
    def arg_name(self):
        return self.name

    @property
    def field_name(self):
        return self.name

    @property
    def number(self):
        return self._field.number

    @property
    def repeated(self):
        return self._field.label == 3

    @property
    def cpp_type(self):
        raise NotImplementedError

    @property
    def reference_type(self):
        return f"{self.cpp_type} "

    @property
    def const_reference_type(self):
        return f"{self.cpp_type} "

    @property
    def public_content(self) -> str:
        return [self.class_member]

    @property
    def protected_content(self) -> str:
        return []

    @property
    def class_member(self) -> str:
        return f"{self.cpp_type} {self.field_name}{{{self.default_value}}};  // NOLINT"

    @property
    def decode_varint_content(self) -> str:
        content = self.decode_varint
        if content is None:
            return None
        return dedent(
            f"""\
        case {self.number}: {{
          this->{self.field_name} = {content};
          return true;
        }}"""
        )

    decode_varint = None

    @property
    def decode_length_content(self) -> str:
        content = self.decode_length
        if content is None:
            return None
        return dedent(
            f"""\
        case {self.number}: {{
          this->{self.field_name} = {content};
          return true;
        }}"""
        )

    decode_length = None

    @property
    def decode_32bit_content(self) -> str:
        content = self.decode_32bit
        if content is None:
            return None
        return dedent(
            f"""\
        case {self.number}: {{
          this->{self.field_name} = {content};
          return true;
        }}"""
        )

    decode_32bit = None

    @property
    def decode_64bit_content(self) -> str:
        content = self.decode_64bit
        if content is None:
            return None
        return dedent(
            f"""\
        case {self.number}: {{
          this->{self.field_name} = {content};
          return true;
        }}"""
        )

    decode_64bit = None

    @property
    def encode_content(self):
        return f"buffer.{self.encode_func}({self.number}, this->{self.field_name});"

    encode_func = None

    @property
    def dump_content(self):
        o = f'out.append("  {self.name}: ");\n'
        o += self.dump(f"this->{self.field_name}") + "\n"
        o += f'out.append("\\n");\n'
        return o

    dump = None


TYPE_INFO = {}


def register_type(name):
    def func(value):
        TYPE_INFO[name] = value
        return value

    return func


@register_type(1)
class DoubleType(TypeInfo):
    cpp_type = "double"
    default_value = "0.0"
    decode_64bit = "value.as_double()"
    encode_func = "encode_double"

    def dump(self, name):
        o = f'sprintf(buffer, "%g", {name});\n'
        o += f"out.append(buffer);"
        return o


@register_type(2)
class FloatType(TypeInfo):
    cpp_type = "float"
    default_value = "0.0f"
    decode_32bit = "value.as_float()"
    encode_func = "encode_float"

    def dump(self, name):
        o = f'sprintf(buffer, "%g", {name});\n'
        o += f"out.append(buffer);"
        return o


@register_type(3)
class Int64Type(TypeInfo):
    cpp_type = "int64_t"
    default_value = "0"
    decode_varint = "value.as_int64()"
    encode_func = "encode_int64"

    def dump(self, name):
        o = f'sprintf(buffer, "%ll", {name});\n'
        o += f"out.append(buffer);"
        return o


@register_type(4)
class UInt64Type(TypeInfo):
    cpp_type = "uint64_t"
    default_value = "0"
    decode_varint = "value.as_uint64()"
    encode_func = "encode_uint64"

    def dump(self, name):
        o = f'sprintf(buffer, "%ull", {name});\n'
        o += f"out.append(buffer);"
        return o


@register_type(5)
class Int32Type(TypeInfo):
    cpp_type = "int32_t"
    default_value = "0"
    decode_varint = "value.as_int32()"
    encode_func = "encode_int32"

    def dump(self, name):
        o = f'sprintf(buffer, "%d", {name});\n'
        o += f"out.append(buffer);"
        return o


@register_type(6)
class Fixed64Type(TypeInfo):
    cpp_type = "uint64_t"
    default_value = "0"
    decode_64bit = "value.as_fixed64()"
    encode_func = "encode_fixed64"

    def dump(self, name):
        o = f'sprintf(buffer, "%ull", {name});\n'
        o += f"out.append(buffer);"
        return o


@register_type(7)
class Fixed32Type(TypeInfo):
    cpp_type = "uint32_t"
    default_value = "0"
    decode_32bit = "value.as_fixed32()"
    encode_func = "encode_fixed32"

    def dump(self, name):
        o = f'sprintf(buffer, "%u", {name});\n'
        o += f"out.append(buffer);"
        return o


@register_type(8)
class BoolType(TypeInfo):
    cpp_type = "bool"
    default_value = "false"
    decode_varint = "value.as_bool()"
    encode_func = "encode_bool"

    def dump(self, name):
        o = f"out.append(YESNO({name}));"
        return o


@register_type(9)
class StringType(TypeInfo):
    cpp_type = "std::string"
    default_value = ""
    reference_type = "std::string &"
    const_reference_type = "const std::string &"
    decode_length = "value.as_string()"
    encode_func = "encode_string"

    def dump(self, name):
        o = f'out.append("\'").append({name}).append("\'");'
        return o


@register_type(11)
class MessageType(TypeInfo):
    @property
    def cpp_type(self):
        return self._field.type_name[1:]

    default_value = ""

    @property
    def reference_type(self):
        return f"{self.cpp_type} &"

    @property
    def const_reference_type(self):
        return f"const {self.cpp_type} &"

    @property
    def encode_func(self):
        return f"encode_message<{self.cpp_type}>"

    @property
    def decode_length(self):
        return f"value.as_message<{self.cpp_type}>()"

    def dump(self, name):
        o = f"{name}.dump_to(out);"
        return o


@register_type(12)
class BytesType(TypeInfo):
    cpp_type = "std::string"
    default_value = ""
    reference_type = "std::string &"
    const_reference_type = "const std::string &"
    decode_length = "value.as_string()"
    encode_func = "encode_string"

    def dump(self, name):
        o = f'out.append("\'").append({name}).append("\'");'
        return o


@register_type(13)
class UInt32Type(TypeInfo):
    cpp_type = "uint32_t"
    default_value = "0"
    decode_varint = "value.as_uint32()"
    encode_func = "encode_uint32"

    def dump(self, name):
        o = f'sprintf(buffer, "%u", {name});\n'
        o += f"out.append(buffer);"
        return o


@register_type(14)
class EnumType(TypeInfo):
    @property
    def cpp_type(self):
        return f"enums::{self._field.type_name[1:]}"

    @property
    def decode_varint(self):
        return f"value.as_enum<{self.cpp_type}>()"

    default_value = ""

    @property
    def encode_func(self):
        return f"encode_enum<{self.cpp_type}>"

    def dump(self, name):
        o = f"out.append(proto_enum_to_string<{self.cpp_type}>({name}));"
        return o


@register_type(15)
class SFixed32Type(TypeInfo):
    cpp_type = "int32_t"
    default_value = "0"
    decode_32bit = "value.as_sfixed32()"
    encode_func = "encode_sfixed32"

    def dump(self, name):
        o = f'sprintf(buffer, "%d", {name});\n'
        o += f"out.append(buffer);"
        return o


@register_type(16)
class SFixed64Type(TypeInfo):
    cpp_type = "int64_t"
    default_value = "0"
    decode_64bit = "value.as_sfixed64()"
    encode_func = "encode_sfixed64"

    def dump(self, name):
        o = f'sprintf(buffer, "%ll", {name});\n'
        o += f"out.append(buffer);"
        return o


@register_type(17)
class SInt32Type(TypeInfo):
    cpp_type = "int32_t"
    default_value = "0"
    decode_varint = "value.as_sint32()"
    encode_func = "encode_sint32"

    def dump(self, name):
        o = f'sprintf(buffer, "%d", {name});\n'
        o += f"out.append(buffer);"
        return o


@register_type(18)
class SInt64Type(TypeInfo):
    cpp_type = "int64_t"
    default_value = "0"
    decode_varint = "value.as_sint64()"
    encode_func = "encode_sin64"

    def dump(self):
        o = f'sprintf(buffer, "%ll", {name});\n'
        o += f"out.append(buffer);"
        return o


class RepeatedTypeInfo(TypeInfo):
    def __init__(self, field):
        super().__init__(field)
        self._ti = TYPE_INFO[field.type](field)

    @property
    def cpp_type(self):
        return f"std::vector<{self._ti.cpp_type}>"

    @property
    def reference_type(self):
        return f"{self.cpp_type} &"

    @property
    def const_reference_type(self):
        return f"const {self.cpp_type} &"

    @property
    def decode_varint_content(self) -> str:
        content = self._ti.decode_varint
        if content is None:
            return None
        return dedent(
            f"""\
        case {self.number}: {{
          this->{self.field_name}.push_back({content});
          return true;
        }}"""
        )

    @property
    def decode_length_content(self) -> str:
        content = self._ti.decode_length
        if content is None:
            return None
        return dedent(
            f"""\
        case {self.number}: {{
          this->{self.field_name}.push_back({content});
          return true;
        }}"""
        )

    @property
    def decode_32bit_content(self) -> str:
        content = self._ti.decode_32bit
        if content is None:
            return None
        return dedent(
            f"""\
        case {self.number}: {{
          this->{self.field_name}.push_back({content});
          return true;
        }}"""
        )

    @property
    def decode_64bit_content(self) -> str:
        content = self._ti.decode_64bit
        if content is None:
            return None
        return dedent(
            f"""\
        case {self.number}: {{
          this->{self.field_name}.push_back({content});
          return true;
        }}"""
        )

    @property
    def _ti_is_bool(self):
        # std::vector is specialized for bool, reference does not work
        return isinstance(self._ti, BoolType)

    @property
    def encode_content(self):
        return f"""\
        for (auto {'' if self._ti_is_bool else '&'}it : this->{self.field_name}) {{
          buffer.{self._ti.encode_func}({self.number}, it, true);
        }}"""

    @property
    def dump_content(self):
        o = f'for (const auto {"" if self._ti_is_bool else "&"}it : this->{self.field_name}) {{\n'
        o += f'  out.append("  {self.name}: ");\n'
        o += indent(self._ti.dump("it")) + "\n"
        o += f'  out.append("\\n");\n'
        o += f"}}\n"
        return o


def build_enum_type(desc):
    name = desc.name
    out = f"enum {name} : uint32_t {{\n"
    for v in desc.value:
        out += f"  {v.name} = {v.number},\n"
    out += "};\n"

    cpp = f"template<>\n"
    cpp += f"const char *proto_enum_to_string<enums::{name}>(enums::{name} value) {{\n"
    cpp += f"  switch (value) {{\n"
    for v in desc.value:
        cpp += f'    case enums::{v.name}: return "{v.name}";\n'
    cpp += f'    default: return "UNKNOWN";\n'
    cpp += f"  }}\n"
    cpp += f"}}\n"

    return out, cpp


def build_message_type(desc):
    public_content = []
    protected_content = []
    decode_varint = []
    decode_length = []
    decode_32bit = []
    decode_64bit = []
    encode = []
    dump = []

    for field in desc.field:
        if field.label == 3:
            ti = RepeatedTypeInfo(field)
        else:
            ti = TYPE_INFO[field.type](field)
        protected_content.extend(ti.protected_content)
        public_content.extend(ti.public_content)
        encode.append(ti.encode_content)

        if ti.decode_varint_content:
            decode_varint.append(ti.decode_varint_content)
        if ti.decode_length_content:
            decode_length.append(ti.decode_length_content)
        if ti.decode_32bit_content:
            decode_32bit.append(ti.decode_32bit_content)
        if ti.decode_64bit_content:
            decode_64bit.append(ti.decode_64bit_content)
        if ti.dump_content:
            dump.append(ti.dump_content)

    cpp = ""
    if decode_varint:
        decode_varint.append("default:\n  return false;")
        o = f"bool {desc.name}::decode_varint(uint32_t field_id, ProtoVarInt value) {{\n"
        o += "  switch (field_id) {\n"
        o += indent("\n".join(decode_varint), "    ") + "\n"
        o += "  }\n"
        o += "}\n"
        cpp += o
        prot = "bool decode_varint(uint32_t field_id, ProtoVarInt value) override;"
        protected_content.insert(0, prot)
    if decode_length:
        decode_length.append("default:\n  return false;")
        o = f"bool {desc.name}::decode_length(uint32_t field_id, ProtoLengthDelimited value) {{\n"
        o += "  switch (field_id) {\n"
        o += indent("\n".join(decode_length), "    ") + "\n"
        o += "  }\n"
        o += "}\n"
        cpp += o
        prot = "bool decode_length(uint32_t field_id, ProtoLengthDelimited value) override;"
        protected_content.insert(0, prot)
    if decode_32bit:
        decode_32bit.append("default:\n  return false;")
        o = f"bool {desc.name}::decode_32bit(uint32_t field_id, Proto32Bit value) {{\n"
        o += "  switch (field_id) {\n"
        o += indent("\n".join(decode_32bit), "    ") + "\n"
        o += "  }\n"
        o += "}\n"
        cpp += o
        prot = "bool decode_32bit(uint32_t field_id, Proto32Bit value) override;"
        protected_content.insert(0, prot)
    if decode_64bit:
        decode_64bit.append("default:\n  return false;")
        o = f"bool {desc.name}::decode_64bit(uint32_t field_id, Proto64bit value) {{\n"
        o += "  switch (field_id) {\n"
        o += indent("\n".join(decode_64bit), "    ") + "\n"
        o += "  }\n"
        o += "}\n"
        cpp += o
        prot = "bool decode_64bit(uint32_t field_id, Proto64bit value) override;"
        protected_content.insert(0, prot)

    o = f"void {desc.name}::encode(ProtoWriteBuffer buffer) const {{\n"
    o += indent("\n".join(encode)) + "\n"
    o += "}\n"
    cpp += o
    prot = "void encode(ProtoWriteBuffer buffer) const override;"
    public_content.append(prot)

    o = f"void {desc.name}::dump_to(std::string &out) const {{\n"
    if dump:
        o += f"  char buffer[64];\n"
        o += f'  out.append("{desc.name} {{\\n");\n'
        o += indent("\n".join(dump)) + "\n"
        o += f'  out.append("}}");\n'
    else:
        o += f'  out.append("{desc.name} {{}}");\n'
    o += "}\n"
    cpp += o
    prot = "void dump_to(std::string &out) const override;"
    public_content.append(prot)

    out = f"class {desc.name} : public ProtoMessage {{\n"
    out += " public:\n"
    out += indent("\n".join(public_content)) + "\n"
    out += " protected:\n"
    out += indent("\n".join(protected_content)) + "\n"
    out += "};\n"
    return out, cpp


file = d.file[0]
content = file_header
content += """\
#pragma once

#include "proto.h"

namespace esphome {
namespace api {

"""

cpp = file_header
cpp += """\
#include "api_pb2.h"
#include "esphome/core/log.h"

namespace esphome {
namespace api {

"""

content += "namespace enums {\n\n"

for enum in file.enum_type:
    s, c = build_enum_type(enum)
    content += s
    cpp += c

content += "\n}  // namespace enums\n\n"

mt = file.message_type

for m in mt:
    s, c = build_message_type(m)
    content += s
    cpp += c

content += """\

}  // namespace api
}  // namespace esphome
"""
cpp += """\

}  // namespace api
}  // namespace esphome
"""

with open(root / "api_pb2.h", "w") as f:
    f.write(content)

with open(root / "api_pb2.cpp", "w") as f:
    f.write(cpp)

SOURCE_BOTH = 0
SOURCE_SERVER = 1
SOURCE_CLIENT = 2

RECEIVE_CASES = {}

class_name = "APIServerConnectionBase"

ifdefs = {}


def get_opt(desc, opt, default=None):
    if not desc.options.HasExtension(opt):
        return default
    return desc.options.Extensions[opt]


def build_service_message_type(mt):
    snake = camel_to_snake(mt.name)
    id_ = get_opt(mt, pb.id)
    if id_ is None:
        return None

    source = get_opt(mt, pb.source, 0)

    ifdef = get_opt(mt, pb.ifdef)
    log = get_opt(mt, pb.log, True)
    nodelay = get_opt(mt, pb.no_delay, False)
    hout = ""
    cout = ""

    if ifdef is not None:
        ifdefs[str(mt.name)] = ifdef
        hout += f"#ifdef {ifdef}\n"
        cout += f"#ifdef {ifdef}\n"

    if source in (SOURCE_BOTH, SOURCE_SERVER):
        # Generate send
        func = f"send_{snake}"
        hout += f"bool {func}(const {mt.name} &msg);\n"
        cout += f"bool {class_name}::{func}(const {mt.name} &msg) {{\n"
        if log:
            cout += f'  ESP_LOGVV(TAG, "{func}: %s", msg.dump().c_str());\n'
        # cout += f'  this->set_nodelay({str(nodelay).lower()});\n'
        cout += f"  return this->send_message_<{mt.name}>(msg, {id_});\n"
        cout += f"}}\n"
    if source in (SOURCE_BOTH, SOURCE_CLIENT):
        # Generate receive
        func = f"on_{snake}"
        hout += f"virtual void {func}(const {mt.name} &value){{}};\n"
        case = ""
        if ifdef is not None:
            case += f"#ifdef {ifdef}\n"
        case += f"{mt.name} msg;\n"
        case += f"msg.decode(msg_data, msg_size);\n"
        if log:
            case += f'ESP_LOGVV(TAG, "{func}: %s", msg.dump().c_str());\n'
        case += f"this->{func}(msg);\n"
        if ifdef is not None:
            case += f"#endif\n"
        case += "break;"
        RECEIVE_CASES[id_] = case

    if ifdef is not None:
        hout += f"#endif\n"
        cout += f"#endif\n"

    return hout, cout


hpp = file_header
hpp += """\
#pragma once

#include "api_pb2.h"
#include "esphome/core/defines.h"

namespace esphome {
namespace api {

"""

cpp = file_header
cpp += """\
#include "api_pb2_service.h"
#include "esphome/core/log.h"

namespace esphome {
namespace api {

static const char *TAG = "api.service";

"""

hpp += f"class {class_name} : public ProtoService {{\n"
hpp += " public:\n"

for mt in file.message_type:
    obj = build_service_message_type(mt)
    if obj is None:
        continue
    hout, cout = obj
    hpp += indent(hout) + "\n"
    cpp += cout

cases = list(RECEIVE_CASES.items())
cases.sort()
hpp += " protected:\n"
hpp += f"  bool read_message(uint32_t msg_size, uint32_t msg_type, uint8_t *msg_data) override;\n"
out = f"bool {class_name}::read_message(uint32_t msg_size, uint32_t msg_type, uint8_t *msg_data) {{\n"
out += f"  switch(msg_type) {{\n"
for i, case in cases:
    c = f"case {i}: {{\n"
    c += indent(case) + "\n"
    c += f"}}"
    out += indent(c, "    ") + "\n"
out += "    default: \n"
out += "      return false;\n"
out += "  }\n"
out += "  return true;\n"
out += "}\n"
cpp += out
hpp += "};\n"

serv = file.service[0]
class_name = "APIServerConnection"
hpp += "\n"
hpp += f"class {class_name} : public {class_name}Base {{\n"
hpp += " public:\n"
hpp_protected = ""
cpp += "\n"

m = serv.method[0]
for m in serv.method:
    func = m.name
    inp = m.input_type[1:]
    ret = m.output_type[1:]
    is_void = ret == "void"
    snake = camel_to_snake(inp)
    on_func = f"on_{snake}"
    needs_conn = get_opt(m, pb.needs_setup_connection, True)
    needs_auth = get_opt(m, pb.needs_authentication, True)

    ifdef = ifdefs.get(inp, None)

    if ifdef is not None:
        hpp += f"#ifdef {ifdef}\n"
        hpp_protected += f"#ifdef {ifdef}\n"
        cpp += f"#ifdef {ifdef}\n"

    hpp_protected += f"  void {on_func}(const {inp} &msg) override;\n"
    hpp += f"  virtual {ret} {func}(const {inp} &msg) = 0;\n"
    cpp += f"void {class_name}::{on_func}(const {inp} &msg) {{\n"
    body = ""
    if needs_conn:
        body += "if (!this->is_connection_setup()) {\n"
        body += "  this->on_no_setup_connection();\n"
        body += "  return;\n"
        body += "}\n"
    if needs_auth:
        body += "if (!this->is_authenticated()) {\n"
        body += "  this->on_unauthenticated_access();\n"
        body += "  return;\n"
        body += "}\n"

    if is_void:
        body += f"this->{func}(msg);\n"
    else:
        body += f"{ret} ret = this->{func}(msg);\n"
        ret_snake = camel_to_snake(ret)
        body += f"if (!this->send_{ret_snake}(ret)) {{\n"
        body += f"  this->on_fatal_error();\n"
        body += "}\n"
    cpp += indent(body) + "\n" + "}\n"

    if ifdef is not None:
        hpp += f"#endif\n"
        hpp_protected += f"#endif\n"
        cpp += f"#endif\n"

hpp += " protected:\n"
hpp += hpp_protected
hpp += "};\n"

hpp += """\

}  // namespace api
}  // namespace esphome
"""
cpp += """\

}  // namespace api
}  // namespace esphome
"""

with open(root / "api_pb2_service.h", "w") as f:
    f.write(hpp)

with open(root / "api_pb2_service.cpp", "w") as f:
    f.write(cpp)

prot.unlink()