Generator for ProtoBuf (proto3)

README.rst

@@ -138,8 +138,8 @@ Call the generator with the appropriate target:
 .. code-block::
 
     usage: aas-core-codegen [-h] --model_path MODEL_PATH --snippets_dir
-                            SNIPPETS_DIR --output_dir OUTPUT_DIR --target
-                            {csharp,cpp,golang,java,jsonschema,python,typescript,rdf_shacl,xsd,jsonld_context}
+                            SNIPPETS_DIR --output_dir OUTPUT_DIR --target                          
+                            {csharp,cpp,golang,java,jsonschema,python,typescript,rdf_shacl,xsd,jsonld_context,protobuf}
                             [--version]
 
     Generate implementations and schemas based on an AAS meta-model.
@@ -153,7 +153,7 @@ Call the generator with the appropriate target:
                             specific code snippets
       --output_dir OUTPUT_DIR
                             path to the generated code
-      --target {csharp,cpp,golang,java,jsonschema,python,typescript,rdf_shacl,xsd,jsonld_context}
+      --target {csharp,cpp,golang,java,jsonschema,python,typescript,rdf_shacl,xsd,jsonld_context,protobuf}
                             target language or schema
       --version             show the current version and exit
 

aas_core_codegen/cpp/aas_common/__init__.py

@@ -1,4 +1,5 @@
 """Generate C++ code for common functions."""
+
 from aas_core_codegen.cpp.aas_common import _generate
 
 generate_header = _generate.generate_header

aas_core_codegen/intermediate/_translate.py

@@ -1044,9 +1044,11 @@ def _to_arguments(parsed: Sequence[parse.Argument]) -> List[Argument]:
         Argument(
             name=parsed_arg.name,
             type_annotation=_to_type_annotation(parsed_arg.type_annotation),
-            default=_DefaultPlaceholder(parsed=parsed_arg.default)  # type: ignore
-            if parsed_arg.default is not None
-            else None,
+            default=(
+                _DefaultPlaceholder(parsed=parsed_arg.default)  # type: ignore
+                if parsed_arg.default is not None
+                else None
+            ),
             parsed=parsed_arg,
         )
         for parsed_arg in parsed
