Timers

The shc.timer module provides two different kinds of timer objects to be used in SHC configuration: “Schedule Timers” are Subscribable objects, that publish a None value regularly, based on some configurable schedule, e.g. with a fixed interval, a pattern of allowed datetime values or some amout of time after application startup. They are usually used to trigger a Logic Handler on a regular base. “Delay Timers” on the other hand, take a subscribable object and re-publish its published values after a certain delay time (following different rules)

Schedule Timers

class shc.timer.Once(offset: timedelta = datetime.timedelta(0), random: Optional[timedelta] = None, random_function: str = 'uniform')

A schedule timer which only triggers only once at startup of the SHC application, optionally delayed by some static offset and an additional random offset.

Parameters:

  • offset – The single trigger event is delayed by this timedelta after SHC application startup

  • random – A maximum offset to add to or substract from each trigger time. A random timedelta (roughly) between -random and +`random` will be added to the trigger time. If the resulting trigger time lies in the past, the trigger will executed directly upon application startup.

  • random_function – Identifier of a random distribution function to be used for calculating the random offset. See _random_time()’s documentation for supported values.

class shc.timer.Every(delta: timedelta, align: bool = True, offset: timedelta = datetime.timedelta(0), random: Optional[timedelta] = None, random_function: str = 'uniform')

A schedule timer that periodically triggers with a given interval, optionally extended/shortened by a random time.

It may either be triggered once on startup of SHC and then enter its periodical loop or be aligned with the wall clock, i.e. be triggered when the current time in the UNIX era is multiple of the interval. This alignment is enabled by default. It has the advantage of keeping the length of the interval, even when the SHC application is restarted—als long as application’s downtime does not fall together with the application’s downtime or in the gap between calculated time and (random) offset. In this case the trigger execution will be skipped once.

Thus, if it is important that the trigger being executed with at least the given interval, the align functionality should be turned off. Otherwise it will cause a more stable interval.

Parameters:

  • delta – The interval between trigger events

  • align – If True, the interval is kept even after restarts. If False, the timer triggers once upon application start up and keeps the interval from that point on.

  • offset – An offset to add to the trigger times. May be negative. This can be used for explicitly offsetting multiple Every timers with the same delta.

  • random – A maximum offset to add to or substract from each trigger time. A random timedelta (roughly) between -random and +`random` will be added to each individual trigger time.

  • random_function – Identifier of a random distribution function to be used for calculating the random offset. See _random_time()’s documentation for supported values.

class shc.timer.At(year: Optional[Union[int, Iterable[int], EveryNth]] = None, month: Optional[Union[int, Iterable[int], EveryNth]] = None, day: Optional[Union[int, Iterable[int], EveryNth]] = None, weeknum: Optional[Union[int, Iterable[int], EveryNth]] = None, weekday: Optional[Union[int, Iterable[int], EveryNth]] = None, hour: Optional[Union[int, Iterable[int], EveryNth]] = 0, minute: Optional[Union[int, Iterable[int], EveryNth]] = 0, second: Optional[Union[int, Iterable[int], EveryNth]] = 0, millis: Optional[Union[int, Iterable[int], EveryNth]] = 0, random: Optional[timedelta] = None, random_function: str = 'uniform')

Periodic timer which triggers on specific datetime values according to spec based on the Gregorian calendar and wall clock times. For each field in (year, month, day, hour, minute, second, millisecond) or (year, week, weekday, hour, minute, second, millisecond), a pattern may be specified, which may be

  • None (all values allowed for that field)

  • a single int value

  • a sorted list of valid int values

  • an EveryNth object (which is a wrapper around integers): It specifies that every nth value is valid (e.g. month=EveryNth(2) equals month=[1,3,5,7,9,11], hour=EveryNth(6) equals hour=[0,6,12,18])

The timer is scheduled for the next datetime matching the specified pattern of each field.

