Source code for configurationmodel.configdomain

"""
This module defines the objects that can appear in the Device
configuration object database.
"""

#
# Copyright Qualcomm Technologies Inc, 2019.
# All Rights Reserved
#

# Python import
import string
import json
from copy import deepcopy
from typing import Tuple, Callable, Any
import logging

# Thirdparty imports
from zope.component import getUtility

import transaction
import persistent
from persistent.list import PersistentList

# Imports from local packages
from qct_interfaces.state import ICurrentDatabaseModel
#from pyconfigtool.utils import trace

[docs]class ConfigDomain(persistent.Persistent): """This class defines the container of a set of configuration attributes that relevant to one domain. Each ``ConfigDomain`` contains attributes and child ``ConfigDomain`` objects. The UI usually represents a domain as a node in the tree of data, with an associated panel of individual data values, which are held in a :py:class:`ConfigDomainContainer` object. """ def __init__(self, name: str, description: str = '', key: str = None): self.name = name if key is None: self.key = _convert_name_to_key(name) else: self.key = key self.description_txt = description self.description_html = None # A list of child domains - references to ConfigDomain objects self.child_domains = PersistentList() self.config_container = ConfigDomainContainer() # type: ConfigDomainContainer self._parent = None @property def parent(self): """Returns the parent ConfigDomain object, or None if this is the root of the tree. """ return self._parent
[docs] def parents(self): """This generator method returns all the parents of the current ConfigDomain, starting with the root object and ending with the immediate parent of this instance. """ _p = self.parent if _p: yield from _p.parents() yield _p
[docs] def add_child_domain(self, child): """Add a ConfigDomain object to be the child of this one. The child's parent is updated to refer to this domain. """ self.child_domains.append(child) child.owned_by(self)
[docs] def owned_by(self, parent): """Take ownership of this ConfigDomain, which just sets the _parent attribute. """ self._parent = parent
[docs] def dump_domains(self, depth: int = 0) -> None: """Dump the ConfigDomain, useful for debugging. """ logging.debug( "%sDomain: %s %s %s %s %s", " " * depth, self.name, self.description_txt, self.description_html, self.key, self.config_container) if self.config_container: self.config_container.dump_container(depth) for domain in self.child_domains: domain.dump_domains(depth + 1)
[docs] def add_dummy_data(self) -> None: """Debug and test utility that adds a few dummy child ConfigDomain nodes to the current domain node, and then a transaction is committed. """ domain_1 = ConfigDomain('Hardware Configuration', 'some stuff goes here') domain_2 = ConfigDomain('Bluetooth Setting', 'more stuff goes here') domain_a = ConfigDomain('PIO Pins', 'who knows what goes here') self.add_child_domain(domain_1) self.add_child_domain(domain_2) domain_1.add_child_domain(domain_a) container = ConfigDomainContainer() container.add_attribute(ConfigAttributeCheckItem('enable_pios', 'Enable PIOs', 'Enable PIO sub-system')) domain_a.config_container = container transaction.commit()
[docs] def walk_domains(self, callback: Callable[[Any], None]) -> None: """This method will call the callback supplying this domain object first. Then it will call :py:meth:`ConfigContainer.walk_attributes` if this domain has a ``ConfigContainer``, and then finally call this method recursively on any child domains. """ callback(self) if self.config_container: self.config_container.walk_attributes(callback) for domain in self.child_domains: domain.walk_domains(callback)
[docs] def search_domains(self, callback: Callable[[Any], None]) -> None: """This method will walk all object under it calling the callback with a reference to that object. If the callback returns a non-``None`` result, the recursive walk is stopped, and that value is returned. """ result = callback(self) if result is not None: return result if self.config_container: result = self.config_container.search_attributes(callback) if result is not None: return result for domain in self.child_domains: result = domain.search_domains(callback) if result is not None: return result return None
[docs] def add_attribute(self, attr) -> None: """Add an attribute to the container. """ self.config_container.add_attribute(attr)
[docs]class ConfigAttributeBase(persistent.Persistent): # pylint: disable=too-many-instance-attributes """The base class for all single-value attribute elements. """ def __init__(self, key: str, name: str, # pylint: disable=too-many-arguments description: str, attr_type: str, attr_default: object = None, attr_value: object = None, view: str = 'default-view', **kwargs): if key is None: key = _convert_name_to_key(name) self.attr_key = key self.attr_name = name self.attr_type = attr_type self.attr_default = attr_default self.attr_value = attr_value self.attr_view = view self.attr_description = description # added in version 1.1 self.attr_visibility_spec = kwargs.get('attr_visibility_spec') # DependencyReferenceSpec self.attr_enablement_spec = kwargs.get('attr_enablement_spec') # DependencyReferenceSpec # added in 1.3 self.index = kwargs.get('index', None) # added in 1.4 self.attr_meta_data = kwargs.get('attr_meta_data', None) # added in 1.5 self.locked = kwargs.get('locked', False) # added in 1.6 self.css = kwargs.get('css', '')
[docs] def dump_attr(self, depth: int): """Debug utility. """ # Display each attribute expanding value lists if isinstance(self.attr_value, PersistentList): for idx, attr_value in enumerate(self.attr_value): logging.debug( '%s Attr: %s %s %s', ' ' * depth, self.attr_key + '.%d' % idx, self.attr_name, attr_value) else: logging.debug( '%s Attr: %s %s %s', ' ' * depth, self.attr_key, self.attr_name, self.attr_value)
[docs]class ConfigAttributeCheckItem(ConfigAttributeBase): """This data model item represents a boolean value, which is normally displayed as a checkbox item. """ def __init__(self, key: str, name: str, description: str, checked: bool = False, **kwargs): if 'attr_default' not in kwargs: kwargs['attr_default'] = checked if 'attr_value' not in kwargs: kwargs['attr_value'] = checked if 'view' not in kwargs: kwargs['view'] = 'qtcheckbox' super().__init__( key=key, name=name, description=description, attr_type='bool', **kwargs) @property def checked(self): """Return the value of the attribute, which should be a boolean. """ if self.index is not None: return self.attr_value[self.index] return self.attr_value @checked.setter def checked(self, val): """Set the value of the attribute, which should be a boolean. """ assert isinstance(val, bool), 'Expected a bool for a ConfigAttributeCheckItem' if self.index is not None: self.attr_value[self.index] = val else: self.attr_value = val
[docs]class ConfigAttributeIntegerItem(ConfigAttributeBase): """This data model item represents an integer value, which is normally displayed as a spinner box item. """ def __init__(self, key: str, name: str, description: str, value: int = 0, min_value=None, max_value=None, **kwargs): if 'attr_default' not in kwargs: kwargs['attr_default'] = value if 'attr_value' not in kwargs: kwargs['attr_value'] = value if 'view' not in kwargs: kwargs['view'] = 'qtspinbox' super().__init__( key=key, name=name, description=description, attr_type='int', **kwargs) self.min_value = min_value self.max_value = max_value self.wraps = kwargs.get("wraps", False) if 'hex_digits' in kwargs: self.hex_digits = kwargs['hex_digits'] @property def value(self): """Return the value of the attribute, which should be an integer. """ if self.index is not None: return self.attr_value[self.index] return self.attr_value @value.setter def value(self, val): """Set the value of the attribute, which should be a integer, and in the specified range, if there is one defined. """ assert isinstance(val, int), 'Expected an integer for a ConfigAttributeIntegerItem' if self.min_value is not None and val < self.min_value: raise ValueError("Attribute value is less than minimum") if self.max_value is not None and val > self.max_value: raise ValueError("Attribute value is greater than the maximum") if self.index is not None: self.attr_value[self.index] = val else: self.attr_value = val
[docs]class ConfigAttributeFloatItem(ConfigAttributeBase): """This data model item represents an integer value, which is normally displayed as a spinner box item. """ def __init__(self, key: str, name: str, # pylint: disable=too-many-arguments description: str, value: float = 0.0, precision: int = 1, min_value=None, max_value=None, **kwargs): if 'attr_default' not in kwargs: kwargs['attr_default'] = value if 'attr_value' not in kwargs: kwargs['attr_value'] = value if 'view' not in kwargs: kwargs['view'] = 'qtdoublespinbox' super().__init__( key=key, name=name, description=description, attr_type='float', **kwargs) self.min_value = min_value self.max_value = max_value self.precision = precision self.wraps = kwargs.get("wraps", False) @property def value(self): """Return the value of the attribute, which should be an integer. """ if self.index is not None: return self.attr_value[self.index] return self.attr_value @value.setter def value(self, val): """Set the value of the attribute, which should be a float, and in the specified range, if there is one defined. """ assert isinstance(val, float), 'Expected a float for a ConfigAttributeFloatItem' if self.min_value is not None and val < self.min_value: raise ValueError("Attribute value is less than minimum") if self.max_value is not None and val > self.max_value: raise ValueError("Attribute value is greater that the maximum") if self.index is not None: self.attr_value[self.index] = val else: self.attr_value = val
[docs]class ConfigAttributeComboboxItem(ConfigAttributeBase): """This data model item represents a single choice from a specified list of options. The options should be expressed as a tuple, where each option is a two-tuple: a key, followed by the visible string that is presented in the user interface for that option. The value of the currently-selected option is the corresponding **key**, not the value that is displayed in the user interface. """ def __init__(self, key: str, name: str, description: str, value: str, choices: Tuple[Tuple[str]], **kwargs): if 'attr_default' not in kwargs: kwargs['attr_default'] = value if 'attr_value' not in kwargs: kwargs['attr_value'] = value if 'view' not in kwargs: kwargs['view'] = 'qtcombobox' super().__init__( key=key, name=name, description=description, attr_type='combo', **kwargs) if isinstance(value, PersistentList): for val in value: if val not in [key for key, _ in choices]: logging.error( "Current value '%s' of '%s' not in the list of options %s", val, key, [key for key, _ in choices]) else: if value not in [key for key, _ in choices]: logging.error( "Current value '%s' of '%s' not in the list of options %s", value, key, [key for key, _ in choices]) self.attr_choices = choices @property def value(self): """Return the value of the attribute, which should be one of the choice keys. """ if self.index is not None: return self.attr_value[self.index] return self.attr_value @value.setter def value(self, val): """Return the value of the attribute, which should be one of the choice keys. """ if self.index is not None: self.attr_value[self.index] = val else: self.attr_value = val
[docs] def set_index(self, idx): """Set the value of this attrbute to the ``idx``-th one in the list of choices for the combobox. """ assert 0 <= idx < len(self.attr_choices) if self.index is not None: self.attr_value[self.index] = self.attr_choices[idx][0] else: self.attr_value = self.attr_choices[idx][0]
[docs]class ConfigAttributeTextItem(ConfigAttributeBase): """This data model item represents a single unicode string which is usually displayed as a single line editor box. """ def __init__(self, key: str, name: str, description: str, value: str, **kwargs): if 'attr_default' not in kwargs: kwargs['attr_default'] = value if 'attr_value' not in kwargs: kwargs['attr_value'] = value if 'view' not in kwargs: kwargs['view'] = 'qtlineedit' super().__init__( key=key, name=name, description=description, attr_type='text', **kwargs) @property def value(self): """Return the value of the attribute, which should be the input text string. """ if self.index is not None: return self.attr_value[self.index] return self.attr_value @value.setter def value(self, val): """Set the value of the attribute, which should be a float, and in the specified range, if there is one defined. """ assert isinstance(val, str), 'Expected a string for a ConfigAttributeTextItem' if self.index is not None: self.attr_value[self.index] = val else: self.attr_value = val
[docs]class ConfigAttributeTableItem(ConfigAttributeBase): """This data model item represents a table. Tables contain a list of other ConfigAttributes which have been initialized with PersistentList type values. Each column displays a different ConfigAttribute and each row is populated with the indexed element from the PersistentList held by the ConfigAttributes value property. """ def __init__(self, key: str, name: str, description: str, value: PersistentList, **kwargs): if 'view' not in kwargs: kwargs['view'] = 'qttable' if 'attr_value' not in kwargs: kwargs['attr_value'] = value super().__init__( key=key, name=name, description=description, attr_type='table', **kwargs) @property def table(self): """Return the value of the attribute, which should be the list of other attributes. """ if self.index is not None: return self.attr_value[self.index] return self.attr_value @table.setter def table(self, val): """Set the value of the attribute, which should be a table. """ assert isinstance(val, TableDomain),\ 'Expected a TableDomain for ConfigAttributeTableItem' if self.index is not None: self.attr_value[self.index] = val else: self.attr_value = val
[docs] def dump_attr(self, depth: int): """Debug utility. """ # Display the table view attribute logging.debug( '%s Attr: %s %s %s', ' ' * depth, self.attr_key, self.attr_name, self.attr_value) self.attr_value.dump_domains(depth + 2)
[docs]class TableDomain(ConfigDomain): """This class defines the container of a set of configuration attributes that are relevant to one table. Each ``TableDomain`` contains attributes. The UI represents a table domain as a table of data values, which are held in a :py:class:`QTableWidget` object. """ def __init__(self, name: str, description: str = '', key: str = None): super().__init__(name, description, key) self.table_attributes = PersistentList()
[docs] def add_attribute(self, attr: ConfigAttributeBase) -> None: """Add an attribute to the container. """ if not isinstance(attr.attr_value, PersistentList): attr.index = 0 assert attr.attr_value attr.attr_value = PersistentList([attr.attr_value]) self.table_attributes.append(attr) self.extend_attributes()
[docs] def extend_attributes(self, minimum=0): """ Extend all table attributes to the length of the longest attribute or the ``minimum`` :param minimum: Minimum number of entries for each data item :return: None """ # Extend attribute value lists if needed to fill table table_rows = max(map(len, [attr.attr_value for attr in self.table_attributes]), default=0) table_rows = max([table_rows, minimum]) for table_attribute in self.table_attributes: attr_rows = len(table_attribute.attr_value) if attr_rows < table_rows: table_attribute.attr_value = table_attribute.attr_value + \ [table_attribute.attr_default] * (table_rows - attr_rows)
[docs] def walk_attributes(self, callback: Callable[[Any], None]) -> None: """This method will first call the callback with this container instance as the only argument, and then it will call the callback for each of the attributes that are contained in this container. """ callback(self) for attr in self.table_attributes: callback(attr)
[docs] def search_attributes(self, callback: Callable[[Any], Any]) -> Any: """This method will first call the callback with this container instance as the only argument. If a non-``None`` result is returned, this this result is returned immediately. If not then it will then call the callback for each of the attributes that are contained in this container, stoping whenever the callback returns a non-``None`` value. """ result = callback(self) if result is not None: return result for attr in self.table_attributes: result = callback(attr) if result is not None: return result return None
[docs] def add_child_domain(self, child): """Tables cannot have subdomains. """ raise NotImplementedError
[docs] def dump_domains(self, depth: int = 0) -> None: """Dump the TableDomain, useful for debugging. """ logging.debug( "%sTableDomain: %s %s %s %s", " " * depth, self.name, self.description_txt, self.description_html, self.key) for attr in self.table_attributes: attr.dump_attr(depth)
[docs] def add_dummy_data(self) -> None: """Debug and test utility that adds some dummy child attributes to the current table domain, and then a transaction is committed. """ self.add_attribute( ConfigAttributeComboboxItem( 'pio', 'PIO', 'PIO sub-system', PersistentList(['PIO-1', 'PIO-2', 'PIO-3', 'PIO-4']), (('PIO-1', '1'), ('PIO-2', '2'), ('PIO-3', '3'), ('PIO-4', '4')))) self.add_attribute( ConfigAttributeCheckItem( 'enable_pios', 'Enable PIOs', 'Enable PIO sub-system', PersistentList([True, False, True, False]))) transaction.commit()
[docs] def search_domains(self, callback: Callable[[Any], None]) -> None: """This method will walk all object under it calling the callback with a reference to that object. If the callback returns a non-``None`` result, the recursive walk is stopped, and that value is returned. """ result = callback(self) if result is not None: return result result = self.search_attributes(callback) if result is not None: return result return None
[docs]class ConfigAttributeStruct(ConfigAttributeBase): """This data model item represents a table. Tables contain a list of other ConfigAttributes which have been initialized with PersistentList type values. Each column displays a different ConfigAttribute and each row is populated with the indexed element from the PersistentList held by the ConfigAttributes value property. """ def __init__(self, key: str, name: str, description: str, value: PersistentList, **kwargs): if 'view' not in kwargs: kwargs['view'] = 'qtstruct' if 'attr_value' not in kwargs: kwargs['attr_value'] = value super().__init__( key=key, name=name, description=description, attr_type='struct', **kwargs) @property def domain(self): """Return the value of the attribute, which should be the list of structure instances. """ return self.attr_value @domain.setter def domain(self, val): """Set the value of the attribute, which should be a list of ConfigDomains. """ self.attr_value = val
[docs] def dump_attr(self, depth: int): """Debug utility. """ # Display the struct view attribute logging.debug( '%s Attr: %s %s %s', ' ' * depth, self.attr_key, self.attr_name, self.attr_value) self.attr_value.dump_domains(depth + 2)
[docs]class StructDomain(ConfigDomain): """``StructDomain`` defines a set of configuration structures each containing lists of attributes that are relevant to a single struct instance. The UI typically represents a struct domain as a tabbed set of data values, which are held in a :py:class:`QTabWidget` object. """ def __init__(self, name: str, description: str = '', key: str = None): super().__init__(name, description, key) self.structs = PersistentList() self.config_container = None self.index = 0
[docs] def add_struct(self) -> object: """Add a ConfigDomainContainer to the list. """ container = ConfigDomainContainer() self.structs.append(container) return container
[docs] def clone_struct(self, index) -> None: """Add a copy of a ConfigDomainContainer to the list. """ container = deepcopy(self.structs[index]) self.structs.insert(index, container)
[docs] def delete_struct(self, index) -> None: """Add a ConfigDomainContainer to the list. """ self.structs.pop(index)
[docs] def add_attribute(self, attr: ConfigAttributeBase) -> None: """Add an attribute to the container. """ self.structs[self.index].add_attribute(attr)
[docs] def walk_attributes(self, callback: Callable[[Any], None]) -> None: """This method will first call the callback with this container instance as the only argument, and then it will call the callback for each of the attributes that are contained in this container. """ callback(self) for struct in self.structs.data: struct.walk_attributes(callback)
[docs] def search_attributes(self, callback: Callable[[Any], Any]) -> Any: """This method will first call the callback with this container instance as the only argument. If a non-``None`` result is returned, this this result is returned immediately. If not then it will then call the callback for each of the attributes that are contained in this container, stoping whenever the callback returns a non-``None`` value. """ result = callback(self) if result is not None: return result for struct in self.structs: result = struct.search_attributes(callback) if result is not None: return result return None
[docs] def add_child_domain(self, child): """Structs cannot have subdomains. """ raise NotImplementedError
[docs] def dump_domains(self, depth: int = 0) -> None: """Dump the StructDomain, useful for debugging. """ logging.debug( "%sStructDomain: %s %s %s %s", " " * depth, self.name, self.description_txt, self.description_html, self.key) for struct in self.structs: for attr in struct.config_attributes: attr.dump_attr(depth)
[docs] def search_domains(self, callback: Callable[[Any], None]) -> None: """This method will walk all object under it calling the callback with a reference to that object. If the callback returns a non-``None`` result, the recursive walk is stopped, and that value is returned. """ result = callback(self) if result is not None: return result return self.search_attributes(callback)
[docs] def add_dummy_data(self) -> None: """Debug and test utility that adds some dummy child attributes to the structure domain, clones it and then a transaction is committed. """ self.key = 'struct' container = self.add_struct() container.add_attribute( ConfigAttributeComboboxItem( 'struct.0.pio', 'PIO', 'PIO sub-system', 'PIO-4', (('PIO-1', '1'), ('PIO-2', '2'), ('PIO-3', '3'), ('PIO-4', '4')))) container.add_attribute( ConfigAttributeCheckItem( 'struct.0.enable_pio', 'Enable PIOs', 'Enable PIO sub-system', True)) self.clone_struct(0) transaction.commit()
[docs]class ConfigDomainContainer(persistent.Persistent): """This container object contains all the attributes that should be displayed on the right-hand-side pane when the tree item to which this container is attached is clicked on. """ def __init__(self): # A list of ConfigAttribute-derived objects self.config_attributes = PersistentList() self.config_view = 'data-driven'
[docs] def add_attribute(self, attr: ConfigAttributeBase) -> None: """Add an attribute to the container. """ self.config_attributes.append(attr)
[docs] def dump_container(self, depth: int) -> None: """Debug utility method. """ for attr in self.config_attributes: attr.dump_attr(depth)
[docs] def walk_attributes(self, callback: Callable[[Any], None]) -> None: """This method will first call the callback with this container instance as the only argument, and then it will call the callback for each of the attributes that are contained in this container. """ callback(self) for attr in self.config_attributes: callback(attr) if isinstance(attr, (ConfigAttributeTableItem, ConfigAttributeStruct)): attr.attr_value.walk_attributes(callback)
[docs] def search_attributes(self, callback: Callable[[Any], Any]) -> Any: """This method will first call the callback with this container instance as the only argument. If a non-``None`` result is returned, this this result is returned immediately. If not then it will then call the callback for each of the attributes that are contained in this container, stoping whenever the callback returns a non-``None`` value. """ result = callback(self) if result is not None: return result for attr in self.config_attributes: result = callback(attr) if result is not None: return result if hasattr(attr, 'domain'): return attr.domain.search_domains(callback) if hasattr(attr, 'table'): return attr.table.search_domains(callback) return None
[docs]class DependencyReferenceSpec(persistent.Persistent): """The class describes a dependency between one attribute and the current value of some other attribute in the database. For example, it is used to define a link between the enablement of one attribute with the value of some other check box. """ def __init__(self, depends_on: str, condition_test: object = None, invert: bool = False): self.depends_on = depends_on self.condition_test = condition_test self.invert = invert @property def depends_on_attribute(self): """Lazy evaluation to find the attribute to which this object describes a dependency on. This read-only property searches the database for an attribute of a particular key, and returns that. Given that this could be an expensive operation, the attibute that is found is cached in the this object. In order that this cached value is not persisted, its name has to start with ``_v_``, which is the indication to the ZODB persistence layer that this is a volatile attribute. """ if hasattr(self, '_v_cached_attr'): return self._v_cached_attr database = getUtility(ICurrentDatabaseModel).value attr = database.find_attr_by_key(self.depends_on) setattr(self, '_v_cached_attr', attr) return attr
[docs] def check(self) -> bool: """Perform the dependency check on the referenced attribute. The sense of the test is inverted if the ``invert=True`` was passed into the constructor. """ attr = self.depends_on_attribute if not attr: return True if self.invert: if self.condition_test: return attr.attr_value != self.condition_test return not bool(attr.attr_value) if self.condition_test: return attr.attr_value == self.condition_test return bool(attr.attr_value)
_ALLOWED_CHARS = string.ascii_letters + '0123456789_-' def _convert_name_to_key(name: str) -> str: """Utility to convert the human-readable name of a ConfigDomain object to a key that can be used as attributes in XML and dictionary keys in YAML. """ key = name.replace(' ', '_') key = key.lower() return ''.join([c for c in key if c in _ALLOWED_CHARS])
[docs]def get_metadata(config_item, key=None): """Return a python object containing the contents of the json decoded metadata associated with a config item. If there is no metadata or the metadata is empty or the metadata of the requested key is empty, or the metadata is just "default" then return a python list ``["default"]`` Optionally a metadata key name can be provided to acess a top-level key. e.g. writers = get_attribute_metadata(attr, "writers") on attr.attr_meta_data = '{"writers":["this","that"]}' returns the python list ["this", "that"] """ default = [] if isinstance(config_item, ConfigAttributeBase): default = ["default"] try: if not config_item.attr_meta_data: return default metadata = json.loads(config_item.attr_meta_data) if not metadata or metadata == "default": return default if key: return metadata.get(key, default) except (ValueError, SyntaxError) as err: logging.error( "Error reading metadata in attribute %s\n%s\n%s", config_item.attr_key, config_item.attr_meta_data, err) return default