Python code capable of parsing matter IDL files (including some unit tests)

BUILD.gn

@@ -160,6 +160,7 @@ if (current_toolchain != "${dir_pw_toolchain}/default:default") {
       deps = [
         "//:fake_platform_tests",
         "//scripts/build:build_examples.tests",
+        "//scripts/idl:idl.tests",
         "//src:tests_run",
       ]
     }

scripts/idl/BUILD.gn

@@ -0,0 +1,35 @@
+# Copyright (c) 2022 Project CHIP Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import("//build_overrides/build.gni")
+import("//build_overrides/chip.gni")
+
+import("//build_overrides/pigweed.gni")
+import("$dir_pw_build/python.gni")
+
+pw_python_package("idl") {
+  setup = [ "setup.py" ]
+  inputs = [
+    # Dependency grammar
+    "matter_grammar.lark",
+  ]
+
+  sources = [
+    "__init__.py",
+    "matter_idl_parser.py",
+    "matter_idl_types.py",
+  ]
+
+  tests = [ "test_matter_idl_parser.py" ]
+}

scripts/idl/matter_grammar.lark

@@ -0,0 +1,59 @@
+struct: "struct"i id "{" struct_field* "}"
+enum: "enum"i id ":" data_type "{" enum_entry* "}"
+
+event: event_priority "event"i id "=" number "{" struct_field* "}"
+
+?event_priority: "critical"i -> critical_priority
+               | "info"i     -> info_priority
+               | "debug"i    -> debug_priority
+
+attribute: attribute_tag* "attribute"i   field
+attribute_tag: "readonly"i -> attr_readonly
+             | "global"i -> attr_global
+
+request_struct: "request"i struct
+response_struct: "response"i struct
+
+command: "command"i id "(" id? ")" ":" id "=" number ";"
+
+cluster: cluster_side "cluster"i id "=" number "{" (enum|event|attribute|struct|request_struct|response_struct|command)* "}"
+?cluster_side: "server"i -> server_cluster
+             | "client"i -> client_cluster
+
+endpoint: "endpoint"i number "{" endpoint_cluster* "}"
+endpoint_cluster: endpoint_cluster_type "cluster"i id ";"
+
+?endpoint_cluster_type: "server"i   -> endpoint_server_cluster
+                      | "binding"i  -> endpoint_binding_to_cluster
+
+enum_entry: id "=" number ";"
+number: POSITIVE_INTEGER | HEX_INTEGER
+
+struct_field: member_attribute* field
+
+member_attribute: "optional"i -> optional
+                | "nullable"i -> nullable
+
+field: data_type id list_marker? "=" number ";"
+list_marker: "[" "]"
+
+data_type: type ("<" number ">")?
+
+id: ID
+type: ID
+
+COMMENT: "{" /(.|\n)+/ "}"
+       | "//" /.*/
+
+POSITIVE_INTEGER: /\d+/ 
+HEX_INTEGER: /0x[A-Fa-f0-9]+/ 
+ID: /[a-zA-Z_][a-zA-Z0-9_]*/
+
+idl: (struct|enum|cluster|endpoint)*
+
+%import common.WS
+%import common.C_COMMENT
+%import common.CPP_COMMENT
+%ignore WS
+%ignore C_COMMENT
+%ignore CPP_COMMENT

scripts/idl/matter_idl_parser.py