Parameters:

  • year – A pattern for the year value of the next trigger date/time (range: 0–2200)

  • month – A pattern for the month value of the next trigger date/time (range: 1–12). Must not be used together with weeknum or weekday.

  • day – A pattern for the day of month value of the next trigger date/time (range: 1–31). Must not be used together with weeknum or weekday.

  • weeknum – A pattern for the week of year value of the next trigger date/time according to the ISO weeknumber (range: 1-53). Must not be used together with month or day.

  • weekday – A pattern for the weekday value of the next trigger date/time (range: 1(monday) - 7(sunday)). Must not be used together with month or day.

  • hour – A pattern for the hour value of the next trigger time (range: 0–23).

  • minute – A pattern for the minute value of the next trigger time (range: 0–59).

  • second – A pattern for the second value of the next trigger time (range: 0–59).

  • millis – A pattern for the millisecond value of the next trigger time (range: 0–999).

  • random – A maximum offset to add to or substract from each trigger time. A random timedelta (roughly) between -random and +`random` will be added to each individual trigger time.

  • random_function – Identifier of a random distribution function to be used for calculating the random offset. See _random_time()’s documentation for supported values.

class shc.timer.EveryNth

A special integer class to be used as an argument for At.__init__() to specify that every nth number of the as a valid number of the specific field for triggering the timer.

E.g. month=EveryNth(2) equals month=[1,3,5,7,9,11], hour=EveryNth(6) equals hour=[0,6,12,18]

shc.timer._random_time(maximum_offset: Optional[timedelta], random_function: str = 'uniform') → timedelta

Generate a random timedelta within a given range.

Parameters:

  • maximum_offset – The maximum absolute value of the random timedelta. Depending on the random function, this might not be interpreted strictly (e.g. for the ‘gauss’ function, the value does only hit the interval by 96% chance). If None, return value is 0 seconds.

  • random_function – The random function / random distribution to use. Currently supported are ‘uniform’ and ‘gauss’.

Returns:

A timedelta roughly between -maximum_offset and +maximum_offset.

Convenience Timer Decorators

@shc.timer.once(*args, **kwargs) → Callable[[Callable[[T, List[Any]], Awaitable[None]]], Callable[[T, List[Any]], Awaitable[None]]]

Decorator for logic handlers to instantiate a Once timer and let it trigger the decorated logic handler.

All positional and keyword arguments are passed to Once.__init__().

@shc.timer.every(*args, **kwargs) → Callable[[Callable[[T, List[Any]], Awaitable[None]]], Callable[[T, List[Any]], Awaitable[None]]]

Decorator for logic handlers to instantiate a Every timer and let it trigger the decorated logic handler.

All positional and keyword arguments are passed to Every.__init__().

@shc.timer.at(*args, **kwargs) → Callable[[Callable[[T, List[Any]], Awaitable[None]]], Callable[[T, List[Any]], Awaitable[None]]]

Decorator for logic handlers to instantiate a At timer and let it trigger the decorated logic handler.

All positional and keyword arguments are passed to At.__init__().

Delay Timers

class shc.timer.TOn(wrapped: Subscribable[bool], delay: timedelta)

Power-up delay for bool-type Subscribable objects

This object wraps a Subscribable bool object and applies a power-up / turn-on delay to its value. I.e., a False value is re-published immediately, a True value is delayed for the configured time period. If a False is received during the turn-on delay period, the True value is superseded and the TOn’s value does not change to True at all.

The TOn object is Subscribable and Readable for publishing and providing the delayed value. Thus, it can be used like an SHC Expression. To actually use it in an Expression, you can use the EX property, which wraps the TOn object in a shc.expressions.ExpressionWrapper.

Parameters:

  • wrapped – The Subscribable object to be wrapped for delaying its value

  • delay – The power-up delay time. E.g. datetime.timedelta(seconds=5)

class shc.timer.TOff(wrapped: Subscribable[bool], delay: timedelta)

Turn-off delay for bool-type Subscribable objects

This object wraps a Subscribable bool object and applies a turn-off delay to its value. I.e., a True value is re-published immediately, a False value is delayed for the configured time period. If a True is received during the turn-off delay period, the False value is superseded and the TOff’s value does stay True continuously.

