import logging
import warnings
import caspia.meadow.rules
from caspia.meadow.rules import OnDoRule, OnDoRuleTrigger, ThermostatRule
from caspia.reactive import FunctionObserver, Observable, Observer
logger = logging.getLogger(__name__)
[docs]class Automator:
def _get_rule_options(self, func):
kwargs = getattr(func, '_rule_options_', {})
setattr(func, '_rule_options_', kwargs)
return kwargs
[docs] def when(self, observable):
if not isinstance(observable, Observable):
raise ValueError('condition must be an Observable instance')
def decorator(func):
options = self._get_rule_options(func)
assert 'on' not in options
assert 'trigger' not in options
options['on'] = observable
options['trigger'] = OnDoRuleTrigger.ON_TRUE
return func
return decorator
[docs] def on_update(self, observable):
if not isinstance(observable, Observable):
raise ValueError('parameter must be an Observable instance')
def decorator(func):
options = self._get_rule_options(func)
assert 'on' not in options
assert 'trigger' not in options
options['on'] = observable
options['trigger'] = OnDoRuleTrigger.ON_UPDATE
return func
return decorator
[docs] def create_rule(self, name):
def decorator(func):
options = self._get_rule_options(func)
if 'on' not in options:
raise ValueError('%s is not a rule function' % func)
observer = FunctionObserver()
observer.subscribe_next(func)
rule = OnDoRule(on=options['on'],
do=observer,
trigger=options['trigger'],
identifier=name,
name=name)
caspia.meadow.rules.handler.rule_handler(rule)
func.rule = rule
return func
return decorator
# pylint: disable=unsubscriptable-object
[docs] def create_thermostat(self, name, temperature: Observable[float], heating: Observer[float],
params: dict):
rule = ThermostatRule(name, temperature=temperature, heating=heating, params=params)
caspia.meadow.rules.handler.rule_handler(rule)
return rule
[docs] def create_automation(self, name):
warnings.warn('create_automation is deprecated, please use create_rule', DeprecationWarning)
return self.create_rule(name)