@@ -3446,9 +3448,11 @@ def _second_pass_to_stack_constructors_in_place(
                 if ancestor is None:
                     errors.append(
                         Error(
-                            cls.constructor.parsed.node
-                            if cls.constructor.parsed is not None
-                            else cls.parsed.node,
+                            (
+                                cls.constructor.parsed.node
+                                if cls.constructor.parsed is not None
+                                else cls.parsed.node
+                            ),
                             f"In the constructor of the class {cls.name!r} "
                             f"the super-constructor for "
                             f"the class {statement.super_name!r} is invoked, "
@@ -3461,9 +3465,11 @@ def _second_pass_to_stack_constructors_in_place(
                 if id(ancestor) not in cls.inheritance_id_set:
                     errors.append(
                         Error(
-                            cls.constructor.parsed.node
-                            if cls.constructor.parsed is not None
-                            else cls.parsed.node,
+                            (
+                                cls.constructor.parsed.node
+                                if cls.constructor.parsed is not None
+                                else cls.parsed.node
+                            ),
                             f"In the constructor of the class {cls.name!r} "
                             f"the super-constructor for "
                             f"the class {statement.super_name!r} is invoked, "

aas_core_codegen/java/xmlization/_generate.py

@@ -2031,7 +2031,7 @@ def _generate_serialize(
 public static void to(
 {I}IClass that,
 {I}XMLStreamWriter writer) throws SerializeException {{
-{I}VisitorWithWriter visitor = new VisitorWithWriter(); 
+{I}VisitorWithWriter visitor = new VisitorWithWriter();
 {I}visitor.visit(
 {II}that, writer);
 }}"""

aas_core_codegen/main.py

@@ -17,6 +17,7 @@
 import aas_core_codegen.typescript.main as typescript_main
 import aas_core_codegen.xsd.main as xsd_main
 import aas_core_codegen.jsonld.main as jsonld_main
+import aas_core_codegen.protobuf.main as protobuf_main
 from aas_core_codegen import run, specific_implementations
 from aas_core_codegen.common import LinenoColumner, assert_never
 
@@ -36,6 +37,7 @@ class Target(enum.Enum):
     RDF_SHACL = "rdf_shacl"
     XSD = "xsd"
     JSONLD_CONTEXT = "jsonld_context"
+    PROTOBUF = "protobuf"
 
 
 class Parameters:
@@ -164,6 +166,9 @@ def execute(params: Parameters, stdout: TextIO, stderr: TextIO) -> int:
     elif params.target is Target.JSONLD_CONTEXT:
         return jsonld_main.execute(context=run_context, stdout=stdout, stderr=stderr)
 
+    elif params.target is Target.PROTOBUF:
+        return protobuf_main.execute(run_context, stdout=stdout, stderr=stderr)
+
     else:
         assert_never(params.target)
 

aas_core_codegen/protobuf/__init__.py

@@ -0,0 +1 @@
+"""Generate ProtoBuf files based on the intermediate meta-model."""

aas_core_codegen/protobuf/common.py

@@ -0,0 +1,232 @@
+"""Provide common functions shared among different ProtoBuf code generation modules."""
+
+import re
+from typing import List, cast, Optional
+
+from icontract import ensure, require
+
+from aas_core_codegen import intermediate
+from aas_core_codegen.common import Stripped, assert_never
+from aas_core_codegen.protobuf import naming as proto_naming
+
+
+@ensure(lambda result: result.startswith('"'))
+@ensure(lambda result: result.endswith('"'))
+def string_literal(text: str) -> Stripped:
+    """Generate a ProtoBuf string literal from the ``text``."""
+    escaped = []  # type: List[str]
+
+    for character in text:
+        code_point = ord(character)
+
+        if character == "\a":
+            escaped.append("\\a")
+        elif character == "\b":
+            escaped.append("\\b")
+        elif character == "\f":
+            escaped.append("\\f")
+        elif character == "\n":
+            escaped.append("\\n")
+        elif character == "\r":
+            escaped.append("\\r")
+        elif character == "\t":
+            escaped.append("\\t")
+        elif character == "\v":
+            escaped.append("\\v")
+        elif character == '"':
+            escaped.append('\\"')
+        elif character == "\\":
+            escaped.append("\\\\")
+        elif code_point < 32:
+            # Non-printable ASCII characters
+            escaped.append(f"\\x{ord(character):x}")
+        elif 255 < code_point < 65536:
+            # Above ASCII
+            escaped.append(f"\\u{ord(character):04x}")
+        elif code_point >= 65536:
+            # Above Unicode Binary Multilingual Pane
+            escaped.append(f"\\U{ord(character):08x}")
+        else:
+            escaped.append(character)
+
+    return Stripped('"{}"'.format("".join(escaped)))
+
+
+def needs_escaping(text: str) -> bool:
+    """Check whether the ``text`` contains a character that needs escaping."""
+    for character in text:
+        if character == "\a":
+            return True
+        elif character == "\b":
+            return True
+        elif character == "\f":
+            return True
+        elif character == "\n":
+            return True
+        elif character == "\r":
+            return True
+        elif character == "\t":
+            return True
+        elif character == "\v":
+            return True
+        elif character == '"':
+            return True
+        elif character == "\\":
+            return True
+        else:
+            pass
+
+    return False
+
+
+PRIMITIVE_TYPE_MAP = {
+    intermediate.PrimitiveType.BOOL: Stripped("bool"),
+    intermediate.PrimitiveType.INT: Stripped("int64"),
+    intermediate.PrimitiveType.FLOAT: Stripped("double"),
+    intermediate.PrimitiveType.STR: Stripped("string"),
+    intermediate.PrimitiveType.BYTEARRAY: Stripped("bytes"),
+}
+
+
+def _assert_all_primitive_types_are_mapped() -> None:
+    """Assert that we have explicitly mapped all the primitive types to ProtoBuf."""
+    all_primitive_literals = set(literal.value for literal in PRIMITIVE_TYPE_MAP)
+
+    mapped_primitive_literals = set(
+        literal.value for literal in intermediate.PrimitiveType
+    )
+
+    all_diff = all_primitive_literals.difference(mapped_primitive_literals)
+    mapped_diff = mapped_primitive_literals.difference(all_primitive_literals)
+
+    messages = []  # type: List[str]
+    if len(mapped_diff) > 0:
+        messages.append(
+            f"More primitive maps are mapped than there were defined "
+            f"in the ``intermediate._types``: {sorted(mapped_diff)}"
+        )
+
+    if len(all_diff) > 0:
+        messages.append(
+            f"One or more primitive types in the ``intermediate._types`` were not "
+            f"mapped in PRIMITIVE_TYPE_MAP: {sorted(all_diff)}"
+        )
+
+    if len(messages) > 0:
+        raise AssertionError("\n\n".join(messages))
+
+
+_assert_all_primitive_types_are_mapped()
+
+
+# fmt: off
+@require(
+    lambda our_type_qualifier:
+    not (our_type_qualifier is not None)
+    or not our_type_qualifier.endswith('.')
+)
+# fmt: on
+def generate_type(
+    type_annotation: intermediate.TypeAnnotationUnion,
+    our_type_qualifier: Optional[Stripped] = None,
+) -> Stripped:
+    """
+    Generate the ProtoBuf type for the given type annotation.
+
+    ``our_type_prefix`` is appended to all our types, if specified.
+    """
+    our_type_prefix = "" if our_type_qualifier is None else f"{our_type_qualifier}."
+    if isinstance(type_annotation, intermediate.PrimitiveTypeAnnotation):
+        return PRIMITIVE_TYPE_MAP[type_annotation.a_type]
+
+    elif isinstance(type_annotation, intermediate.OurTypeAnnotation):
+        our_type = type_annotation.our_type
+
+        if isinstance(our_type, intermediate.Enumeration):
+            return Stripped(
+                our_type_prefix + proto_naming.enum_name(type_annotation.our_type.name)
+            )
+
+        elif isinstance(our_type, intermediate.ConstrainedPrimitive):
+            return PRIMITIVE_TYPE_MAP[our_type.constrainee]
+
+        elif isinstance(our_type, intermediate.Class):
+            return Stripped(our_type_prefix + proto_naming.class_name(our_type.name))
+
+    elif isinstance(type_annotation, intermediate.ListTypeAnnotation):
+        item_type = generate_type(
+            type_annotation=type_annotation.items, our_type_qualifier=our_type_qualifier
+        )
+
+        return Stripped(f"repeated {item_type}")
+
+    elif isinstance(type_annotation, intermediate.OptionalTypeAnnotation):
+        value = generate_type(
+            type_annotation=type_annotation.value, our_type_qualifier=our_type_qualifier
+        )
+
+        # careful: do not generate "optional" keyword for list-type elements since otherwise we get invalid
+        # constructs like "optional repeated <type> <name>"
+        if isinstance(type_annotation.value, intermediate.ListTypeAnnotation):
+            return Stripped(f"{value}")
+        else:
+            return Stripped(f"optional {value}")
+
+    else:
+        assert_never(type_annotation)
+
+    raise AssertionError("Should not have gotten here")
+
+
+INDENT = "  "
+INDENT2 = INDENT * 2
+INDENT3 = INDENT * 3
+INDENT4 = INDENT * 4
+INDENT5 = INDENT * 5
+INDENT6 = INDENT * 6
+
+# noinspection RegExpSimplifiable
+NAMESPACE_IDENTIFIER_RE = re.compile(
+    r"[a-zA-Z_][a-zA-Z_0-9]*(\.[a-zA-Z_][a-zA-Z_0-9]*)*"
+)
+
+
+class NamespaceIdentifier(str):
+    """Capture a namespace identifier."""
+
+    @require(lambda identifier: NAMESPACE_IDENTIFIER_RE.fullmatch(identifier))
+    def __new__(cls, identifier: str) -> "NamespaceIdentifier":
+        return cast(NamespaceIdentifier, identifier)
+
+
+WARNING = Stripped(
+    """\
+/*
+ * This code has been automatically generated by aas-core-codegen.
+ * Do NOT edit or append.
+ */"""
+)
+
+
+# fmt: off
+@ensure(
+    lambda namespace, result:
+    not (namespace != "Aas") or len(result) == 1,
+    "Exactly one block of stripped text to be appended to the list of using directives "
+    "if this using directive is necessary"
+)
+@ensure(
+    lambda namespace, result:
+    not (namespace == "Aas") or len(result) == 0,
+    "Empty list if no directive is necessary"
+)
+# fmt: on
+def generate_using_aas_directive_if_necessary(
+    namespace: NamespaceIdentifier,
+) -> List[Stripped]:
+    """
+    Generates the import directive for the AAS namespace.
+
+    This method is not to be used because proto3 does not need namespaces.
+    """
+    raise NotImplementedError("Not using the Aas namespace.")