Generate OPC UA node set

We generate an OPC UA node set so that the model instances can be
transferred through OPC UA as channel. The mapping to OPC UA information
model is kept minimal and as simple as possible to avoid maintenance
issues in the future.

Author: mristin

README.rst

@@ -139,7 +139,7 @@ Call the generator with the appropriate target:
 
     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,protobuf,python_protobuf}
+                            {csharp,cpp,golang,java,jsonschema,python,typescript,rdf_shacl,xsd,jsonld_context,protobuf,python_protobuf,opcua}
                             [--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,protobuf,python_protobuf}
+      --target {csharp,cpp,golang,java,jsonschema,python,typescript,rdf_shacl,xsd,jsonld_context,protobuf,python_protobuf,opcua}
                             target language or schema
       --version             show the current version and exit
 

aas_core_codegen/intermediate/_translate.py

@@ -4437,6 +4437,54 @@ def _verify_only_simple_type_patterns(symbol_table: SymbolTable) -> List[Error]:
     return errors
 
 
+def _verify_invariant_descriptions_unique(symbol_table: SymbolTable) -> List[Error]:
+    """Check that no two invariants share the same description."""
+    errors = []  # type: List[Error]
+    for something_with_invariants in itertools.chain(
+        symbol_table.classes, symbol_table.constrained_primitives
+    ):
+        description_map = dict()  # type: MutableMapping[str, Invariant]
+
+        for invariant in something_with_invariants.invariants:
+            conflicting_invariant = description_map.get(invariant.description, None)
+            if conflicting_invariant is not None:
+                what: str
+                if isinstance(something_with_invariants, Class):
+                    what = "class"
+                elif isinstance(something_with_invariants, ConstrainedPrimitive):
+                    what = "constrained primitive"
+                else:
+                    assert_never(something_with_invariants)
+                    raise AssertionError("Unexpected execution path")
+
+                if conflicting_invariant.specified_for is invariant.specified_for:
+                    errors.append(
+                        Error(
+                            invariant.parsed.node,
+                            f"The invariants' descriptions need to be unique, "
+                            f"but they conflict "
+                            f"in the {what} {invariant.specified_for.name!r} "
+                            f"for the description: {invariant.description!r}",
+                        )
+                    )
+                else:
+                    errors.append(
+                        Error(
+                            invariant.parsed.node,
+                            f"The invariants' descriptions need to be unique, "
+                            f"but an invariant from "
+                            f"the {what} {invariant.specified_for.name!r} "
+                            f"and from "
+                            f"the {what} {conflicting_invariant.specified_for.name!r} "
+                            f"conflict for the description: {invariant.description!r}",
+                        )
+                    )
+            else:
+                description_map[invariant.description] = invariant
+
+    return errors
+
+
 def _verify_patterns_anchored_at_start_and_end(
     symbol_table: SymbolTable,
 ) -> List[Error]:
@@ -4622,6 +4670,8 @@ def _verify(symbol_table: SymbolTable, ontology: _hierarchy.Ontology) -> List[Er
 
     errors.extend(_verify_patterns_anchored_at_start_and_end(symbol_table=symbol_table))
 
+    errors.extend(_verify_invariant_descriptions_unique(symbol_table=symbol_table))
+
     if len(errors) > 0:
         return errors
 

aas_core_codegen/main.py

@@ -19,6 +19,7 @@
 import aas_core_codegen.jsonld.main as jsonld_main
 import aas_core_codegen.protobuf.main as protobuf_main
 import aas_core_codegen.python_protobuf.main as python_protobuf_main
+import aas_core_codegen.opcua.main as opcua_main
 from aas_core_codegen import run, specific_implementations
 from aas_core_codegen.common import LinenoColumner, assert_never
 
@@ -40,6 +41,7 @@ class Target(enum.Enum):
     JSONLD_CONTEXT = "jsonld_context"
     PROTOBUF = "protobuf"
     PYTHON_PROTOBUF = "python_protobuf"
+    OPCUA = "opcua"
 
 
 class Parameters:
@@ -174,6 +176,9 @@ def execute(params: Parameters, stdout: TextIO, stderr: TextIO) -> int:
     elif params.target is Target.PYTHON_PROTOBUF:
         return python_protobuf_main.execute(run_context, stdout=stdout, stderr=stderr)
 
+    elif params.target is Target.OPCUA:
+        return opcua_main.execute(run_context, stdout=stdout, stderr=stderr)
+
     else:
         assert_never(params.target)
 

aas_core_codegen/opcua/__init__.py

@@ -0,0 +1 @@
+"""Generate the OPC UA Schema node set corresponding to the meta-model."""