Scheduler and Triggers
The system has a scheduler.
The system scheduler has just one job: it notifies Events gathered from a triggered Scheduler Trigger to the specified Performers. The Protocol Commands inside the specified Performers will be executed just after the Events are being notified to their Appliances.
Hint
Scheduler Triggers are not Protocol Triggers.
But, both Scheduler Triggers and Protocol Triggers have Events to be notified.
Scheduler Triggers can use Protocol Triggers to define themselves.
Scheduler Trigger vs Protocol Triggers
Time based triggers
You need to use a Scheduler Trigger instead of a Protocol Trigger when the trigger logic is time based.
Suppose you need a mean for a lux value and suppose your device is sending lux data every time the lux is changing.
You can calculate a mean using a protocol trigger: this trigger will be highly sensible to lux changes.
Or you can calculate a time based mean using a scheduler trigger: this trigger will be less sensible to lux changes.
Reusable triggers
If you want to define a trigger once and reuse it with multiple performers then you need a Scheduler Trigger.
A trigger associated with a lux sensor is probably useful for multiple Appliances. All the project’s curtains and lights models may need to know the sun brightness level.
Instead of creating one protocol trigger in a performer for every curtain and light, a single scheduler trigger using a performer with just one protocol trigger can be scheduled.
State triggers
If you need a trigger for the Appliance state you need a Scheduler Trigger.
If you need to know if the Appliance has entered a new state or exited an old state then you need a schedule trigger.
Scheduler (Class Diagram)
Scheduler (Sequence Diagram)
Trigger Class
- class home.scheduler.Trigger(name: str, events: Iterable[home.Event], *args, **kwargs)
A Scheduler Trigger.
When triggered, given events are notified to scheduled Performers.
cron.Trigger
- class home.scheduler.trigger.cron.Trigger(name: str, events: List[home.Event], *args, **kwargs)
An extension of the APScheduler CronTrigger.
Example
The following is a Scheduler Trigger triggered from Monday to Friday at 7:15 am.
It has one event: home.event.sleepiness.Event.Awake. If notified to a curtain Appliance will probably raise the curtain:
!cron.Trigger
name: "wakeup time scheduler trigger"
notify events:
- !home.event.sleepiness.Event.Awake
day_of_week: "mon-fri"
hour: 7
minute: 15
The Scheduler Trigger is scheduled with the specified Performers through this syntax:
!schedule
trigger: "wakeup time scheduler trigger"
for performers: "curtain command performer"
The Performer is defined like this:
!Performer
name: "curtain command performer"
for appliance: "a curtain appliance"
commands:
- !knx_plugin.command.dpt_updown.UpDown {addresses: [0x1111,]}
triggers: []
If needed, a list of Performers can be created:
curtain commands:
- "curtain command performer"
- "another curtain command performer"
And could be scheduled with the Scheduler Trigger:
!schedule
trigger: "wakeup time scheduler trigger"
for performers: "curtain commands"
interval.Trigger
- class home.scheduler.trigger.interval.Trigger(name: str, events: Iterable[home.Event], *args, **kwargs)
An extension of the APScheduler IntervalTrigger.
Example
This Scheduler Trigger can be used, for example, to poll a device state every specified interval of time.
To continuously poll a device, the Protocol Command used by the Performer, should, probably, ignore the Appliance State and always build a command Description for the Protocol Gateway to poll the device state.
date.Trigger
- class home.scheduler.trigger.date.Trigger(name: str, events: Iterable[home.Event], *args, **kwargs)
An extension of the APScheduler DateTrigger.
sun.sunrise.Trigger
- class home.scheduler.trigger.sun.sunrise.Trigger(name: str, events: Iterable[home.Event], latitude: float, longitude: float, elevation: int)
A Trigger triggered by the sunrise.
When triggered will notify a home.event.sun.phase.Event.Sunrise plus other given events.
Example
!sun.sunrise.Trigger
name: "sunrise scheduler trigger"
notify more events: []
latitude: 44.00
longitude: 12.00
elevation: 280
sun.twilight.civil.sunrise.Trigger
- class home.scheduler.trigger.sun.twilight.civil.sunrise.Trigger(name: str, events: Iterable[home.Event], latitude: float, longitude: float, elevation: int)
A Trigger triggered when starts the morning civil twilight
When triggered will notify a home.event.sun.twilight.civil.Event.Sunrise plus other given events.
Example
!sun.twilight.civil.Trigger
name: "start of morning twilight scheduler trigger"
notify more events: []
latitude: 44.00
longitude: 12.00
elevation: 280
sun.sunset.Trigger
- class home.scheduler.trigger.sun.sunset.Trigger(name: str, events: Iterable[home.Event], latitude: float, longitude: float, elevation: int)
A Trigger triggered by the sunset.
When triggered will notify a home.event.sun.phase.Event.Sunset plus other given events.
Example
!sun.sunset.Trigger
name: "sunset scheduler trigger"
notify more events: []
latitude: 44.00
longitude: 12.00
elevation: 280
sun.twilight.civil.sunset.Trigger
- class home.scheduler.trigger.sun.twilight.civil.sunset.Trigger(name: str, events: Iterable[home.Event], latitude: float, longitude: float, elevation: int)
A Trigger triggered when ends the evening civil twilight
When triggered will notify a home.event.sun.twilight.civil.Event.Sunset plus other given events.
Example
!sun.twilight.civil.sunset.Trigger
name: "end of civil twilight scheduler trigger"
notify more events: []
latitude: 44.00
longitude: 12.00
elevation: 280
sun.sunhit.Trigger
- class home.scheduler.trigger.sun.sunhit.Trigger(name: str, events: Iterable[home.Event], latitude: float, longitude: float, elevation: int, position: home.scheduler.trigger.sun.Position)
Triggered when an object, with the given Position, is hit by the sun.
When triggered will notify a home.event.sun.hit.Event.Sunhit plus other given events.
Example
This trigger could be used, when you do not have an internal lux sensor, to approximate the time of day when the sun is entering from a window:
!sun.sunhit.Trigger
name: "sun hit east exposed window"
notify more events: []
latitude: 45
longitude: 12
elevation: 800
position: !Position {bottom_altitude: 10, upper_altitude: 90, min_azimuth: 45, max_azimuth: 135}
!sun.sunhit.Trigger
name: "sun hit sud exposed window"
notify more events: []
latitude: 45
longitude: 12
elevation: 800
position: !Position {bottom_altitude: 10, upper_altitude: 90, min_azimuth: 135, max_azimuth: 225}
!sun.sunhit.Trigger
name: "sun hit west exposed window"
notify more events: []
latitude: 45
longitude: 12
elevation: 800
position: !Position {bottom_altitude: 10, upper_altitude: 90, min_azimuth: 225, max_azimuth: 315}
sun.sunleft.Trigger
- class home.scheduler.trigger.sun.sunleft.Trigger(name: str, events: Iterable[home.Event], latitude: float, longitude: float, elevation: int, position: home.scheduler.trigger.sun.Position)
Triggered when an object, with the given Position, is no more hit by the sun.
When triggered will notify a home.event.sun.hit.Event.Sunleft plus other given events.
Example
This trigger should probably be used every time you use a sun.sunhit.Trigger.
Its definition is like the following:
!sun.sunleft.Trigger
name: "sun left east exposed window"
notify more events: []
latitude: 45
longitude: 12
elevation: 800
position: !Position {bottom_altitude: 10, upper_altitude: 90, min_azimuth: 20, max_azimuth: 145}
protocol.Trigger
- class home.scheduler.trigger.protocol.Trigger(name: str, events: Iterable[home.Events], protocol_trigger: home.protocol.Trigger, *args, **kwargs)
A Scheduler Trigger triggered when its Protocol Trigger is triggered.
When triggered will notify given events plus the protocol trigger events.
Example
An example showing how curtain Appliances can be notified of a strong wind Event through a KNX message.
The Protocol Trigger is bound with the anemometer Appliance through the following Performer:
!Performer
name: "strong wind performer"
for appliance: "anemometer"
commands: []
triggers:
- !knx_plugin.trigger.dpt_value_wsp.Strong {addresses: [0xAAAA,], events: [home.event.wind.Event.Strong,]}
A Protocol Scheduler Trigger is built for the Protocol Trigger found in the above Performer:
!protocol.Trigger
name: "strong wind scheduler trigger"
notify more events: []
when triggered performers: "strong wind performer"
The Scheduler Trigger is scheduled to notify events to the curtains:
!schedule
trigger: "strong wind performer"
for performers: "curtain commands"
protocol.delay.Trigger
- class home.scheduler.trigger.protocol.delay.Trigger(name: str, events: Iterable[home.Event], protocol_trigger: home.protocol.Trigger, timeout_seconds: float)
A Scheduler Trigger triggered timeout_seconds after its Protocol Trigger has been triggered. If the Protocol Trigger has been triggered twice, the old scheduler trigger is disabled and new one is started.
Example
An example showing how a Light can be potentially turned Off seconds after a movement sensor is no more triggered
The Protocol Trigger is bound with the movement sensor Appliance through the following Performer:
!Performer
name: "courtesy off performer"
for appliance: "movement sensor"
commands: []
triggers:
- !knx_plugin.trigger.dpt_switch.Off {addresses: [0xBBBB,], events: [home.event.courtesy.Event.Off,]}
A Protocol Scheduler Trigger is built for the Protocol Trigger found in the above Performer:
!protocol.delay.Trigger
name: "courtesy off scheduler trigger"
notify more events: []
when triggered performers: "courtesy off performer"
and timeout expires: 180
The Scheduler Trigger is scheduled to notify events to the lights:
!schedule
trigger: "courtesy off scheduler trigger"
for performers: "lights commands"
protocol.enum.Trigger
- class home.scheduler.trigger.protocol.enum.Trigger(name: str, events: Iterable[home.Event], selected: home.enum.definition.Event, direction: str, protocol_trigger: home.protocol.Trigger)
A Scheduler Trigger triggered by a Protocol Trigger, when triggered will remove selected Event and add the next or previous Event chosen in the given enum Event class. The chosen Event will be added to the list of events to be notified.
Example
Choose a user (A/B/C), using a knx up/down button, to select a playlist for a sound player.
Define Performers and Protocol Triggers:
- !Performer
name: "next user button"
for appliance: "a sound player"
triggers:
- !knx_plugin.trigger.dpt_updown.Up { addresses: [ 0x0C0A ] }
commands: [ ]
- !Performer
name: "previous user button"
for appliance: "a sound player"
triggers:
- !knx_plugin.trigger.dpt_updown.Down { addresses: [ 0x0C0A ] }
commands: [ ]
Define Scheduler Triggers:
- !protocol.enum.Trigger
name: "choose next user bagno"
notify more events: []
plus selected event: !home.event.user.Event.A
or: "next"
when triggered performers: "next user button"
- !protocol.enum.Trigger
name: "choose previous user"
notify more events: []
plus selected event: !home.event.user.Event.A
or: "previous"
when triggered performers: "previous user button"
Schedule the Scheduler Triggers:
- !schedule
trigger: "choose next user bagno"
for performers: "comandi sonos bagno"
- !schedule
trigger: "choose previous user bagno"
for performers: "comandi sonos bagno"
protocol.mean.Trigger
- class home.scheduler.trigger.protocol.mean.Trigger(name: str, events: Iterable[home.Event], protocol_trigger: home.protocol.mean.Mixin, num_of_samples: int, timeout_seconds: float)
A Scheduler Trigger which collects Protocol Trigger values in a time base.
protocol.mean.LesserThan
- class home.scheduler.trigger.protocol.mean.LesserThan(name: str, events: Iterable[home.Event], protocol_trigger: home.protocol.mean.Mixin, num_of_samples: int, hit_value: float, timeout_seconds: float)
A Scheduler Trigger triggered when collected Protocol Trigger values in a time base are lesser than given hit value.
>>> import sys >>> import home >>> import asyncio >>> from apscheduler.schedulers.asyncio import AsyncIOScheduler >>> >>> class PT(home.protocol.Trigger, home.protocol.mean.Mixin): ... def __init__(self, description, events): ... super(PT, self).__init__(description, events) ... self._values = [2, 3, 5, 8] ... def is_triggered(self, another_description): ... return True ... def make(self): ... ... ... def make_from(cls, msg): ... ... ... def get_value(self, description): ... try: ... return self._values.pop() ... except IndexError: ... return 4 >>> >>> async def wait_for_mean(scheduler_trigger, protocol_trigger): ... while not scheduler_trigger.is_triggered(protocol_trigger): ... await asyncio.sleep(0.01) ... asyncio.get_event_loop().stop() >>> >>> protocol_trigger = PT({"a": "description"}, ["an event"]) >>> scheduler_trigger = home.scheduler.trigger.protocol.mean.LesserThan("a mean scheduler trigger", ... ["another event"], ... protocol_trigger, ... 10, ... 5, ... 0.1) >>> scheduler = AsyncIOScheduler() >>> performer = home.Performer("pippo", home.appliance.light.Appliance("a light", []), [], []) >>> job = scheduler.add_job(wait_for_mean, ... scheduler_trigger, ... args=(scheduler_trigger, protocol_trigger, ), ... misfire_grace_time=180, ... coalesce=False) >>> scheduler.start() >>> asyncio.get_event_loop().run_forever()
Example
Calculate mean value for wind speed in a time base of 60 seconds. Notify when speed is weak.
Define the trigger Performer, using an home assistant sensor:
- !Performer
name: "wind performer"
for appliance: "anemometer"
commands: []
triggers:
- !home_assistant_plugin.service.sensor.float.trigger.Always {entity_id: "sensor.wind_speed"}
Define the Scheduler Trigger:
- !protocol.mean.LesserThan
name: "weak wind trigger"
notify more events:
- !home.event.wind.Event.Weak
when triggered performers: "wind performer"
num of samples: 5
hit value: 1
timeout seconds: 60
Schedule the Scheduler Trigger:
- !schedule
trigger: "weak wind trigger"
for performers: "a curtain command performer"
protocol.mean.GreaterThan
- class home.scheduler.trigger.protocol.mean.GreaterThan(name: str, events: Iterable[home.Event], protocol_trigger: home.protocol.mean.Mixin, num_of_samples: int, hit_value: float, timeout_seconds: float)
A Scheduler Trigger triggered when collected Protocol Trigger values in a time base are greater than given hit value.
>>> import sys >>> import home >>> import asyncio >>> from apscheduler.schedulers.asyncio import AsyncIOScheduler >>> >>> class PT(home.protocol.Trigger, home.protocol.mean.Mixin): ... def __init__(self, description, events): ... super(PT, self).__init__(description, events) ... self._values = [8, 6, 5, 2, ] ... def is_triggered(self, another_description): ... return True ... def make(self): ... ... ... def make_from(cls, msg): ... ... ... def get_value(self, description): ... try: ... return self._values.pop() ... except IndexError: ... return 5 >>> >>> async def wait_for_mean(scheduler_trigger, protocol_trigger): ... while not scheduler_trigger.is_triggered(protocol_trigger): ... await asyncio.sleep(0.01) ... asyncio.get_event_loop().stop() >>> >>> protocol_trigger = PT({"a": "description"}, ["an event"]) >>> scheduler_trigger = home.scheduler.trigger.protocol.mean.GreaterThan("a mean scheduler trigger", ... ["another event"], ... protocol_trigger, ... 10, ... 4, ... 0.1) >>> scheduler = AsyncIOScheduler() >>> performer = home.Performer("pippo", home.appliance.light.Appliance("a light", []), [], []) >>> job = scheduler.add_job(wait_for_mean, ... scheduler_trigger, ... args=(scheduler_trigger, protocol_trigger, ), ... misfire_grace_time=180, ... coalesce=False) >>> scheduler.start() >>> asyncio.get_event_loop().run_forever()
Example
Calculate mean value for sun brightness in a time base of 60 seconds. Notify when brightness is high.
Define the trigger Performer, using a knx sensor:
- !Performer
name: "luxmeter"
for appliance: "luxmeter"
commands: []
triggers:
- !knx_plugin.trigger.dpt_value_lux.Always {addresses: [0x0A04]}
Define the Scheduler Trigger:
- !protocol.mean.GreaterThan
name: "sun is bright (slowly changing)"
notify more events:
- !home.event.sun.brightness.Event.Bright
when triggered performers: "luxmeter"
num of samples: 250
hit value: 40000
timeout seconds: 60
Schedule the Scheduler Trigger:
- !schedule
trigger: "sun is bright (slowly changing)"
for performers: "a curtain command performer"
protocol.mean.InBetween
- class home.scheduler.trigger.protocol.mean.InBetween(name: str, events: Iterable[home.Event], protocol_trigger: home.protocol.mean.Mixin, num_of_samples: int, min_value: float, max_value: float, timeout_seconds: float)
A Scheduler Trigger triggered when collected Protocol Trigger values in a time base are lesser than max value and greater than min value.
>>> import sys >>> import home >>> import asyncio >>> from apscheduler.schedulers.asyncio import AsyncIOScheduler >>> >>> class PT(home.protocol.Trigger, home.protocol.mean.Mixin): ... def __init__(self, description, events): ... super(PT, self).__init__(description, events) ... self._values = [8, 6, 5, 2, 4] ... def is_triggered(self, another_description): ... return True ... def make(self): ... ... ... def make_from(cls, msg): ... ... ... def get_value(self, description): ... try: ... return self._values.pop() ... except IndexError: ... return 4 >>> >>> async def wait_for_mean(scheduler_trigger, protocol_trigger): ... while not scheduler_trigger.is_triggered(protocol_trigger): ... await asyncio.sleep(0.01) ... asyncio.get_event_loop().stop() >>> >>> protocol_trigger = PT({"a": "description"}, ["an event"]) >>> scheduler_trigger = home.scheduler.trigger.protocol.mean.InBetween("a mean scheduler trigger", ... ["another event"], ... protocol_trigger, ... 10, ... 4, ... 5, ... 0.1) >>> scheduler = AsyncIOScheduler() >>> performer = home.Performer("pippo", home.appliance.light.Appliance("a light", []), [], []) >>> job = scheduler.add_job(wait_for_mean, ... scheduler_trigger, ... args=(scheduler_trigger, protocol_trigger, ), ... misfire_grace_time=180, ... coalesce=False) >>> scheduler.start() >>> asyncio.get_event_loop().run_forever()
protocol.multi.Trigger
- class home.scheduler.trigger.protocol.multi.Trigger(name: str, events: Iterable[home.Event], positive_a: home.protocol.Trigger, negative_a: home.protocol.Trigger, positive_b: home.protocol.Trigger, negative_b: home.protocol.Trigger)
A Scheduler Trigger triggered when both trigger_positive_a and trigger_positive_b Protocol Triggers are triggered.
When triggered will notify given events plus all the protocol triggers events.
circadian_rhythm.Trigger
- class home.scheduler.trigger.circadian_rhythm.Trigger(name: str, events: Iterable[home.Event], events_in_a_day: List[home.Event])
A Scheduler Trigger triggered every 24*60 minutes / num of events_in_a_day.
Once every 24*60 minutes / num of events_in_a_day this trigger will notify one event in events_in_a_day. Starting from midnight.
Every time the trigger is triggered it will notify all the events specified in events.
Example
An example showing how the temperature’s Light could be adjusted during the day:
!circadian_rhythm.Trigger
name: "adjust light temperature during the day"
events: []
events_in_a_day:
- !light.event.circadian_rhythm.temperature.Event { value: 2700 }
- !light.event.circadian_rhythm.temperature.Event { value: 2700 }
- !light.event.circadian_rhythm.temperature.Event { value: 3600 }
- !light.event.circadian_rhythm.temperature.Event { value: 4600 }
- !light.event.circadian_rhythm.temperature.Event { value: 5600 }
- !light.event.circadian_rhythm.temperature.Event { value: 6500 }
- !light.event.circadian_rhythm.temperature.Event { value: 6500 }
- !light.event.circadian_rhythm.temperature.Event { value: 5600 }
- !light.event.circadian_rhythm.temperature.Event { value: 4600 }
- !light.event.circadian_rhythm.temperature.Event { value: 3600 }
- !light.event.circadian_rhythm.temperature.Event { value: 2700 }
- !light.event.circadian_rhythm.temperature.Event { value: 2700 }
A Protocol Command is linked with the lifx bulb Appliance through the following Performer:
!Performer
name: "lifx bulb command"
for appliance: "lifx bulb"
commands:
- !lifx_plugin.command.SetColor {addresses: [["172.31.10.245", 56700]]}
triggers: []
The Scheduler Trigger is scheduled to notify events to the light:
!schedule
trigger: "adjust light temperature during the day"
for performers: "lifx bulb command"
state.entering.Trigger
- class home.scheduler.trigger.state.entering.Trigger(name: str, events: Iterable[home.Events], state: str)
A Scheduler Trigger triggered when the given states, are changed and are become equals to the specified state.
Example
Set the lowest possible volume when entering Fade In state in a sound player.
Define the Scheduler Trigger:
- !state.entering.delay.Trigger
name: "fade in vol 1"
notify events:
- !sound.player.event.fade_in.volume.Event { value: 1 }
when appliance state became: "Fade In"
Schedule the Scheduler Trigger:
- !schedule
trigger: "fade in vol 1"
for performers: "fade sonos"
Define the Performer used by the scheduler:
- !Performer
name: "fade sonos"
for appliance: "sonos bagno"
commands:
- !soco_plugin.command.volume.ramp.Command { addresses: [ "Bagno" ], fields: { "ramp_type": 'SLEEP_TIMER_RAMP_TYPE' } }
triggers: [ ]
state.entering.delay.Trigger
- class home.scheduler.trigger.state.entering.delay.Trigger(name: str, events: Iterable[home.Event], state: str, timeout_seconds: float)
>>> import home >>> off = home.appliance.sound.player.state.off.State() >>> fade_in = home.appliance.sound.player.state.fade_in.State() >>> trigger = home.scheduler.trigger.state.entering.delay.Trigger("adjust volume", [], 'Fade In', 1) >>> trigger.is_triggered(off, fade_in) True
Example
Ramp up the volume when entering Fade In state in a sound player.
Define the Scheduler Trigger:
- !state.entering.delay.Trigger
name: "fade in vol 5"
notify events:
- !sound.player.event.fade_in.volume.Event { value: 5 }
when appliance state became: "Fade In"
and timeout expires: 10
Schedule the Scheduler Trigger:
- !schedule
trigger: "fade in vol 5"
for performers: "fade sonos"
Define the Performer used by the scheduler:
- !Performer
name: "fade sonos"
for appliance: "sonos bagno"
commands:
- !soco_plugin.command.volume.ramp.Command { addresses: [ "Bagno" ], fields: { "ramp_type": 'SLEEP_TIMER_RAMP_TYPE' } }
triggers: [ ]
state.entering.delay.duration.Trigger
- class home.scheduler.trigger.state.entering.delay.duration.Trigger(name: str, events: Iterable[home.Event], state: str)
A Scheduler Trigger triggered timeout_seconds (a value changing with the duration attribute of the new state) after a given Appliance state has been triggered. If an Appliance state has been triggered twice, the old scheduler trigger is disabled and a new one is started.
Example
Turn off a sprinkler, through a knx switch, after a variable time, timeout starts when the Sprinkler Appliance state became “On”. The time is expressed by the Appliance state through its duration property.
Define the Scheduler Trigger:
- !state.entering.delay.duration.Trigger
name: "disable sprinkler"
notify events:
- !home.event.enable.Event.Off
when appliance state became (and appliance state duration elapsed): "On"
Schedule the Scheduler Trigger:
- !schedule
trigger: "disable sprinkler"
for performers: "sprinkler command performer"
Define the Performer used by the scheduler:
- !Performer
name: "sprinkler command performer"
for appliance: "a sprinkler"
commands:
- !knx_plugin.command.dpt_switch.OnOff { addresses: [ 0x0C1C ] }
triggers: [ ]
state.exiting.Trigger
- class home.scheduler.trigger.state.exiting.Trigger(name: str, events: Iterable[home.Events], state: str)
A Scheduler Trigger triggered when the given states, are changed and the old state was equal to the specified state.
state.exiting.delay.Trigger
- class home.scheduler.trigger.state.exiting.delay.Trigger(name: str, events: Iterable[home.Event], state: str, timeout_seconds: float)