Source code for automate.program

# -*- coding: utf-8 -*-
# (c) 2015 Tuomas Airaksinen
# This file is part of Automate.
# Automate is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
# Automate is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# GNU General Public License for more details.
# You should have received a copy of the GNU General Public License
# along with Automate.  If not, see <>.
# ------------------------------------------------------------------
# If you like Automate, please take a look at this page:

import logging

from traits.api import cached_property, on_trait_change, CFloat, Instance, CBool, CSet, Property

from .common import (LogicStr, Lock, NameOrSensorActuatorBaseTrait,
from .systemobject import SystemObject
from .callable import AbstractCallable

[docs]class ProgrammableSystemObject(SystemObject): """ System object with standard program features (i.e. conditions & actions). """ simple_view = ['active'] view = ["name", "priority", "tags", "active_condition_str", "update_condition_str", "on_activate_str", "on_deactivate_str", "on_update_str", "triggers_str"] + SystemObject.view callables = ['on_activate', 'on_deactivate', 'on_update', 'active_condition', 'update_condition'] def get_default_callables(self): from automate.callables import Value, Empty return dict( active_condition=Value(True), update_condition=Value(True), on_activate=Empty(), on_update=Empty(), on_deactivate=Empty(), ) #: A condition Callable which determines the condition, when the program is activated. Program deactivates, when #: condition turns to False. When program is activated, on_activate action is executed. When program #: deactivates. on_deactivate is executed. active_condition = Instance(AbstractCallable) #: An action Callable to be executed when Program actives. on_activate = Instance(AbstractCallable) #: An action Callable to be executed when Program deactivates. on_deactivate = Instance(AbstractCallable) #: When program is active, this is the condition Callable that must equal to ``True`` in order to #: on_update action to be executed. Whenever a trigger is changed, this condition is checked and #: if ``True``, on_update is executed. update_condition = Instance(AbstractCallable) #: Action Callable to be executed if Program is active and update_condition is ``True``. on_update = Instance(AbstractCallable) #: When programs sets Actuator status, the actual status of Actuator is determined by a program that has highest #: priority. Lower priority programs are stacked and used only if higher priority programs are deactivated. priority = CFloat(1) #: Is program active? Automatically changed. In UIs you can fake the program active status by changing this. #: Normally do not change manually. active = CBool(False, transient=True) #: Status property is introduced to have interface compability with Status objects. #: For plain Programs, status equals to the result of its active condition Callable. status = Property(depends_on='active') @cached_property def _get_status(self): return _trigger_lock = Instance(Lock, transient=True) def __init__(self, *args, **kwargs): self._trigger_lock = Lock('triggerlock') super(ProgrammableSystemObject, self).__init__(*args, **kwargs) def __setstate__(self, *args, **kwargs): self._trigger_lock = Lock('triggerlock') return super(ProgrammableSystemObject, self).__setstate__(*args, **kwargs) #: (read-only property) Set of triggers, that cause this Program conditions to be checked #: (and actions to be executed). This data is updated from custom triggers list, conditions and actions. actual_triggers = Property(trait=CSet(trait=Instance(AbstractStatusObject)), depends_on='triggers, triggers_items, exclude_triggers, exclude_triggers_items, ' 'update_condition.triggers, active_condition.triggers, on_update.triggers, ' 'on_activate.triggers') #: (read-only property) Set of targets that this Program might touch. This data is updated #: from custom targets list and actions. actual_targets = Property(trait=CSet(trait=Instance(AbstractStatusObject)), depends_on='targets, targets_items, on_update.targets, on_activate.targets, ' 'on_deactivate.targets') #: Custom set of additional triggers, whose status change will trigger this Program conditions/actions triggers = CSet(trait=NameOrSensorActuatorBaseTrait) #: Triggers in this set do not trigger the program actions/conditions even if they are introduced by #: Callables etc. exclude_triggers = CSet(trait=NameOrSensorActuatorBaseTrait) #: Additional targets. Not usually needed, but if you want to set status for some reason by some custom function, #: for example, then you need to use this. targets = CSet(trait=NameOrSensorActuatorBaseTrait) logger = Instance(logging.Logger, transient=True) # Getter and setter used by _str versions of conditions, actions and targets&triggers lists. def _str_getter(self, name, trait): attr_name = name[:-4] attr = getattr(self, attr_name) if hasattr(attr, "give_str"): return attr.give_str() else: return str(attr) def _str_setter(self, name, value): attr_name = name[:-4] f = self.system.eval_in_system_namespace(value) if f is not None: setattr(self, attr_name, f) # For UIs, string versions of the Callables update_condition_str = Property(depends_on="update_condition,,", trait=LogicStr, transient=True, fset=_str_setter, fget=_str_getter) on_update_str = Property(depends_on="on_update,,", trait=LogicStr, transient=True, fset=_str_setter, fget=_str_getter) active_condition_str = Property(depends_on="active_condition,,", trait=LogicStr, transient=True, fset=_str_setter, fget=_str_getter) on_activate_str = Property(depends_on="on_activate,,", trait=LogicStr, transient=True, fset=_str_setter, fget=_str_getter) on_deactivate_str = Property(depends_on="on_deactivate,,", trait=LogicStr, transient=True, fset=_str_setter, fget=_str_getter) targets_str = Property(depends_on="targets,", trait=LogicStr, transient=True, fset=_str_setter, fget=_str_getter) triggers_str = Property(depends_on="triggers,", trait=LogicStr, transient=True, fset=_str_setter, fget=_str_getter) @cached_property def _get_actual_triggers(self): for c in [self.update_condition, self.active_condition, self.on_update, self.on_activate]: c.setup_callable_system(self.system) actual_triggers = (self.triggers | self.update_condition.triggers | self.active_condition.triggers | self.on_update.triggers | self.on_activate.triggers) - self.exclude_triggers self.logger.debug('_get_actual_triggers for %s gives %s', self, actual_triggers) return actual_triggers @cached_property def _get_actual_targets(self): for c in [self.on_update, self.on_activate, self.on_deactivate]: c.setup_callable_system(self.system) actual_targets = self.targets | self.on_update.targets | self.on_activate.targets | self.on_deactivate.targets self.logger.debug('_get_actual_targets for %s gives %s', self, actual_targets) return actual_targets @on_trait_change('actual_triggers') def actual_triggers_changed(self, obj, name, old, new): if old is None: old = set() if old == new: return self.logger.debug('Actual triggers changed by %s: %s->%s', name, old, new) for t in old - new: self.logger.debug("Removing trigger %s", t) t.on_trait_change(self.trigger_status_changed, "status", remove=True) for t in new - old: self.logger.debug("Adding trigger %s", t) t.on_trait_change(self.trigger_status_changed, "status") old_active = = bool( if != old_active: self.update_activation( @on_trait_change('actual_targets') def actual_targets_changed(self, obj, name, old, new): if old == new: return if old is None: old = set() self.logger.debug('Actual targets changed %s->%s', old, new) old_active = new_active = = bool( if new_active != old_active: self.update_activation(new_active) if old_active == new_active == True: for i in old - new: i.deactivate_program(self) for i in new - old: i.activate_program(self) def update_activation(self, new_active, trigger=None): if new_active: for t in self.actual_targets: t.activate_program(self) self.on_activate.setup_callable_system(self.system), trigger=trigger, action='activate') if bool(, trigger=trigger)): self.on_update.setup_callable_system(self.system) self.on_update.cancel(self), trigger=trigger, action='update') else: self.on_deactivate.setup_callable_system(self.system), trigger=trigger, action='deactivate') self.on_update.cancel(self) self.on_activate.cancel(self) for t in self.actual_targets: t.deactivate_program(self) def trigger_status_changed(self, obj, name, old, new): self.logger.debug("Trigger status changed from %s %s: %s->%s", obj, name, old, new) with self._trigger_lock: old_active = new_active = = bool(, trigger=obj)) if new_active != old_active: self.update_activation(new_active, trigger=obj) if old_active == new_active == True: if bool(, trigger=obj)): self.on_update.cancel(self), trigger=obj, action='update') self.logger.debug("Trigger status changing ready") @on_trait_change("active_condition, on_activate, on_deactivate") def _update_activation_actions(self, name, new): self.logger.debug('Update activation actions %s', name) getattr(self, name).setup_callable_system(self.system) if name == 'active_condition': old_active = = bool( if old_active != self.update_activation( elif name == 'on_activate' and self.on_activate.cancel(self), action='activate') @on_trait_change('update_condition, on_update') def _update_update_actions(self, name, new): self.logger.debug('Update update actions %s', name) getattr(self, name).setup_callable_system(self.system) if and bool( self.on_update.cancel(self), action='update') def setup_system(self, system, *args, **kwargs): from .callables import Value c = self.get_default_callables() c.pop('active_condition') # We do not want to activate program at this phase. c.pop('update_condition') # nor do we want to run on_update self.active_condition = Value(False) self.update_condition = Value(False) for key, value in list(c.items()): setattr(self, key, value) super(ProgrammableSystemObject, self).setup_system(system, *args, **kwargs) self.logger.debug('setup_system done for %s', self) def _priority_changed(self): self.logger.debug('Priority changed') for t in self.actual_targets: t.update_program_stack()
class Program(ProgrammableSystemObject): is_program = True editable = True def __lt__(self, other): return < class DefaultProgram(ProgrammableSystemObject): editable = True def __init__(self, system, name='', **traits): self._passed_arguments = name, traits self.setup_system(system) self.setup_callables() def get_default_callables(self): from automate.callables import Value callables = super(DefaultProgram, self).get_default_callables() callables['active_condition'] = Value(False) return callables