| #!/usr/bin/env python |
| |
| import dataclasses |
| import functools |
| import logging |
| from typing import Dict, Optional, Set |
| |
| from lark import Lark |
| from lark.lexer import Token |
| from lark.visitors import Transformer, v_args |
| |
| try: |
| from matter_idl.matter_idl_types import AccessPrivilege |
| except ModuleNotFoundError: |
| import os |
| import sys |
| sys.path.append(os.path.dirname( |
| os.path.dirname(os.path.abspath(__file__)))) |
| |
| from matter_idl.matter_idl_types import AccessPrivilege |
| |
| from matter_idl.matter_idl_types import (ApiMaturity, Attribute, AttributeInstantiation, AttributeOperation, AttributeQuality, |
| AttributeStorage, Bitmap, Cluster, Command, CommandInstantiation, CommandQuality, |
| ConstantEntry, DataType, DeviceType, Endpoint, Enum, Event, EventPriority, EventQuality, |
| Field, FieldQuality, Idl, ParseMetaData, ServerClusterInstantiation, Struct, StructQuality, |
| StructTag) |
| |
| |
| def UnionOfAllFlags(flags_list): |
| if not flags_list: |
| return None |
| return functools.reduce(lambda a, b: a | b, flags_list) |
| |
| |
| class PrefixCppDocComment: |
| def __init__(self, token): |
| self.start_pos = token.start_pos |
| # Assume CPP comments: /**...*/ |
| self.value_len = len(token.value) # includes /***/ AND whitespace |
| self.value = token.value[3:-2].strip() |
| |
| def appply_to_idl(self, idl: Idl, content: str): |
| if self.start_pos is None: |
| return |
| |
| actual_pos = self.start_pos + self.value_len |
| while content[actual_pos] in ' \t\n\r': |
| actual_pos += 1 |
| |
| # A doc comment will apply to any supported element assuming it immediately |
| # preceeds id (skipping whitespace) |
| for item in self.supported_types(idl): |
| meta = item.parse_meta |
| if meta and meta.start_pos == actual_pos: |
| item.description = self.value |
| return |
| |
| def supported_types(self, idl: Idl): |
| """List all types supported by doc comments.""" |
| for cluster in idl.clusters: |
| yield cluster |
| |
| for command in cluster.commands: |
| yield command |
| |
| def __repr__(self): |
| return ("PREFIXDoc: %r at %r" % (self.value, self.start_pos)) |
| |
| |
| class AddServerClusterToEndpointTransform: |
| """Provides an 'apply' method that can be run on endpoints |
| to add a server cluster to the given endpoint. |
| """ |
| |
| def __init__(self, cluster: ServerClusterInstantiation): |
| self.cluster = cluster |
| |
| def apply(self, endpoint): |
| endpoint.server_clusters.append(self.cluster) |
| |
| |
| class AddBindingToEndpointTransform: |
| """Provides an 'apply' method that can be run on endpoints |
| to add a cluster binding to the given endpoint. |
| """ |
| |
| def __init__(self, name): |
| self.name = name |
| |
| def apply(self, endpoint): |
| endpoint.client_bindings.append(self.name) |
| |
| |
| class AddDeviceTypeToEndpointTransform: |
| """Provides an 'apply' method that can be run on endpoints |
| to add a device type to it |
| """ |
| |
| def __init__(self, device_type: DeviceType): |
| self.device_type = device_type |
| |
| def apply(self, endpoint): |
| endpoint.device_types.append(self.device_type) |
| |
| |
| class MatterIdlTransformer(Transformer): |
| """ |
| A transformer capable to transform data parsed by Lark according to |
| matter_grammar.lark. |
| |
| Generally transforms a ".matter" file into an Abstract Syntax Tree (AST). |
| End result will be a `matter_idl_types.Idl` value that represents the |
| entire parsed .matter file. |
| |
| The content of this file closely resembles the .lark input file and its |
| purpose is to convert LARK tokens (that ar generally inputted by name) |
| into underlying python types. |
| |
| Some documentation to get started is available at |
| https://lark-parser.readthedocs.io/en/latest/visitors.html#transformer |
| |
| TLDR would be: |
| When the ".lark" defines a token like `foo: number`, the transformer |
| has the option to define a method called `foo` which will take the |
| parsed input (as strings unless transformed) and interpret them. |
| |
| Actual parametes to the methods depend on the rules multiplicity and/or |
| optionally. |
| """ |
| |
| def __init__(self, skip_meta): |
| self.skip_meta = skip_meta |
| self.doc_comments = [] |
| |
| def positive_integer(self, tokens): |
| """Numbers in the grammar are integers or hex numbers. |
| """ |
| if len(tokens) != 1: |
| raise Exception("Unexpected argument counts") |
| |
| n = tokens[0].value |
| if n.startswith('0x'): |
| return int(n[2:], 16) |
| else: |
| return int(n) |
| |
| @v_args(inline=True) |
| def negative_integer(self, value): |
| return -value |
| |
| @v_args(inline=True) |
| def integer(self, value): |
| return value |
| |
| def bool_default_true(self, _): |
| return True |
| |
| def bool_default_false(self, _): |
| return False |
| |
| def provisional_api_maturity(self, _): |
| return ApiMaturity.PROVISIONAL |
| |
| def internal_api_maturity(self, _): |
| return ApiMaturity.INTERNAL |
| |
| def deprecated_api_maturity(self, _): |
| return ApiMaturity.DEPRECATED |
| |
| def stable_api_maturity(self, _): |
| return ApiMaturity.STABLE |
| |
| def id(self, tokens): |
| """An id is a string containing an identifier |
| """ |
| if len(tokens) != 1: |
| raise Exception("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 Exception("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 Exception("Unexpected size for data type") |
| |
| @v_args(inline=True) |
| def constant_entry(self, api_maturity, id, number): |
| if api_maturity is None: |
| api_maturity = ApiMaturity.STABLE |
| return ConstantEntry(name=id, code=number, api_maturity=api_maturity) |
| |
| @v_args(inline=True) |
| def enum(self, id, type, *entries): |
| return Enum(name=id, base_type=type, entries=list(entries)) |
| |
| @v_args(inline=True) |
| def bitmap(self, id, type, *entries): |
| return Bitmap(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 FieldQuality.OPTIONAL |
| |
| def nullable(self, _): |
| return FieldQuality.NULLABLE |
| |
| def fabric_sensitive(self, _): |
| return FieldQuality.FABRIC_SENSITIVE |
| |
| def attr_readonly(self, _): |
| return AttributeQuality.READABLE |
| |
| def attr_nosubscribe(self, _): |
| return AttributeQuality.NOSUBSCRIBE |
| |
| def attr_timed(self, _): |
| return AttributeQuality.TIMED_WRITE |
| |
| def attribute_qualities(self, qualities): |
| return UnionOfAllFlags(qualities) or AttributeQuality.NONE |
| |
| def struct_fabric_scoped(self, _): |
| return StructQuality.FABRIC_SCOPED |
| |
| def struct_qualities(self, qualities): |
| return UnionOfAllFlags(qualities) or StructQuality.NONE |
| |
| def critical_priority(self, _): |
| return EventPriority.CRITICAL |
| |
| def info_priority(self, _): |
| return EventPriority.INFO |
| |
| def debug_priority(self, _): |
| return EventPriority.DEBUG |
| |
| def event_fabric_sensitive(self, _): |
| return EventQuality.FABRIC_SENSITIVE |
| |
| def event_qualities(selt, qualities): |
| return UnionOfAllFlags(qualities) or EventQuality.NONE |
| |
| def timed_command(self, _): |
| return CommandQuality.TIMED_INVOKE |
| |
| def fabric_scoped_command(self, _): |
| return CommandQuality.FABRIC_SCOPED |
| |
| def command_qualities(self, attrs): |
| return UnionOfAllFlags(attrs) or CommandQuality.NONE |
| |
| def struct_field(self, args): |
| # Last argument is the named_member, the rest |
| # are qualities |
| field = args[-1] |
| field.qualities = UnionOfAllFlags(args[1:-1]) or FieldQuality.NONE |
| if args[0] is not None: |
| field.api_maturity = args[0] |
| return field |
| |
| def command_access(self, privilege): |
| return privilege[0] |
| |
| def command_with_access(self, args): |
| # Arguments |
| # - optional access for invoke |
| # - event identifier (name) |
| init_args = { |
| "name": args[-1] |
| } |
| if len(args) > 1: |
| init_args["invokeacl"] = args[0] |
| |
| return init_args |
| |
| # NOTE: awkward inline because the order of 'meta, children' vs 'children, meta' was flipped |
| # between lark versions in https://github.com/lark-parser/lark/pull/993 |
| @v_args(meta=True, inline=True) |
| def command(self, meta, *tuple_args): |
| # The command takes 4 arguments if no input argument, 5 if input |
| # argument is provided |
| args = list(tuple_args) # convert from tuple |
| if len(args) != 5: |
| args.insert(2, None) |
| |
| meta = None if self.skip_meta else ParseMetaData(meta) |
| |
| cmd = Command( |
| parse_meta=meta, |
| qualities=args[0], |
| input_param=args[2], output_param=args[3], code=args[4], |
| **args[1], |
| ) |
| |
| return cmd |
| |
| def event_access(self, privilege): |
| return privilege[0] |
| |
| def event_with_access(self, args): |
| # Arguments |
| # - optional access for read |
| # - event identifier (name) |
| init_args = { |
| "name": args[-1] |
| } |
| if len(args) > 1: |
| init_args["readacl"] = args[0] |
| |
| return init_args |
| |
| @v_args(inline=True) |
| def cluster_revision(self, revision): |
| return revision |
| |
| def event(self, args): |
| return Event(qualities=args[0], priority=args[1], code=args[3], fields=args[4:], **args[2]) |
| |
| def view_privilege(self, args): |
| return AccessPrivilege.VIEW |
| |
| def operate_privilege(self, args): |
| return AccessPrivilege.OPERATE |
| |
| def manage_privilege(self, args): |
| return AccessPrivilege.MANAGE |
| |
| def administer_privilege(self, args): |
| return AccessPrivilege.ADMINISTER |
| |
| def read_access(self, args): |
| return AttributeOperation.READ |
| |
| def write_access(self, args): |
| return AttributeOperation.WRITE |
| |
| @v_args(inline=True) |
| def attribute_access_entry(self, operation, access): |
| return (operation, access) |
| |
| def attribute_access(self, value): |
| # return value as-is to not need to deal with trees in `attribute_with_access` |
| return value |
| |
| def attribute_with_access(self, args): |
| # Input arguments are: |
| # - acl (optional list of pairs operation + access) |
| # - field definition |
| acl = {} |
| if len(args) > 1: |
| for operation, access in args[0]: |
| if operation == AttributeOperation.READ: |
| acl['readacl'] = access |
| elif operation == AttributeOperation.WRITE: |
| acl['writeacl'] = access |
| else: |
| raise Exception( |
| "Unknown attribute operation: %r" % operation) |
| |
| return (args[-1], acl) |
| |
| def ram_attribute(self, _): |
| return AttributeStorage.RAM |
| |
| def persist_attribute(self, _): |
| return AttributeStorage.PERSIST |
| |
| def callback_attribute(self, _): |
| return AttributeStorage.CALLBACK |
| |
| @v_args(meta=True, inline=True) |
| def endpoint_attribute_instantiation(self, meta, storage, id, default=None): |
| meta = None if self.skip_meta else ParseMetaData(meta) |
| return AttributeInstantiation(parse_meta=meta, name=id, storage=storage, default=default) |
| |
| @v_args(meta=True, inline=True) |
| def endpoint_command_instantiation(self, meta, id): |
| meta = None if self.skip_meta else ParseMetaData(meta) |
| return CommandInstantiation(parse_meta=meta, name=id) |
| |
| @v_args(meta=True, inline=True) |
| def endpoint_emitted_event(self, meta, id): |
| meta = None if self.skip_meta else ParseMetaData(meta) |
| return id |
| |
| def ESCAPED_STRING(self, s): |
| # handle escapes, skip the start and end quotes |
| return s.value[1:-1].encode('utf-8').decode('unicode-escape') |
| |
| @v_args(inline=True) |
| def attribute(self, qualities, definition_tuple): |
| (definition, acl) = definition_tuple |
| |
| # 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 AttributeQuality.READABLE not in qualities: |
| qualities |= AttributeQuality.READABLE |
| qualities |= AttributeQuality.WRITABLE |
| |
| return Attribute(definition=definition, qualities=qualities, **acl) |
| |
| @v_args(inline=True) |
| def struct(self, qualities, id, *fields): |
| return Struct(name=id, qualities=qualities, 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, id, code, *fields): |
| return Struct(name=id, tag=StructTag.RESPONSE, code=code, fields=list(fields)) |
| |
| @v_args(inline=True) |
| def endpoint(self, number, *transforms): |
| endpoint = Endpoint(number=number) |
| |
| for t in transforms: |
| t.apply(endpoint) |
| |
| return endpoint |
| |
| @v_args(inline=True) |
| def endpoint_device_type(self, name, code, version): |
| return AddDeviceTypeToEndpointTransform(DeviceType(name=name, code=code, version=version)) |
| |
| @v_args(inline=True) |
| def endpoint_cluster_binding(self, id): |
| return AddBindingToEndpointTransform(id) |
| |
| @v_args(meta=True, inline=True) |
| def endpoint_server_cluster(self, meta, id, *content): |
| meta = None if self.skip_meta else ParseMetaData(meta) |
| |
| attributes = [] |
| commands = [] |
| events = set() |
| |
| for item in content: |
| if isinstance(item, AttributeInstantiation): |
| attributes.append(item) |
| elif isinstance(item, CommandInstantiation): |
| commands.append(item) |
| else: |
| events.add(item) |
| return AddServerClusterToEndpointTransform( |
| ServerClusterInstantiation(parse_meta=meta, name=id, attributes=attributes, events_emitted=events, commands=commands)) |
| |
| @v_args(inline=True) |
| def cluster_content(self, api_maturity, element): |
| if api_maturity is not None: |
| element.api_maturity = api_maturity |
| return element |
| |
| @v_args(inline=True, meta=True) |
| def cluster(self, meta, api_maturity, name, code, revision, *content): |
| meta = None if self.skip_meta else ParseMetaData(meta) |
| |
| if api_maturity is None: |
| api_maturity = ApiMaturity.STABLE |
| |
| if not revision: |
| revision = 1 |
| |
| result = Cluster(parse_meta=meta, name=name, code=code, |
| revision=revision, api_maturity=api_maturity) |
| |
| for item in content: |
| if isinstance(item, Enum): |
| result.enums.append(item) |
| elif isinstance(item, Bitmap): |
| result.bitmaps.append(item) |
| elif isinstance(item, Event): |
| result.events.append(item) |
| elif isinstance(item, Attribute): |
| result.attributes.append(item) |
| elif isinstance(item, Struct): |
| result.structs.append(item) |
| elif isinstance(item, Command): |
| result.commands.append(item) |
| else: |
| raise Exception("UNKNOWN cluster content item: %r" % item) |
| |
| return result |
| |
| def idl(self, items): |
| clusters = [] |
| endpoints = [] |
| |
| global_bitmaps = [] |
| global_enums = [] |
| global_structs = [] |
| |
| for item in items: |
| if isinstance(item, Cluster): |
| clusters.append(item) |
| elif isinstance(item, Endpoint): |
| endpoints.append(item) |
| elif isinstance(item, Enum): |
| global_enums.append(dataclasses.replace(item, is_global=True)) |
| elif isinstance(item, Bitmap): |
| global_bitmaps.append(dataclasses.replace(item, is_global=True)) |
| elif isinstance(item, Struct): |
| global_structs.append(dataclasses.replace(item, is_global=True)) |
| else: |
| raise Exception("UNKNOWN idl content item: %r" % item) |
| |
| return Idl(clusters=clusters, endpoints=endpoints, global_bitmaps=global_bitmaps, global_enums=global_enums, global_structs=global_structs) |
| |
| def prefix_doc_comment(self): |
| print("TODO: prefix") |
| |
| # Processing of (potential-doc)-comments: |
| def c_comment(self, token: Token): |
| """Processes comments starting with "/*" """ |
| if token.value.startswith("/**"): |
| self.doc_comments.append(PrefixCppDocComment(token)) |
| |
| |
| def _referenced_type_names(cluster: Cluster) -> Set[str]: |
| """ |
| Return the names of all data types referenced by the given cluster. |
| """ |
| types = set() |
| for s in cluster.structs: |
| for f in s.fields: |
| types.add(f.data_type.name) |
| |
| for e in cluster.events: |
| for f in e.fields: |
| types.add(f.data_type.name) |
| |
| for a in cluster.attributes: |
| types.add(a.definition.data_type.name) |
| |
| return types |
| |
| |
| class GlobalMapping: |
| """ |
| Maintains global type mapping from an IDL |
| """ |
| |
| def __init__(self, idl: Idl): |
| self.bitmap_map = {b.name: b for b in idl.global_bitmaps} |
| self.enum_map = {e.name: e for e in idl.global_enums} |
| self.struct_map = {s.name: s for s in idl.global_structs} |
| |
| self.global_types = set(self.bitmap_map.keys()).union(set(self.enum_map.keys())).union(set(self.struct_map.keys())) |
| |
| # Spec does not enforce unique naming in bitmap/enum/struct, however in practice |
| # if we have both enum Foo and bitmap Foo for example, it would be impossible |
| # to disambiguate `attribute Foo foo = 1` for the actual type we want. |
| # |
| # As a result, we do not try to namespace this and just error out |
| if len(self.global_types) != len(self.bitmap_map) + len(self.enum_map) + len(self.struct_map): |
| raise ValueError("Global type names are not unique.") |
| |
| def merge_global_types_into_cluster(self, cluster: Cluster) -> Cluster: |
| """ |
| Merges all referenced global types (bitmaps/enums/structs) into the cluster types. |
| This happens recursively. |
| """ |
| global_types_added = set() |
| |
| changed = True |
| while changed: |
| changed = False |
| for type_name in _referenced_type_names(cluster): |
| if type_name not in self.global_types: |
| continue # not a global type name |
| |
| if type_name in global_types_added: |
| continue # already added |
| |
| # check if this is a global type |
| if type_name in self.bitmap_map: |
| global_types_added.add(type_name) |
| changed = True |
| cluster.bitmaps.append(self.bitmap_map[type_name]) |
| elif type_name in self.enum_map: |
| global_types_added.add(type_name) |
| changed = True |
| cluster.enums.append(self.enum_map[type_name]) |
| elif type_name in self.struct_map: |
| global_types_added.add(type_name) |
| changed = True |
| cluster.structs.append(self.struct_map[type_name]) |
| |
| return cluster |
| |
| |
| def _merge_global_types_into_clusters(idl: Idl) -> Idl: |
| """ |
| Adds bitmaps/enums/structs from idl.global_* into clusters as long as |
| clusters reference those type names |
| """ |
| mapping = GlobalMapping(idl) |
| return dataclasses.replace(idl, clusters=[mapping.merge_global_types_into_cluster(cluster) for cluster in idl.clusters]) |
| |
| |
| class ParserWithLines: |
| def __init__(self, skip_meta: bool, merge_globals: bool): |
| self.transformer = MatterIdlTransformer(skip_meta) |
| self.merge_globals = merge_globals |
| |
| # NOTE: LALR parser is fast. While Earley could parse more ambigous grammars, |
| # earley is much slower: |
| # - 0.39s LALR parsing of all-clusters-app.matter |
| # - 2.26s Earley parsing of the same thing. |
| # For this reason, every attempt should be made to make the grammar context free |
| self.parser = Lark.open( |
| 'matter_grammar.lark', rel_to=__file__, start='idl', parser='lalr', propagate_positions=True, |
| maybe_placeholders=True, |
| # separate callbacks to ignore from regular parsing (no tokens) |
| # while still getting notified about them |
| lexer_callbacks={ |
| 'C_COMMENT': self.transformer.c_comment, |
| } |
| ) |
| |
| def parse(self, file: str, file_name: Optional[str] = None): |
| idl = self.transformer.transform(self.parser.parse(file)) |
| idl.parse_file_name = file_name |
| |
| # ZAP may generate the same definition of clusters several times. |
| # Validate that if a cluster is defined, its definition is IDENTICAL |
| # |
| # TODO: this is not ideal and zap provides some iteration that seems |
| # to not care about side: `all_user_clusters_irrespective_of_side` |
| # however that one loses at least `description` and switches |
| # ordering. |
| # |
| # As a result, for now allow multiple definitions IF AND ONLY IF identical |
| # |
| # A zap PR to allow us to not need this is: |
| # https://github.com/project-chip/zap/pull/1216 |
| clusters: Dict[int, Cluster] = {} |
| for c in idl.clusters: |
| if c.code in clusters: |
| if c != clusters[c.code]: |
| raise Exception( |
| f"Different cluster definition for {c.name}/{c.code}") |
| else: |
| clusters[c.code] = c |
| idl.clusters = [c for c in clusters.values()] |
| |
| for comment in self.transformer.doc_comments: |
| comment.appply_to_idl(idl, file) |
| |
| if self.merge_globals: |
| idl = _merge_global_types_into_clusters(idl) |
| |
| return idl |
| |
| |
| def CreateParser(skip_meta: bool = False, merge_globals=True): |
| """ |
| Generates a parser that will process a ".matter" file into a IDL |
| |
| Arguments: |
| skip_meta - do not add metadata (line position) for items. Metadata is |
| useful for error reporting, however it does not work well |
| for unit test comparisons |
| |
| merge_globals - places global items (enums/bitmaps/structs) into any |
| clusters that reference them, so that cluster types |
| are self-sufficient. Useful as a backwards-compatible |
| code generation if global definitions are not supported. |
| |
| """ |
| return ParserWithLines(skip_meta, merge_globals) |
| |
| |
| if __name__ == '__main__': |
| # This Parser is generally not intended to be run as a stand-alone binary. |
| # The ability to run is for debug and to print out the parsed AST. |
| import pprint |
| |
| import click |
| |
| # 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(list(__LOG_LEVELS__.keys()), case_sensitive=False), |
| help='Determines the verbosity of script output.') |
| @click.argument('filename') |
| def main(log_level, filename=None): |
| logging.basicConfig( |
| level=__LOG_LEVELS__[log_level], |
| format='%(asctime)s %(levelname)-7s %(message)s', |
| ) |
| |
| logging.info("Starting to parse ...") |
| data = CreateParser().parse(open(filename).read(), file_name=filename) |
| logging.info("Parse completed") |
| |
| logging.info("Data:") |
| pprint.pp(data) |
| |
| main(auto_envvar_prefix='CHIP') |