@@ -0,0 +1,240 @@
+#!/usr/bin/env python
+
+import logging
+
+from lark import Lark
+from lark.visitors import Transformer, v_args
+
+try:
+    from .matter_idl_types import *
+except:
+    import os
+    import sys
+    sys.path.append(os.path.abspath(os.path.dirname(__file__)))
+
+    from matter_idl_types import *
+
+
+class MatterIdlTransformer(Transformer):
+    """A transformer capable to transform data
+       parsed by Lark according to matter_grammar.lark
+    """
+
+    def number(self, tokens):
+        """Numbers in the grammar are integers or hex numbers.
+        """
+        if len(tokens) != 1:
+            raise Error("Unexpected argument counts")
+
+        n = tokens[0].value
+        if n.startswith('0x'):
+            return int(n[2:], 16)
+        else:
+            return int(n)
+
+    def id(self, tokens):
+        """An id is a string containing an identifier
+        """
+        if len(tokens) != 1:
+            raise Error("Unexpected argument counts")
+        return tokens[0].value
+
+    def type(self, tokens):
+        """A type is just a string for the type
+        """
+        if len(tokens) != 1:
+            raise Error("Unexpected argument counts")
+        return tokens[0].value
+
+    def data_type(self, tokens):
+        if len(tokens) == 1:
+            return DataType(name=tokens[0])
+            # Just a string for data type
+        elif len(tokens) == 2:
+            return DataType(name=tokens[0], max_length=tokens[1])
+        else:
+            raise Error("Unexpected size for data type")
+
+    @v_args(inline=True)
+    def enum_entry(self, id, number):
+        return EnumEntry(name=id, code=number)
+
+    @v_args(inline=True)
+    def enum(self, id, type, *entries):
+        return Enum(name=id, base_type=type, entries=list(entries))
+
+    def field(self, args):
+        data_type, name = args[0], args[1]
+        is_list = (len(args) == 4)
+        code = args[-1]
+
+        return Field(data_type=data_type, name=name, code=code, is_list=is_list)
+
+    def optional(self, _):
+        return FieldAttribute.OPTIONAL
+
+    def nullable(self, _):
+        return FieldAttribute.NULLABLE
+
+    def attr_readonly(self, _):
+        return AttributeTag.READABLE
+
+    def attr_global(self, _):
+        return AttributeTag.GLOBAL
+
+    def critical_priority(self, _):
+        return EventPriority.CRITICAL
+
+    def info_priority(self, _):
+        return EventPriority.INFO
+
+    def debug_priority(self, _):
+        return EventPriority.DEBUG
+
+    def endpoint_server_cluster(self, _):
+        return EndpointContentType.SERVER_CLUSTER
+
+    def endpoint_binding_to_cluster(self, _):
+        return EndpointContentType.CLIENT_BINDING
+
+    def struct_field(self, args):
+        # Last argument is the named_member, the rest
+        # are attributes
+        field = args[-1]
+        field.attributes = set(args[:-1])
+        return field
+
+    def server_cluster(self, _):
+        return ClusterSide.SERVER
+
+    def client_cluster(self, _):
+        return ClusterSide.CLIENT
+
+    def command(self, args):
+        # A command has 3 arguments if no input or
+        # 4 arguments if input parameter is available
+        param_in = None
+        if len(args) > 3:
+            param_in = args[1]
+        return Command(name=args[0], input_param=param_in, output_param=args[-2], code=args[-1])
+
+    def event(self, args):
+        return Event(priority=args[0], name=args[1], code=args[2], fields=args[3:], )
+
+    def attribute(self, args):
+        tags = set(args[:-1])
+        # until we support write only (and need a bit of a reshuffle)
+        # if the 'attr_readonly == READABLE' is not in the list, we make things
+        # read/write
+        if AttributeTag.READABLE not in tags:
+            tags.add(AttributeTag.READABLE)
+            tags.add(AttributeTag.WRITABLE)
+
+        return Attribute(definition=args[-1], tags=tags)
+
+    @v_args(inline=True)
+    def struct(self, id, *fields):
+        return Struct(name=id, fields=list(fields))
+
+    @v_args(inline=True)
+    def request_struct(self, value):
+        value.tag = StructTag.REQUEST
+        return value
+
+    @v_args(inline=True)
+    def response_struct(self, value):
+        value.tag = StructTag.RESPONSE
+        return value
+
+    @v_args(inline=True)
+    def endpoint(self, number, *clusters):
+        endpoint = Endpoint(number=number)
+
+        for t, name in clusters:
+            if t == EndpointContentType.CLIENT_BINDING:
+                endpoint.client_bindings.append(name)
+            elif t == EndpointContentType.SERVER_CLUSTER:
+                endpoint.server_clusters.append(name)
+            else:
+                raise Error("Unknown endpoint content: %r" % t)
+
+        return endpoint
+
+    @v_args(inline=True)
+    def endpoint_cluster(self, t, id):
+        return (t, id)
+
+    @v_args(inline=True)
+    def cluster(self, side, name, code, *content):
+        result = Cluster(side=side, name=name, code=code)
+
+        for item in content:
+            if type(item) == Enum:
+                result.enums.append(item)
+            elif type(item) == Event:
+                result.events.append(item)
+            elif type(item) == Attribute:
+                result.attributes.append(item)
+            elif type(item) == Struct:
+                result.structs.append(item)
+            elif type(item) == Command:
+                result.commands.append(item)
+            else:
+                raise Error("UNKNOWN cluster content item: %r" % item)
+
+        return result
+
+    def idl(self, items):
+        idl = Idl()
+
+        for item in items:
+            if type(item) == Enum:
+                idl.enums.append(item)
+            elif type(item) == Struct:
+                idl.structs.append(item)
+            elif type(item) == Cluster:
+                idl.clusters.append(item)
+            elif type(item) == Endpoint:
+                idl.endpoints.append(item)
+            else:
+                raise Error("UNKNOWN idl content item: %r" % item)
+
+        return idl
+
+
+def CreateParser():
+    return Lark.open('matter_grammar.lark', rel_to=__file__, start='idl', parser='lalr', transformer=MatterIdlTransformer())
+
+
+if __name__ == '__main__':
+    import click
+    import coloredlogs
+
+    # Supported log levels, mapping string values required for argument
+    # parsing into logging constants
+    __LOG_LEVELS__ = {
+        'debug': logging.DEBUG,
+        'info': logging.INFO,
+        'warn': logging.WARN,
+        'fatal': logging.FATAL,
+    }
+
+    @click.command()
+    @click.option(
+        '--log-level',
+        default='INFO',
+        type=click.Choice(__LOG_LEVELS__.keys(), case_sensitive=False),
+        help='Determines the verbosity of script output.')
+    @click.argument('filename')
+    def main(log_level, filename=None):
+        coloredlogs.install(level=__LOG_LEVELS__[
+                            log_level], fmt='%(asctime)s %(levelname)-7s %(message)s')
+
+        logging.info("Starting to parse ...")
+        data = CreateParser().parse(open(filename).read())
+        logging.info("Parse completed")
+
+        logging.info("Data:")
+        print(data)
+
+    main()