|
| 1 | +#!/usr/bin/env python3 |
| 2 | +# |
| 3 | +# Copyright (c) 2026 TOKITA Hiroshi |
| 4 | +# |
| 5 | +# SPDX-License-Identifier: Apache-2.0 |
| 6 | + |
| 7 | +from __future__ import annotations |
| 8 | + |
| 9 | +from typing import Dict, List, NamedTuple, Tuple |
| 10 | + |
| 11 | +from google.protobuf.descriptor_pb2 import DescriptorProto, FieldDescriptorProto |
| 12 | + |
| 13 | +__all__ = [ |
| 14 | + "MethodSpec", |
| 15 | +] |
| 16 | + |
| 17 | +_DEFAULT_FIELD_CPP_TYPES: Dict[int, str] = { |
| 18 | + FieldDescriptorProto.TYPE_BOOL: "bool", |
| 19 | + FieldDescriptorProto.TYPE_INT32: "int32_t", |
| 20 | + FieldDescriptorProto.TYPE_INT64: "int64_t", |
| 21 | + FieldDescriptorProto.TYPE_UINT32: "uint32_t", |
| 22 | + FieldDescriptorProto.TYPE_UINT64: "uint64_t", |
| 23 | + FieldDescriptorProto.TYPE_SINT32: "int32_t", |
| 24 | + FieldDescriptorProto.TYPE_SINT64: "int64_t", |
| 25 | + FieldDescriptorProto.TYPE_FIXED32: "uint32_t", |
| 26 | + FieldDescriptorProto.TYPE_FIXED64: "uint64_t", |
| 27 | + FieldDescriptorProto.TYPE_SFIXED32: "int32_t", |
| 28 | + FieldDescriptorProto.TYPE_SFIXED64: "int64_t", |
| 29 | + FieldDescriptorProto.TYPE_FLOAT: "float", |
| 30 | + FieldDescriptorProto.TYPE_DOUBLE: "double", |
| 31 | + FieldDescriptorProto.TYPE_STRING: "const char *", |
| 32 | + FieldDescriptorProto.TYPE_BYTES: "const uint8_t *", |
| 33 | + FieldDescriptorProto.TYPE_ENUM: "int32_t", |
| 34 | +} |
| 35 | + |
| 36 | + |
| 37 | +def _opt_str(options, extension) -> str: |
| 38 | + try: |
| 39 | + return str(options.Extensions[extension]).strip() |
| 40 | + except (AttributeError, KeyError): |
| 41 | + return "" |
| 42 | + |
| 43 | + |
| 44 | +class MethodSpec(NamedTuple): |
| 45 | + """C++ method signature derived from one protobuf RPC method.""" |
| 46 | + |
| 47 | + decl: str |
| 48 | + overload_key: Tuple[str, Tuple[str, ...]] |
| 49 | + call_name: str |
| 50 | + arg_names: List[str] |
| 51 | + returns_void: bool |
| 52 | + visibility: str |
| 53 | + |
| 54 | + @classmethod |
| 55 | + def build( |
| 56 | + cls, |
| 57 | + method, |
| 58 | + opts_pb2, |
| 59 | + message_map: Dict[str, DescriptorProto], |
| 60 | + ) -> "MethodSpec": |
| 61 | + def _resolve_field(field: FieldDescriptorProto) -> Tuple[str, str]: |
| 62 | + field_opts = field.options |
| 63 | + explicit_cpp_type = _opt_str(field_opts, opts_pb2.cpp_type) |
| 64 | + if explicit_cpp_type: |
| 65 | + field_type = explicit_cpp_type |
| 66 | + else: |
| 67 | + if field.label == FieldDescriptorProto.LABEL_REPEATED: |
| 68 | + raise ValueError( |
| 69 | + f"{method.name}: field '{field.name}' is repeated; " |
| 70 | + "C++ header generation requires an explicit " |
| 71 | + "(arduino.cpp_type) override" |
| 72 | + ) |
| 73 | + if field.type not in _DEFAULT_FIELD_CPP_TYPES: |
| 74 | + raise ValueError( |
| 75 | + f"{method.name}: field '{field.name}' has unsupported " |
| 76 | + f"protobuf type '{field.type}'; C++ header generation " |
| 77 | + "requires an explicit (arduino.cpp_type) override" |
| 78 | + ) |
| 79 | + field_type = _DEFAULT_FIELD_CPP_TYPES[field.type] |
| 80 | + field_name = _opt_str(field_opts, opts_pb2.field_cpp_name) or field.name |
| 81 | + return field_type, field_name |
| 82 | + |
| 83 | + input_msg = message_map.get(method.input_type) |
| 84 | + output_msg = message_map.get(method.output_type) |
| 85 | + |
| 86 | + if input_msg is None: |
| 87 | + raise ValueError( |
| 88 | + f"{method.name}: unknown input type '{method.input_type}'; " |
| 89 | + "use google.protobuf.Empty for methods without parameters" |
| 90 | + ) |
| 91 | + if output_msg is None: |
| 92 | + raise ValueError( |
| 93 | + f"{method.name}: unknown output type '{method.output_type}'; " |
| 94 | + "use google.protobuf.Empty for void methods" |
| 95 | + ) |
| 96 | + if len(output_msg.field) > 1: |
| 97 | + raise ValueError( |
| 98 | + f"{method.name}: output type '{method.output_type}' has " |
| 99 | + f"{len(output_msg.field)} fields; C++ header generation supports " |
| 100 | + "void outputs or a single return-value field" |
| 101 | + ) |
| 102 | + |
| 103 | + visibility = _opt_str(method.options, opts_pb2.method_visibility).lower() |
| 104 | + if visibility not in {"public", "protected", "private", ""}: |
| 105 | + raise ValueError( |
| 106 | + f"{method.name}: unsupported method_visibility '{visibility}' " |
| 107 | + "(expected: public, protected, private)" |
| 108 | + ) |
| 109 | + if not visibility: |
| 110 | + visibility = "public" |
| 111 | + |
| 112 | + method_name = _opt_str(method.options, opts_pb2.cpp_name) or method.name |
| 113 | + return_type = _opt_str(method.options, opts_pb2.cpp_return) |
| 114 | + if not return_type: |
| 115 | + return_type = ( |
| 116 | + "void" |
| 117 | + if len(output_msg.field) == 0 |
| 118 | + else _resolve_field(output_msg.field[0])[0] |
| 119 | + ) |
| 120 | + |
| 121 | + params = [_resolve_field(f) for f in input_msg.field] |
| 122 | + arg_types = tuple(ftype for ftype, _ in params) |
| 123 | + arg_names = [name for _, name in params] |
| 124 | + params_blob = ", ".join(f"{ftype} {fname}" for ftype, fname in params) |
| 125 | + |
| 126 | + decl = ( |
| 127 | + f"{method_name}({params_blob})" |
| 128 | + if method_name.startswith("operator ") |
| 129 | + else f"{return_type} {method_name}({params_blob})" |
| 130 | + ) |
| 131 | + |
| 132 | + return cls( |
| 133 | + decl=decl, |
| 134 | + overload_key=(method_name, arg_types), |
| 135 | + call_name=method_name, |
| 136 | + arg_names=arg_names, |
| 137 | + returns_void=(return_type == "void"), |
| 138 | + visibility=visibility, |
| 139 | + ) |
0 commit comments