The TOff object is Subscribable and Readable for publishing and providing the delayed value. Thus, it can be used like an SHC Expression. To actually use it in an Expression, you can use the EX property, which wraps the TOff object in a shc.expressions.ExpressionWrapper.

Parameters:

  • wrapped – The Subscribable object to be wrapped for delaying its value

  • delay – The turn-off delay time. E.g. datetime.timedelta(seconds=5)

class shc.timer.TOnOff(wrapped: Subscribable[bool], delay: timedelta)

Resettable boolean-delay for Subscribable objects

This object wraps a Subscribable bool object and applies a delay to its value. It behaves like a series of a TOn and a TOff timer with the same delay time: All value updates of the wrapped object will be delayed. Albeit, in contrast to a Delay timer, a value update can be aborted by sending the contrary value during the delay period. Thus, short pulses in either direction (fast False→True→False or True→False→True toggling) is suppressed in the TOnOff’s value.

Like all delay timers, the TOnOff object is Subscribable and Readable for publishing and providing the delayed value. Thus, it can be used like an SHC Expression. To actually use it in an Expression, you can use the EX property, which wraps the TOnOff object in a shc.expressions.ExpressionWrapper.

Parameters:

  • wrapped – The Subscribable object to be wrapped for delaying its value

  • delay – The delay time. E.g. datetime.timedelta(seconds=5)

class shc.timer.TPulse(wrapped: Subscribable[bool], delay: timedelta)

Non-retriggerable pulse generator for bool-typed Subscribable objects

This object wraps a Subscribable bool object and creates a fixed length pulse (True period) on each rising edge (False→True) change of its value. The pulse is not retriggerable, i.e. it is not prolonged when a second raising edge occurs during the pulse.

Like all delay timers, the TPulse object is Subscribable and Readable for publishing and providing the pulse signal. Thus, it can be used like an SHC Expression. To actually use it in an Expression, you can use the EX property, which wraps the TPulse object in a shc.expressions.ExpressionWrapper.

Parameters:

  • wrapped – The Subscribable object to be wrapped for delaying its value

  • delay – The pulse length. E.g. datetime.timedelta(seconds=5)

class shc.timer.Delay(wrapped: Subscribable[T], delay: timedelta, initial_value: Optional[T] = None)

A Readable and Subscribable object which wraps another Subscribable object and re-publishes its updates after the time interval specified by delay. It also provides the state/value from delay time ago via read() calls.

Parameters:

  • wrapped – The Subscribable object to be wrapped for delaying its value

  • delay – The delay time. E.g. datetime.timedelta(seconds=5)

  • initial_value – If given, the value which is returned by read() before a value has been received from the wrapped object (and passed the delay period). If not given, read() raises an shc.base.UninitializedError in this case.

Helper Classes

class shc.timer.TimerSwitch(on: Iterable[Subscribable], off: Optional[Iterable[Subscribable]] = None, duration: Optional[timedelta] = None, duration_random: Optional[timedelta] = None)

A helper to program something similar to those old-fashioned digital timer switch adaptors

A TimerSwitch is basically a bool Variable with a fancy constructor to let some timers set the variable to True or False. Additionally, a duration mode is built in, which allows to specify a duration after which the TimerSwitch switches off (is set to False) automatically – similar to the TPulse timer (but with re-trigger-ability).

A simple usage example for switching on at 8:00 and off at 13:37, but 9:00 and 15:00 on weekends may look like:

