# -*- coding: utf-8 -*-
# -----------------------------------------------------------------------------
# Copyright 2015-2018 by Exopy Authors, see AUTHORS for more details.
#
# Distributed under the terms of the BSD license.
#
# The full license is in the file LICENCE, distributed with this software.
# -----------------------------------------------------------------------------
"""Plugin managing the preferences of the TextMonitor such as rules.
"""
import logging
from atom.api import List, Dict, Typed
from ....utils.plugin_tools import (DeclaratorsCollector, ExtensionsCollector,
make_extension_validator,
HasPreferencesPlugin)
from ..base_monitor import Monitor
from .monitor import TextMonitor
from .rules.base import RuleType, Rules, RuleConfig
RULE_TYPE_POINT = 'exopy.measurement.monitors.text_monitor.rules.type'
RULE_CONFIG_POINT = 'exopy.measurement.monitors.text_monitor.rules.config'
logger = logging.getLogger(__name__)
[docs]class TextMonitorPlugin(HasPreferencesPlugin):
"""Plugin managing the preferences of the TextMonitor.
"""
# List of rules which should be created automatically for new monitors.
default_rules = List(default=['Measurement entries', 'Loop progress',
'Instrument ids']).tag(pref=True)
# List of available rule types.
rule_types = List()
# List of available rules configurations.
rules = List()
[docs] def start(self):
"""Start the plugin life-cycle.
"""
super(TextMonitorPlugin, self).start()
self._rule_types = DeclaratorsCollector(workbench=self.workbench,
point=RULE_TYPE_POINT,
ext_class=(Rules, RuleType))
self._rule_types.start()
validator = make_extension_validator(RuleConfig,
attributes=('id', 'description',
'rule_type', 'config')
)
self._rule_configs = ExtensionsCollector(workbench=self.workbench,
point=RULE_CONFIG_POINT,
ext_class=RuleConfig,
validate_ext=validator)
self._rule_configs.start()
# List all the rule types and rules and remove unknown rules from
# the default ones.
self._update_rule_types(None)
self._update_rules(None)
defaults = [r for r in self.default_rules if r in self.rules]
if defaults != self.default_rules:
msg = ('The following rules for the TextMonitor are not defined, '
'and have been removed from the defaults : %s')
removed = set(self.default_rules) - set(defaults)
logger.warning(msg, removed)
self.default_rules = defaults
self._bind_observers()
[docs] def stop(self):
"""Stop the plugin and clear all ressources.
"""
self._unbind_observers()
self.rule_types = []
self.rules = []
self._rule_types.stop()
self._rule_configs.stop()
[docs] def build_rule(self, name_or_config):
""" Build rule from a dict.
Parameters
----------
name_or_config : unicode|dict
Name of the rule to build or dict containing the infos to build the
rule from scratch.
Returns
-------
rule : BaseRule | None
New rule properly initialized.
"""
if not isinstance(name_or_config, dict):
if name_or_config in self._user_rules:
config = self._user_rules[name_or_config].copy()
config['id'] = name_or_config
elif name_or_config in self._rule_configs.contributions:
rule_config = self._rule_configs.contributions[name_or_config]
config = rule_config.config.copy()
config['class_id'] = rule_config.rule_type
config['description'] = rule_config.description
config['id'] = name_or_config
else:
msg = 'Requested rule not found : {}'.format(name_or_config)
logger.warning(msg)
return
else:
config = name_or_config.copy()
class_id = config.pop('class_id')
rule_infos = self._rule_types.contributions.get(class_id)
if rule_infos is not None:
rule = rule_infos.cls()
rule.update_members_from_preferences(config)
return rule
else:
msg = 'Requested rule class not found : {}'.format(class_id)
logger.warning(msg)
[docs] def get_rule_type(self, rule_type_id):
"""Access the class corresponding to a given id.
"""
return self._rule_types.contributions[rule_type_id].cls
[docs] def get_rule_view(self, rule):
"""CReate a view corresponding to the given object.
"""
infos = self._rule_types.contributions[rule.class_id]
is_user = rule.id not in self._rule_configs.contributions
return infos.view(rule=rule, plugin=self,
enabled=is_user)
[docs] def save_rule(self, rule):
"""Add a rule present on a plugin to the saved rules.
"""
self._user_rules[rule.id] = rule.preferences_from_members()
self._update_rules(None)
[docs] def create_monitor(self, default=False):
""" Create a new monitor.
Parameters
----------
default : bool, optionnal
Whether or not to add the default rules to the new monitor.
Returns
-------
monitor : TextMonitor
New text monitor.
"""
exts = [e for e in self.manifest.extensions if e.id == 'monitors']
decl = exts[0].get_child(Monitor)
monitor = TextMonitor(_plugin=self,
declaration=decl)
if default:
rules = []
for rule_name in self.default_rules:
rule = self.build_rule(rule_name)
if rule:
rules.append(rule)
monitor.rules = rules
return monitor
# =========================================================================
# --- Private API ---------------------------------------------------------
# =========================================================================
#: Collect the rule types contributions.
_rule_types = Typed(DeclaratorsCollector)
#: Collect the rule config declarations and use them to update the config
_rule_configs = Typed(ExtensionsCollector)
#: User defined rules config saved in the preferences.
#: TODO This should be saved as sperate files in a folder under the
#: application folder.
_user_rules = Dict().tag(pref=True)
def _update_rule_types(self, change):
"""Update the public rule types class id when new ones get registered.
"""
self.rule_types = list(self._rule_types.contributions)
def _update_rules(self, change):
"""Update the rule names whenever a new contributed rule or a new user
rule is added.
"""
contrib = set(self._rule_configs.contributions)
users = set(self._user_rules)
self.rules = list(contrib | users)
def _bind_observers(self):
"""Observe the collectors to update public attributes.
"""
self._rule_configs.observe('contributions', self._update_rules)
self._rule_types.observe('contributions', self._update_rule_types)
def _unbind_observers(self):
"""Unobserve the collectors.
"""
self._rule_configs.unobserve('contributions', self._update_rules)
self._rule_types.unobserve('contributions', self._update_rule_types)