| # |
| # Copyright (c) 2023 Project CHIP Authors |
| # All rights reserved. |
| # |
| # 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 glob |
| import logging |
| import os |
| import typing |
| import xml.etree.ElementTree as ElementTree |
| from copy import deepcopy |
| from dataclasses import dataclass |
| from enum import Enum, auto |
| from typing import Callable, Optional |
| |
| import chip.clusters as Clusters |
| from chip.tlv import uint |
| from conformance_support import (OPTIONAL_CONFORM, TOP_LEVEL_CONFORMANCE_TAGS, ConformanceDecision, ConformanceException, |
| ConformanceParseParameters, feature, is_disallowed, mandatory, optional, or_operation, |
| parse_callable_from_xml, parse_device_type_callable_from_xml) |
| from global_attribute_ids import GlobalAttributeIds |
| from matter_testing_support import (AttributePathLocation, ClusterPathLocation, CommandPathLocation, DeviceTypePathLocation, |
| EventPathLocation, FeaturePathLocation, ProblemNotice, ProblemSeverity) |
| |
| _PRIVILEGE_STR = { |
| None: "N/A", |
| Clusters.AccessControl.Enums.AccessControlEntryPrivilegeEnum.kView: "V", |
| Clusters.AccessControl.Enums.AccessControlEntryPrivilegeEnum.kOperate: "O", |
| Clusters.AccessControl.Enums.AccessControlEntryPrivilegeEnum.kManage: "M", |
| Clusters.AccessControl.Enums.AccessControlEntryPrivilegeEnum.kAdminister: "A", |
| } |
| |
| |
| def to_access_code(privilege: Clusters.AccessControl.Enums.AccessControlEntryPrivilegeEnum) -> str: |
| return _PRIVILEGE_STR.get(privilege, "") |
| |
| |
| class SpecParsingException(Exception): |
| pass |
| |
| |
| @dataclass |
| class XmlFeature: |
| code: str |
| name: str |
| conformance: Callable[[uint], ConformanceDecision] |
| |
| |
| @dataclass |
| class XmlAttribute: |
| name: str |
| datatype: str |
| conformance: Callable[[uint], ConformanceDecision] |
| read_access: Clusters.AccessControl.Enums.AccessControlEntryPrivilegeEnum |
| write_access: Clusters.AccessControl.Enums.AccessControlEntryPrivilegeEnum |
| write_optional: bool |
| |
| def access_string(self): |
| read_marker = "R" if self.read_access is not Clusters.AccessControl.Enums.AccessControlEntryPrivilegeEnum.kUnknownEnumValue else "" |
| write_marker = "W" if self.write_access is not Clusters.AccessControl.Enums.AccessControlEntryPrivilegeEnum.kUnknownEnumValue else "" |
| read_access_marker = f'{to_access_code(self.read_access)}' |
| write_access_marker = f'{to_access_code(self.write_access)}' |
| return f'{read_marker}{write_marker} {read_access_marker}{write_access_marker}' |
| |
| def __str__(self): |
| return f'{self.name}: datatype: {self.datatype} conformance: {str(self.conformance)}, access = {self.access_string()}' |
| |
| |
| @dataclass |
| class XmlCommand: |
| id: int |
| name: str |
| conformance: Callable[[uint], ConformanceDecision] |
| |
| |
| @dataclass |
| class XmlEvent: |
| name: str |
| conformance: Callable[[uint], ConformanceDecision] |
| |
| |
| @dataclass |
| class XmlCluster: |
| name: str |
| revision: int |
| derived: str |
| feature_map: dict[str, uint] |
| attribute_map: dict[str, uint] |
| command_map: dict[str, uint] |
| # mask to XmlFeature |
| features: dict[uint, XmlFeature] |
| # IDs to class |
| attributes: dict[uint, XmlAttribute] |
| accepted_commands: dict[uint, XmlCommand] |
| generated_commands: dict[uint, XmlCommand] |
| unknown_commands: list[XmlCommand] |
| events: dict[uint, XmlEvent] |
| pics: str |
| |
| |
| class ClusterSide(Enum): |
| SERVER = auto() |
| CLIENT = auto() |
| |
| |
| @dataclass |
| class XmlDeviceTypeClusterRequirements: |
| name: str |
| side: ClusterSide |
| conformance: Callable[[uint, list[uint], list[uint]], ConformanceDecision] |
| # TODO: add element requirements |
| |
| def __str__(self): |
| return f'{self.name}: {str(self.conformance)}' |
| |
| |
| @dataclass |
| class XmlDeviceType: |
| name: str |
| revision: int |
| server_clusters: dict[uint, XmlDeviceTypeClusterRequirements] |
| client_clusters: dict[uint, XmlDeviceTypeClusterRequirements] |
| # Keeping these as strings for now because the exact definitions are being discussed in DMTT |
| classification_class: str |
| classification_scope: str |
| |
| def __str__(self): |
| msg = f'{self.name} - Revision {self.revision}, Class {self.classification_class}, Scope {self.classification_scope}\n' |
| msg += ' Server clusters\n' |
| for id, c in self.server_clusters.items(): |
| msg = msg + f' {id}: {str(c)}\n' |
| msg += ' Client clusters\n' |
| for id, c in self.client_clusters.items(): |
| msg = msg + f' {id}: {str(c)}\n' |
| return msg |
| |
| |
| class CommandType(Enum): |
| ACCEPTED = auto() |
| GENERATED = auto() |
| # This will happen for derived clusters, where the direction isn't noted. On normal clusters, this is a problem. |
| UNKNOWN = auto() |
| |
| |
| # workaround for aliased clusters PICS not appearing in the xml. Remove this once https://github.com/csa-data-model/projects/issues/461 is addressed |
| ALIAS_PICS = {0x040C: 'CMOCONC', |
| 0x040D: 'CDOCONC', |
| 0x0413: 'NDOCONC', |
| 0x0415: 'OZCONC', |
| 0x042A: 'PMICONC', |
| 0x042B: 'FLDCONC', |
| 0x042C: 'PMHCONC', |
| 0x042D: 'PMKCONC', |
| 0x042E: 'TVOCCONC', |
| 0x042F: 'RNCONC', |
| 0x0071: 'HEPAFREMON', |
| 0x0072: 'ACFREMON', |
| 0x0405: 'RH', |
| 0x001C: 'PWM'} |
| |
| CLUSTER_NAME_FIXES = {0x0036: 'WiFi Network Diagnostics', 0x042a: 'PM25 Concentration Measurement', 0x0006: 'On/Off'} |
| DEVICE_TYPE_NAME_FIXES = {0x010b: 'Dimmable Plug-In Unit', 0x010a: 'On/Off Plug-in Unit'} |
| |
| |
| def get_location_from_element(element: ElementTree.Element, cluster_id: int): |
| if element.tag == 'feature': |
| location = FeaturePathLocation(endpoint_id=0, cluster_id=cluster_id, feature_code=element.attrib['code']) |
| elif element.tag == 'command': |
| location = CommandPathLocation(endpoint_id=0, cluster_id=cluster_id, command_id=int(element.attrib['id'], 0)) |
| elif element.tag == 'attribute': |
| location = AttributePathLocation(endpoint_id=0, cluster_id=cluster_id, attribute_id=int(element.attrib['id'], 0)) |
| elif element.tag == 'event': |
| location = EventPathLocation(endpoint_id=0, cluster_id=cluster_id, event_id=int(element.attrib['id'], 0)) |
| else: |
| location = ClusterPathLocation(endpoint_id=0, cluster_id=cluster_id) |
| return location |
| |
| |
| def get_conformance(element: ElementTree.Element, cluster_id: int) -> tuple[ElementTree.Element, typing.Optional[ProblemNotice]]: |
| for sub in element: |
| if sub.tag in TOP_LEVEL_CONFORMANCE_TAGS: |
| return sub, None |
| location = get_location_from_element(element, cluster_id) |
| problem = ProblemNotice(test_name='Spec XML parsing', location=location, |
| severity=ProblemSeverity.WARNING, problem='Unable to find conformance element') |
| return ElementTree.Element(OPTIONAL_CONFORM), problem |
| |
| |
| class ClusterParser: |
| def __init__(self, cluster, cluster_id, name): |
| self._problems: list[ProblemNotice] = [] |
| self._cluster = cluster |
| self._cluster_id = cluster_id |
| self._name = name |
| |
| self._derived = None |
| try: |
| classification = next(cluster.iter('classification')) |
| hierarchy = classification.attrib['hierarchy'] |
| if hierarchy.lower() == 'derived': |
| self._derived = classification.attrib['baseCluster'] |
| except (KeyError, StopIteration): |
| self._derived = None |
| |
| try: |
| classification = next(cluster.iter('classification')) |
| self._pics = classification.attrib['picsCode'] |
| except (KeyError, StopIteration): |
| self._pics = None |
| |
| if self._cluster_id in ALIAS_PICS.keys(): |
| self._pics = ALIAS_PICS[cluster_id] |
| |
| self.feature_elements = self.get_all_feature_elements() |
| self.attribute_elements = self.get_all_attribute_elements() |
| self.command_elements = self.get_all_command_elements() |
| self.event_elements = self.get_all_event_elements() |
| self.params = ConformanceParseParameters(feature_map=self.create_feature_map(), attribute_map=self.create_attribute_map(), |
| command_map=self.create_command_map()) |
| |
| def get_conformance(self, element: ElementTree.Element) -> ElementTree.Element: |
| element, problem = get_conformance(element, self._cluster_id) |
| if problem: |
| self._problems.append(problem) |
| return element |
| |
| def get_access(self, element: ElementTree.Element) -> Optional[ElementTree.Element]: |
| for sub in element: |
| if sub.tag == 'access': |
| return sub |
| return None |
| |
| def get_all_type(self, type_container: str, type_name: str, key_name: str) -> list[tuple[ElementTree.Element, ElementTree.Element, ElementTree.Element]]: |
| ret = [] |
| container_tags = self._cluster.iter(type_container) |
| for container in container_tags: |
| elements = container.iter(type_name) |
| for element in elements: |
| try: |
| element.attrib[key_name] |
| except KeyError: |
| # This is a conformance tag, which uses the same name |
| continue |
| conformance = self.get_conformance(element) |
| access = self.get_access(element) |
| ret.append((element, conformance, access)) |
| return ret |
| |
| def get_all_feature_elements(self) -> list[tuple[ElementTree.Element, ElementTree.Element]]: |
| ''' Returns a list of features and their conformances''' |
| return self.get_all_type('features', 'feature', 'code') |
| |
| def get_all_attribute_elements(self) -> list[tuple[ElementTree.Element, ElementTree.Element]]: |
| ''' Returns a list of attributes and their conformances''' |
| return self.get_all_type('attributes', 'attribute', 'id') |
| |
| def get_all_command_elements(self) -> list[tuple[ElementTree.Element, ElementTree.Element]]: |
| ''' Returns a list of commands and their conformances ''' |
| return self.get_all_type('commands', 'command', 'id') |
| |
| def get_all_event_elements(self) -> list[tuple[ElementTree.Element, ElementTree.Element]]: |
| ''' Returns a list of events and their conformances''' |
| return self.get_all_type('events', 'event', 'id') |
| |
| def create_feature_map(self) -> dict[str, uint]: |
| features = {} |
| for element, _, _ in self.feature_elements: |
| features[element.attrib['code']] = 1 << int(element.attrib['bit'], 0) |
| return features |
| |
| def create_attribute_map(self) -> dict[str, uint]: |
| attributes = {} |
| for element, conformance, _ in self.attribute_elements: |
| attributes[element.attrib['name']] = int(element.attrib['id'], 0) |
| return attributes |
| |
| def create_command_map(self) -> dict[str, uint]: |
| commands = {} |
| for element, _, _ in self.command_elements: |
| commands[element.attrib['name']] = int(element.attrib['id'], 0) |
| return commands |
| |
| def parse_conformance(self, conformance_xml: ElementTree.Element) -> Callable: |
| try: |
| return parse_callable_from_xml(conformance_xml, self.params) |
| except ConformanceException as ex: |
| # Just point to the general cluster, because something is mismatched, but it's not clear what |
| location = ClusterPathLocation(endpoint_id=0, cluster_id=self._cluster_id) |
| self._problems.append(ProblemNotice(test_name='Spec XML parsing', location=location, |
| severity=ProblemSeverity.WARNING, problem=str(ex))) |
| return None |
| |
| def parse_write_optional(self, element_xml: ElementTree.Element, access_xml: ElementTree.Element) -> bool: |
| return access_xml.attrib['write'] == 'optional' |
| |
| def parse_access(self, element_xml: ElementTree.Element, access_xml: ElementTree.Element, conformance: Callable) -> tuple[Optional[Clusters.AccessControl.Enums.AccessControlEntryPrivilegeEnum], Optional[Clusters.AccessControl.Enums.AccessControlEntryPrivilegeEnum], Optional[Clusters.AccessControl.Enums.AccessControlEntryPrivilegeEnum]]: |
| ''' Returns a tuple of access types for read / write / invoke''' |
| def str_to_access_type(privilege_str: str) -> Clusters.AccessControl.Enums.AccessControlEntryPrivilegeEnum: |
| if privilege_str == 'view': |
| return Clusters.AccessControl.Enums.AccessControlEntryPrivilegeEnum.kView |
| if privilege_str == 'operate': |
| return Clusters.AccessControl.Enums.AccessControlEntryPrivilegeEnum.kOperate |
| if privilege_str == 'manage': |
| return Clusters.AccessControl.Enums.AccessControlEntryPrivilegeEnum.kManage |
| if privilege_str == 'admin': |
| return Clusters.AccessControl.Enums.AccessControlEntryPrivilegeEnum.kAdminister |
| |
| # We don't know what this means, for now, assume no access and mark a warning |
| location = get_location_from_element(element_xml, self._cluster_id) |
| self._problems.append(ProblemNotice(test_name='Spec XML parsing', location=location, |
| severity=ProblemSeverity.WARNING, problem=f'Unknown access type {privilege_str}')) |
| return Clusters.AccessControl.Enums.AccessControlEntryPrivilegeEnum.kUnknownEnumValue |
| |
| if access_xml is None: |
| # Derived clusters can inherit their access from the base and that's fine, so don't add an error |
| # Similarly, pure base clusters can have the access defined in the derived clusters. If neither has it defined, |
| # we will determine this at the end when we put these together. |
| # Things with deprecated conformance don't get an access element, and that is also fine. |
| # If a device properly passes the conformance test, such elements are guaranteed not to appear on the device. |
| if self._derived is not None or is_disallowed(conformance): |
| return (None, None, None) |
| |
| location = get_location_from_element(element_xml, self._cluster_id) |
| self._problems.append(ProblemNotice(test_name='Spec XML parsing', location=location, |
| severity=ProblemSeverity.WARNING, problem='Unable to find access element')) |
| return (None, None, None) |
| try: |
| read_access = str_to_access_type(access_xml.attrib['readPrivilege']) |
| except KeyError: |
| read_access = Clusters.AccessControl.Enums.AccessControlEntryPrivilegeEnum.kUnknownEnumValue |
| try: |
| write_access = str_to_access_type(access_xml.attrib['writePrivilege']) |
| except KeyError: |
| write_access = Clusters.AccessControl.Enums.AccessControlEntryPrivilegeEnum.kUnknownEnumValue |
| try: |
| invoke_access = str_to_access_type(access_xml.attrib['invokePrivilege']) |
| except KeyError: |
| invoke_access = Clusters.AccessControl.Enums.AccessControlEntryPrivilegeEnum.kUnknownEnumValue |
| return (read_access, write_access, invoke_access) |
| |
| def parse_features(self) -> dict[uint, XmlFeature]: |
| features = {} |
| for element, conformance_xml, _ in self.feature_elements: |
| mask = 1 << int(element.attrib['bit'], 0) |
| conformance = self.parse_conformance(conformance_xml) |
| if conformance is None: |
| continue |
| features[mask] = XmlFeature(code=element.attrib['code'], name=element.attrib['name'], |
| conformance=conformance) |
| return features |
| |
| def parse_attributes(self) -> dict[uint, XmlAttribute]: |
| attributes = {} |
| for element, conformance_xml, access_xml in self.attribute_elements: |
| code = int(element.attrib['id'], 0) |
| # Some deprecated attributes don't have their types included, for now, lets just fallback to UNKNOWN |
| try: |
| datatype = element.attrib['type'] |
| except KeyError: |
| datatype = 'UNKNOWN' |
| conformance = self.parse_conformance(conformance_xml) |
| if conformance is None: |
| continue |
| if code in attributes: |
| # This is one of those fun ones where two different rows have the same id and name, but differ in conformance and ranges |
| # I don't have a good way to relate the ranges to the conformance, but they're both acceptable, so let's just or them. |
| conformance = or_operation([conformance, attributes[code].conformance]) |
| read_access, write_access, _ = self.parse_access(element, access_xml, conformance) |
| write_optional = False |
| if write_access not in [None, Clusters.AccessControl.Enums.AccessControlEntryPrivilegeEnum.kUnknownEnumValue]: |
| write_optional = self.parse_write_optional(element, access_xml) |
| attributes[code] = XmlAttribute(name=element.attrib['name'], datatype=datatype, |
| conformance=conformance, read_access=read_access, write_access=write_access, write_optional=write_optional) |
| # Add in the global attributes for the base class |
| for id in GlobalAttributeIds: |
| # TODO: Add data type here. Right now it's unused. We should parse this from the spec. |
| attributes[id] = XmlAttribute(name=id.to_name(), datatype="", conformance=mandatory( |
| ), read_access=Clusters.AccessControl.Enums.AccessControlEntryPrivilegeEnum.kView, write_access=Clusters.AccessControl.Enums.AccessControlEntryPrivilegeEnum.kUnknownEnumValue, write_optional=False) |
| return attributes |
| |
| def get_command_type(self, element: ElementTree.Element) -> CommandType: |
| try: |
| if element.attrib['direction'].lower() == 'responsefromserver': |
| return CommandType.GENERATED |
| if element.attrib['direction'].lower() == 'commandtoclient': |
| return CommandType.UNKNOWN |
| if element.attrib['direction'].lower() == 'commandtoserver': |
| return CommandType.ACCEPTED |
| raise Exception(f"Unknown direction: {element.attrib['direction']}") |
| except KeyError: |
| return CommandType.UNKNOWN |
| |
| def parse_unknown_commands(self) -> list[XmlCommand]: |
| commands = [] |
| for element, conformance_xml, access_xml in self.command_elements: |
| if self.get_command_type(element) != CommandType.UNKNOWN: |
| continue |
| code = int(element.attrib['id'], 0) |
| conformance = self.parse_conformance(conformance_xml) |
| commands.append(XmlCommand(id=code, name=element.attrib['name'], conformance=conformance)) |
| return commands |
| |
| def parse_commands(self, command_type: CommandType) -> dict[uint, XmlCommand]: |
| commands = {} |
| for element, conformance_xml, access_xml in self.command_elements: |
| if self.get_command_type(element) != command_type: |
| continue |
| code = int(element.attrib['id'], 0) |
| conformance = self.parse_conformance(conformance_xml) |
| if conformance is None: |
| continue |
| if code in commands: |
| conformance = or_operation([conformance, commands[code].conformance]) |
| commands[code] = XmlCommand(id=code, name=element.attrib['name'], conformance=conformance) |
| return commands |
| |
| def parse_events(self) -> dict[uint, XmlAttribute]: |
| events = {} |
| for element, conformance_xml, access_xml in self.event_elements: |
| code = int(element.attrib['id'], 0) |
| conformance = self.parse_conformance(conformance_xml) |
| if conformance is None: |
| continue |
| if code in events: |
| conformance = or_operation([conformance, events[code].conformance]) |
| events[code] = XmlEvent(name=element.attrib['name'], conformance=conformance) |
| return events |
| |
| def create_cluster(self) -> XmlCluster: |
| try: |
| revision = int(self._cluster.attrib['revision'], 0) |
| except ValueError: |
| revision = 0 |
| return XmlCluster(revision=revision, derived=self._derived, |
| name=self._name, feature_map=self.params.feature_map, |
| attribute_map=self.params.attribute_map, command_map=self.params.command_map, |
| features=self.parse_features(), |
| attributes=self.parse_attributes(), |
| accepted_commands=self.parse_commands(CommandType.ACCEPTED), |
| generated_commands=self.parse_commands(CommandType.GENERATED), |
| unknown_commands=self.parse_unknown_commands(), |
| events=self.parse_events(), pics=self._pics) |
| |
| def get_problems(self) -> list[ProblemNotice]: |
| return self._problems |
| |
| |
| def add_cluster_data_from_xml(xml: ElementTree.Element, clusters: dict[int, XmlCluster], pure_base_clusters: dict[str, XmlCluster], ids_by_name: dict[str, int], problems: list[ProblemNotice]) -> None: |
| ''' Adds cluster data to the supplied dicts as appropriate |
| |
| xml: XML element read from from the XML cluster file |
| clusters: dict of id -> XmlCluster. This function will append new clusters as appropriate to this dict. |
| pure_base_clusters: dict of base name -> XmlCluster. This data structure is used to hold pure base clusters that don't have |
| an ID. This function will append new pure base clusters as appropriate to this dict. |
| ids_by_name: dict of cluster name -> ID. This function will append new IDs as appropriate to this dict. |
| problems: list of any problems encountered during spec parsing. This function will append problems as appropriate to this list. |
| ''' |
| cluster = xml.iter('cluster') |
| for c in cluster: |
| ids = c.iter('clusterId') |
| for id in ids: |
| name = id.get('name') |
| cluster_id = id.get('id') |
| if cluster_id: |
| cluster_id = int(id.get('id'), 0) |
| ids_by_name[name] = cluster_id |
| |
| parser = ClusterParser(c, cluster_id, name) |
| new = parser.create_cluster() |
| problems = problems + parser.get_problems() |
| |
| if cluster_id: |
| clusters[cluster_id] = new |
| else: |
| # Fully derived clusters have no id, but also shouldn't appear on a device. |
| # We do need to keep them, though, because we need to update the derived |
| # clusters. We keep them in a special dict by name, so they can be thrown |
| # away later. |
| pure_base_clusters[name] = new |
| |
| |
| def check_clusters_for_unknown_commands(clusters: dict[int, XmlCluster], problems: list[ProblemNotice]): |
| for id, cluster in clusters.items(): |
| for cmd in cluster.unknown_commands: |
| problems.append(ProblemNotice(test_name="Spec XML parsing", location=CommandPathLocation( |
| endpoint_id=0, cluster_id=id, command_id=cmd.id), severity=ProblemSeverity.WARNING, problem="Command with unknown direction")) |
| |
| |
| class PrebuiltDataModelDirectory(Enum): |
| k1_3 = auto() |
| kInProgress = auto() |
| kMaster = auto() |
| |
| |
| class DataModelLevel(str, Enum): |
| kCluster = 'clusters' |
| kDeviceType = 'device_types' |
| |
| |
| def _get_data_model_directory(data_model_directory: typing.Union[PrebuiltDataModelDirectory, str], data_model_level: DataModelLevel) -> str: |
| if data_model_directory == PrebuiltDataModelDirectory.k1_3: |
| return os.path.join(os.path.dirname(os.path.realpath(__file__)), '..', '..', 'data_model', '1.3', data_model_level) |
| elif data_model_directory == PrebuiltDataModelDirectory.kInProgress: |
| return os.path.join(os.path.dirname(os.path.realpath(__file__)), '..', '..', 'data_model', 'in_progress', data_model_level) |
| elif data_model_directory == PrebuiltDataModelDirectory.kMaster: |
| return os.path.join(os.path.dirname(os.path.realpath(__file__)), '..', '..', 'data_model', 'master', data_model_level) |
| else: |
| return data_model_directory |
| |
| |
| def build_xml_clusters(data_model_directory: typing.Union[PrebuiltDataModelDirectory, str] = PrebuiltDataModelDirectory.kMaster) -> tuple[dict[uint, XmlCluster], list[ProblemNotice]]: |
| dir = _get_data_model_directory(data_model_directory, DataModelLevel.kCluster) |
| |
| clusters: dict[int, XmlCluster] = {} |
| pure_base_clusters: dict[str, XmlCluster] = {} |
| ids_by_name: dict[str, int] = {} |
| problems: list[ProblemNotice] = [] |
| files = glob.glob(f'{dir}/*.xml') |
| if not files: |
| raise SpecParsingException(f'No data model files found in specified directory {dir}') |
| |
| for xml in files: |
| logging.info(f'Parsing file {xml}') |
| tree = ElementTree.parse(f'{xml}') |
| root = tree.getroot() |
| add_cluster_data_from_xml(root, clusters, pure_base_clusters, ids_by_name, problems) |
| |
| # There are a few clusters where the conformance columns are listed as desc. These clusters need specific, targeted tests |
| # to properly assess conformance. Here, we list them as Optional to allow these for the general test. Targeted tests are described below. |
| # Descriptor - TagList feature - this feature is mandated when the duplicate condition holds for the endpoint. It is tested in DESC-2.2 |
| # Actions cluster - all commands - these need to be listed in the ActionsList attribute to be supported. |
| # We do not currently have a test for this. Please see https://github.com/CHIP-Specifications/chip-test-plans/issues/3646. |
| |
| def remove_problem(location: typing.Union[CommandPathLocation, FeaturePathLocation]): |
| nonlocal problems |
| problems = [p for p in problems if p.location != location] |
| |
| descriptor_id = Clusters.Descriptor.id |
| code = 'TAGLIST' |
| mask = clusters[descriptor_id].feature_map[code] |
| clusters[descriptor_id].features[mask].conformance = optional() |
| remove_problem(FeaturePathLocation(endpoint_id=0, cluster_id=descriptor_id, feature_code=code)) |
| action_id = Clusters.Actions.id |
| for c in Clusters.ClusterObjects.ALL_ACCEPTED_COMMANDS[action_id]: |
| clusters[action_id].accepted_commands[c].conformance = optional() |
| remove_problem(CommandPathLocation(endpoint_id=0, cluster_id=action_id, command_id=c)) |
| |
| combine_derived_clusters_with_base(clusters, pure_base_clusters, ids_by_name, problems) |
| |
| # TODO: All these fixups should be removed BEFORE SVE if at all possible |
| # Workaround for Color Control cluster - the spec uses a non-standard conformance. Set all to optional now, will need |
| # to implement either arithmetic conformance handling (once spec changes land here) or specific test |
| # https://github.com/CHIP-Specifications/connectedhomeip-spec/pull/7808 for spec changes. |
| # see 3.2.8. Defined Primaries Information Attribute Set, affects Primary<#>X/Y/Intensity attributes. |
| cc_id = Clusters.ColorControl.id |
| cc_attr = Clusters.ColorControl.Attributes |
| affected_attributes = [cc_attr.Primary1X, |
| cc_attr.Primary1Y, |
| cc_attr.Primary1Intensity, |
| cc_attr.Primary2X, |
| cc_attr.Primary2Y, |
| cc_attr.Primary2Intensity, |
| cc_attr.Primary3X, |
| cc_attr.Primary3Y, |
| cc_attr.Primary3Intensity, |
| cc_attr.Primary4X, |
| cc_attr.Primary4Y, |
| cc_attr.Primary4Intensity, |
| cc_attr.Primary5X, |
| cc_attr.Primary5Y, |
| cc_attr.Primary5Intensity, |
| cc_attr.Primary6X, |
| cc_attr.Primary6Y, |
| cc_attr.Primary6Intensity, |
| ] |
| for a in affected_attributes: |
| clusters[cc_id].attributes[a.attribute_id].conformance = optional() |
| |
| # Workaround for temp control cluster - this is parsed incorrectly in the DM XML and is missing all its attributes |
| # Remove this workaround when https://github.com/csa-data-model/projects/issues/330 is fixed |
| temp_control_id = Clusters.TemperatureControl.id |
| if temp_control_id in clusters and not clusters[temp_control_id].attributes: |
| view = Clusters.AccessControl.Enums.AccessControlEntryPrivilegeEnum.kView |
| none = Clusters.AccessControl.Enums.AccessControlEntryPrivilegeEnum.kUnknownEnumValue |
| clusters[temp_control_id].attributes = { |
| 0x00: XmlAttribute(name='TemperatureSetpoint', datatype='temperature', conformance=feature(0x01, 'TN'), read_access=view, write_access=none, write_optional=False), |
| 0x01: XmlAttribute(name='MinTemperature', datatype='temperature', conformance=feature(0x01, 'TN'), read_access=view, write_access=none, write_optional=False), |
| 0x02: XmlAttribute(name='MaxTemperature', datatype='temperature', conformance=feature(0x01, 'TN'), read_access=view, write_access=none, write_optional=False), |
| 0x03: XmlAttribute(name='Step', datatype='temperature', conformance=feature(0x04, 'STEP'), read_access=view, write_access=none, write_optional=False), |
| 0x04: XmlAttribute(name='SelectedTemperatureLevel', datatype='uint8', conformance=feature(0x02, 'TL'), read_access=view, write_access=none, write_optional=False), |
| 0x05: XmlAttribute(name='SupportedTemperatureLevels', datatype='list', conformance=feature(0x02, 'TL'), read_access=view, write_access=none, write_optional=False), |
| } |
| |
| check_clusters_for_unknown_commands(clusters, problems) |
| |
| return clusters, problems |
| |
| |
| def combine_derived_clusters_with_base(xml_clusters: dict[int, XmlCluster], pure_base_clusters: dict[str, XmlCluster], ids_by_name: dict[str, int], problems: list[ProblemNotice]) -> None: |
| ''' Overrides base elements with the derived cluster values for derived clusters. ''' |
| |
| def combine_attributes(base: dict[uint, XmlAttribute], derived: dict[uint, XmlAttribute], cluster_id: uint, problems: list[ProblemNotice]) -> dict[uint, XmlAttribute]: |
| ret = deepcopy(base) |
| extras = {k: v for k, v in derived.items() if k not in base.keys()} |
| overrides = {k: v for k, v in derived.items() if k in base.keys()} |
| ret.update(extras) |
| for id, override in overrides.items(): |
| if override.conformance: |
| ret[id].conformance = override.conformance |
| if override.read_access: |
| ret[id].read_access = override.read_access |
| if override.write_access: |
| ret[id].write_access = override.write_access |
| if ret[id].read_access is None and ret[id].write_access is None: |
| location = AttributePathLocation(endpoint_id=0, cluster_id=cluster_id, attribute_id=id) |
| problems.append(ProblemNotice(test_name='Spec XML parsing', location=location, |
| severity=ProblemSeverity.WARNING, problem='Unable to find access element')) |
| if ret[id].read_access is None: |
| ret[id].read_access == Clusters.AccessControl.Enums.AccessControlEntryPrivilegeEnum.kUnknownEnumValue |
| if ret[id].write_access is None: |
| ret[id].write_access = Clusters.AccessControl.Enums.AccessControlEntryPrivilegeEnum.kUnknownEnumValue |
| return ret |
| |
| # We have the information now about which clusters are derived, so we need to fix them up. Apply first the base cluster, |
| # then add the specific cluster overtop |
| for id, c in xml_clusters.items(): |
| if c.derived: |
| base_name = c.derived |
| if base_name in ids_by_name: |
| base = xml_clusters[ids_by_name[c.derived]] |
| else: |
| base = pure_base_clusters[base_name] |
| |
| feature_map = deepcopy(base.feature_map) |
| feature_map.update(c.feature_map) |
| attribute_map = deepcopy(base.attribute_map) |
| attribute_map.update(c.attribute_map) |
| command_map = deepcopy(base.command_map) |
| command_map.update(c.command_map) |
| features = deepcopy(base.features) |
| features.update(c.features) |
| attributes = combine_attributes(base.attributes, c.attributes, id, problems) |
| accepted_commands = deepcopy(base.accepted_commands) |
| accepted_commands.update(c.accepted_commands) |
| generated_commands = deepcopy(base.generated_commands) |
| generated_commands.update(c.generated_commands) |
| events = deepcopy(base.events) |
| events.update(c.events) |
| unknown_commands = deepcopy(base.unknown_commands) |
| for cmd in c.unknown_commands: |
| if cmd.id in accepted_commands.keys() and cmd.name == accepted_commands[cmd.id].name: |
| accepted_commands[cmd.id].conformance = cmd.conformance |
| elif cmd.id in generated_commands.keys() and cmd.name == generated_commands[cmd.id].name: |
| generated_commands[cmd.id].conformance = cmd.conformance |
| else: |
| unknown_commands.append(cmd) |
| |
| new = XmlCluster(revision=c.revision, derived=c.derived, name=c.name, |
| feature_map=feature_map, attribute_map=attribute_map, command_map=command_map, |
| features=features, attributes=attributes, accepted_commands=accepted_commands, |
| generated_commands=generated_commands, unknown_commands=unknown_commands, events=events, pics=c.pics) |
| xml_clusters[id] = new |
| |
| |
| def parse_single_device_type(root: ElementTree.Element) -> tuple[list[ProblemNotice], dict[int, XmlDeviceType]]: |
| problems: list[ProblemNotice] = [] |
| device_types: dict[int, XmlDeviceType] = {} |
| device = root.iter('deviceType') |
| for d in device: |
| name = d.attrib['name'] |
| location = DeviceTypePathLocation(device_type_id=0) |
| |
| str_id = d.attrib['id'] |
| if not str_id: |
| if name == "Base Device Type": |
| # Base is special device type, we're going to call it -1 so we can combine and remove it later. |
| str_id = '-1' |
| else: |
| problems.append(ProblemNotice("Parse Device Type XML", location=location, |
| severity=ProblemSeverity.WARNING, problem=f"Device type {name} does not have an ID listed")) |
| break |
| try: |
| id = int(str_id, 0) |
| revision = int(d.attrib['revision'], 0) |
| except ValueError: |
| problems.append(ProblemNotice("Parse Device Type XML", location=location, |
| severity=ProblemSeverity.WARNING, |
| problem=f"Device type {name} does not a valid ID or revision. ID: {str_id} revision: {d.get('revision', 'UNKNOWN')}")) |
| break |
| if id in DEVICE_TYPE_NAME_FIXES: |
| name = DEVICE_TYPE_NAME_FIXES[id] |
| try: |
| classification = next(d.iter('classification')) |
| scope = classification.attrib['scope'] |
| device_class = classification.attrib['class'] |
| except (KeyError, StopIteration): |
| # this is fine for base device type |
| if id == -1: |
| classification = 'BASE' |
| scope = 'BASE' |
| device_class = 'BASE' |
| else: |
| location = DeviceTypePathLocation(device_type_id=id) |
| problems.append(ProblemNotice("Parse Device Type XML", location=location, |
| severity=ProblemSeverity.WARNING, problem="Unable to find classification data for device type")) |
| break |
| device_types[id] = XmlDeviceType(name=name, revision=revision, server_clusters={}, client_clusters={}, |
| classification_class=device_class, classification_scope=scope) |
| clusters = d.iter('cluster') |
| for c in clusters: |
| try: |
| cid = int(c.attrib['id'], 0) |
| conformance_xml, tmp_problem = get_conformance(c, cid) |
| if tmp_problem: |
| problems.append(tmp_problem) |
| conformance = parse_device_type_callable_from_xml(conformance_xml) |
| side_dict = {'server': ClusterSide.SERVER, 'client': ClusterSide.CLIENT} |
| side = side_dict[c.attrib['side']] |
| name = c.attrib['name'] |
| if cid in CLUSTER_NAME_FIXES: |
| name = CLUSTER_NAME_FIXES[cid] |
| cluster = XmlDeviceTypeClusterRequirements(name=name, side=side, conformance=conformance) |
| if side == ClusterSide.SERVER: |
| device_types[id].server_clusters[cid] = cluster |
| else: |
| device_types[id].client_clusters[cid] = cluster |
| except ConformanceException: |
| location = DeviceTypePathLocation(device_type_id=id, cluster_id=cid) |
| problems.append(ProblemNotice("Parse Device Type XML", location=location, |
| severity=ProblemSeverity.WARNING, problem="Unable to parse conformance for cluster")) |
| # TODO: Check for features, attributes and commands as element requirements |
| # NOTE: Spec currently does a bad job of matching these exactly to the names and codes |
| # so this will need a bit of fancy handling here to get this right. |
| return device_types, problems |
| |
| |
| def build_xml_device_types(data_model_directory: typing.Union[PrebuiltDataModelDirectory, str] = PrebuiltDataModelDirectory.kInProgress) -> tuple[dict[int, XmlDeviceType], list[ProblemNotice]]: |
| dir = _get_data_model_directory(data_model_directory, DataModelLevel.kDeviceType) |
| device_types: dict[int, XmlDeviceType] = {} |
| problems = [] |
| for xml in glob.glob(f"{dir}/*.xml"): |
| logging.info(f'Parsing file {xml}') |
| tree = ElementTree.parse(f'{xml}') |
| root = tree.getroot() |
| tmp_device_types, tmp_problems = parse_single_device_type(root) |
| problems = problems + tmp_problems |
| device_types.update(tmp_device_types) |
| |
| if -1 not in device_types.keys(): |
| raise ConformanceException("Base device type not found in device type xml data") |
| |
| # Add in the base device type information and remove the base device type from the device_types |
| for d in device_types.values(): |
| d.server_clusters.update(device_types[-1].server_clusters) |
| device_types.pop(-1) |
| |
| return device_types, problems |