timer_switch = TimerSwitch(on=[At(weekday=[1,2,3,4,5], hour=8),
                               At(weekday=[6,7], hour=9),
                           off=[At(weekday=[1,2,3,4,5], hour=13, minute=37),
                                At(weekday=[6,7], hour=15)]
my_ceiling_lights.connect(timer_switch)

To enable or disable the TimerSwitch, you may want to take a look at the shc.misc.BreakableSubscription.

Parameters:

  • on – A list of timers which should trigger a ‘switch on’ action (i.e. set the TimerSwitch’s value to True when triggering)

  • off – A list of timers which should trigger a ‘switch off’ action (i.e. set the TimerSwitch’s value to False when triggering). Either this parameter or the duration (but not both) must be specified.

  • duration – Period of time, after which the TimerSwitch will be switched off automatically after each ‘switch on’ event. Must be used as an alternative to the off parameter.

  • duration_random – An optional (maximum) offset to add to or substract from the duration period. If duration is used, a random timedelta between -duration_random and +duration_random will be added to each individual ‘on’ period.

class shc.timer.RateLimitedSubscription(wrapped: Subscribable[T], min_interval: float)

A transparent wrapper for Subscribable objects, that delays and drops values to make sure that a given maximum rate of new values is not exceeded.

See also shc.variables.DelayedVariable for a similar (but slightly different) behaviour.

Parameters:

  • wrapped – The Subscribable object to be wrapped

  • min_interval – The minimal allowed interval between published values in seconds

Ramp Generators

class shc.timer.AbstractRamp(type_: Type[T], ramp_duration: timedelta, dynamic_duration: bool = True, max_frequency: float = 25.0, enable_ramp: Optional[Readable[bool]] = None)

Abstract base class for all ramp generators

All ramp generators create smooth transitions from incoming value updates by splitting publishing multiple timed updates each doing a small step towards the target value. They are Readable and Subscribable to be used in Expressions.

In addition, the Ramp generators are Writable and Reading in order to connect them to a stateful object (like a Variable) which also receives value updates from other sources. For this to work flawlessly, the Ramp generator will stop the current ramp in progress, when it receives a value (via write()) from the connected object and it will read the current value of the connected object and use it as the start value for a ramp instead of the last value received from the wrapped object. Both of these features are optional, so the Ramp generator can also be connected to non-readable and non-subscribable objects.

Different derived ramp generator classes for different datatypes exist:

All of these ramp generator types allow to wrap a Subscribable object of one of their value types to subscribe to its value updates to turn them into ramps. Alternatively, they can be initialized with only the concrete value type only, so the ramp_to() method can be triggered manually from logic handlers etc.

As a special case, there’s the FadeStepRamp, which is a RangeFloat1-typed Connectable object, which wraps a Subscribable object of type shc.datatypes.FadeStep and creates ramps from the received fade steps (similar to shc.misc.FadeStepAdapter).

In addition, all of them take the following init parameters:

Parameters:

  • ramp_duration – The duration of the generated ramp/transition. Depending on dynamic_duration this is either the fixed duration of each ramp or it is the duration of a ramp across the full value range, which is dynamically lowered for smaller ramps.

  • dynamic_duration – If True (default) the duration of each ramp is dynamically calculated, such that all ramps/transitions have the same “speed” (only works for Range- and Range-based types). In this case, ramp_duration defines the speed by specifying the maximum duration, resp. the duration of a range across the full value range. If False, all ramps/transitions are stretched to the fixed duration.

  • max_frequency – The maximum frequency of value updates to be emitted by the ramp generator in updates per second (i.e. the “frame rate” of the ramp animation). The frequency is be dynamically reduced, if the resolution of the datatype cannot render the resulting number of steps/frames.

  • enable_ramp – Optional bypass: If a bool-typed readable object is provided, it is read at the beginning of each received value update. If its value evaluates to False, the ramp is bypassed and the new target value is republished immediately. If the value is True, the ramp generator is enabled. If no object is given, the ramp generator is always on.

class shc.timer.IntRamp(wrapped_or_type: Union[Subscribable[IntRampT], Type[IntRampT]], *args, **kwargs)
class shc.timer.FloatRamp(wrapped_or_type: Union[Subscribable[FloatRampT], Type[FloatRampT]], *args, **kwargs)
class shc.timer.HSVRamp(wrapped: Optional[Subscribable[HSVFloat1]], *args, **kwargs)
class shc.timer.RGBHSVRamp(wrapped: Optional[Subscribable[RGBUInt8]], *args, **kwargs)
class shc.timer.RGBWHSVRamp(wrapped: Optional[Subscribable[RGBWUInt8]], *args, **kwargs)
class shc.timer.FadeStepRamp(wrapped: Subscribable[FadeStep], *args, **kwargs)