Variables and Expressions

Variables

Variables are the most important Connectable objects in an SHC application. They allow to keep track of the current state of some dynamic value (allowing to read it at any time), receive updates for that state and re-publish state changes to other connected objects. Thus, they are typically used as a “central connection point” for a certain value, to which all other Connectable objects for that value are connected, as in the following example:

ceiling_lights = shc.Variable(bool, "ceiling lights")
ceiling_lights.connect(knx_connection.group(shc.interfaces.knx.KNXGAD(1, 2, 3), dpt="1", init=True))

web_switch = shc.web.widgets.Switch("Ceiling Lights")\
    .connect(ceiling_lights)

Note

However, Variables are not absolutely necessary to interconnect datapoints of different interfaces with SHC. Especially, when dealing with stateless events (i.e. values which are used to trigger some action instead of representing a change in state), it is more appropriate to connect the relevant objects directly:

some_button = shc.web.widgets.StatelessButton(shc.interfaces.knx.KNXUpDown.DOWN, "down")
some_button.connect(knx_connection.group(shc.interfaces.knx.KNXGAD(3, 2, 1), dpt="1.008"))

In such cases, it is even harmful to use a variable as a middle link, since it only publishes changes in value and suppresses updates of an unchanged value. You may want to take a look at shc.misc.UpdateExchange, which can be used as a central exchange for value updates between multiple connectable objects without suppressing unchanged values.

Variables are shc.base.Readable (providing their current value), shc.base.Writable (to update the value) and shc.base.Subscribable (to publish changes of the value). In addition, they are optionally shc.base.Reading for initialization purposes: When a default provider is set, they will read its value once, immediately after startup of SHC, to initialize their value.

The value type of a variable must be given when it is instantiated. Optionally, a name and and initial value can be provided:

class shc.variables.Variable(type_: Type[T], name: Optional[str] = None, initial_value: Optional[T] = None)

A Variable object for caching and distributing values of a certain type.

Parameters:

type – The Variable’s value type (used for its .type attribute, i.e. for the Connectable type checking mechanism)
name – An optional name of the variable. Used for logging and future displaying purposes.
initial_value – An optional initial value for the Variable. If not provided and no default provider is set via set_provider(), the Variable is initialized with a None value and any read() request will raise an shc.base.UninitializedError until the first value update is received.

Tuple Field Access

When Variables are used with a value type based on typing.NamedTuple, they provide a special feature to access the individual fields of the tuple value: For each (type hinted) field of the named tuple type, a VariableField is created and can be accessed via the Variable.field() method. These objects are Connectable (taking their type attribute from the NamedTuple field’s type hint), which allows to subscribe other objects to that field’s value or let the field be updated from another Subscribable object:

from typing import NamedTuple

class Coordinate(NamedTuple):
    x: float
    y: float

var1 = shc.Variable(Coordinate, initial_value=Coordinate(0.0, 0.0))
# `var1` includes two VariableField objects to access the value's 'x' and 'y' fields.
# They are Connectables of `float` type, so we can connect them to a KNX GroupVariable with DPT 9:
var1.field('x').connect(knx_connection.group(shc.interfaces.knx.KNXGAD(1, 2, 0), dpt="9"))
var1.field('y').connect(knx_connection.group(shc.interfaces.knx.KNXGAD(1, 2, 1), dpt="9"))

@var1.trigger
@shc.handler()
async def my_handler(value, _origin):
    # This handler is triggered on any change of the var1 value, even if it originates from one of
    # the VariableFields. We could check the `origin` list here to find out about the source of the
    # update.
    if value.x > value.y:
        print("Value is in lower right half of the coordinate plane")

If the Variable’s type consists of nested NamedTuples types, the VariableField objects may include VariableFields of the respective sub-field’s type recursively. Equally, they can be retrieved via the VariableField.field() method.

Tip

If you use Python MyPy to statically typecheck your SHC-based application, MyPy will by default not be able to figure out the value type of the VariableField objects retrieved via .field (i.e. it defaults to Any). Thus, it can only warn you about general issues with the usage of the VariableField, but it can not warn you about a type mismatch with a Connectable object connected to it. To add this missing value type inference (which cannot be expressed with Python type hints – as far as I know), SHC comes with a little MyPy plugin.

To enable the MyPy plugin, create a MyPy config file (see https://mypy.readthedocs.io/en/stable/config_file.html) and set plugins = shc.util.mypy_variable_plugin in the [mypy] section. The plugin will only work if the tuple field name is passed to .field() as a string literal, like in the example above.

Tip

Use shc.misc.UpdateExchange to split up NamedTuple-based value updates in a stateless way: It provides an equal way for subscribing to fields of the NamedTuple via the shc.misc.UpdateExchange.field() method but does not store the latest value and does not suppress value updates with unchanged values.

DelayedVariable

class shc.variables.DelayedVariable(type_: Type[T], name: Optional[str] = None, initial_value: Optional[T] = None, publish_delay: timedelta = datetime.timedelta(microseconds=250000))

A Variable object, which delays the updates to avoid publishing half-updated values

This is achieved by delaying the publishing of a newly received value by a configurable amount of time (publish_delay). If more value updates are received while a previous update publishing is still pending, the latest value will be published at the originally scheduled publishing time. There will be no publishing of the intermediate values. The next value update received after the publishing will be delayed by the configured delay time again, resulting in a maximum update interval of the specified delay time.

This is similar (but slightly different) to the behaviour of shc.misc.RateLimitedSubscription.

Parameters:

type – The Variable’s value type (used for its .type attribute, i.e. for the Connectable type checking mechanism)
name – An optional name of the variable. Used for logging and future displaying purposes.
initial_value – An optional initial value for the Variable. If not provided and no default provider is set via set_provider(), the Variable is initialized with a None value and any read() request will raise an shc.base.UninitializedError until the first value update is received.
publish_delay – Amount of time to delay the publishing of a new value.

Expressions

Wouldn’t it be cool, if we could define logical-arithmetical relations between Variables (or other Connectables) as simple Python expressions, which would be evaluated for every value update? Example:

fan_state = shc.Variable(bool)
fan_switch = shc.Variable(bool)
temperature = shc.Variable(float)

# WARNING: This will not work! Read on before copy&pasting …
fan_state = fan_switch and temperature > 25

However, there are two caveats with this approach:

Using the = assignment operator does not work like this, as it would simply override the Variable object in the fan_state variable. But, we can make the result of that expression a Connectable object, so it can be connected with the result variable.
Overloading the operators like this may have unwanted side effects in the Python code, especially for the comparison operators like ==. Thus, we use a wrapper (ExpressionWrapper) for the Connectable objects which provides the overloaded operators to use them in expressions. For convenience, Variables provide this wrapper as a property .EX.

With those two fixes, the expression from above looks as follows:

fan_state = shc.Variable(bool)
fan_switch = shc.Variable(bool)
temperature = shc.Variable(float)

fan_state.connect(fan_switch.EX.and_(temperature.EX > 25))

There are still limitations to this expression syntax:

Python does not allow to override the and, or, and not operators. As a workaround, Expressions and ExpressionWrappers provide methods .and_(other) .or_(other) .not_(other) . Additionally, there are and_(), or_(), not_() functions, which apply the boolean operators in a functional style.
To ensure the static type checking mechanism for Connectable objects, we need to infer the type of an expression based on the operands’ value types. Since there is no generic type inference mechanism in Python at runtime, the shc.expression module contains a dict of type mapping rules for each supported operator. Only Connectables with value types covered by these rules can be used in expressions. Currently, such rules are only available for some basic builtin Python types and don’t even support subclasses of these types.

Tip

To use custom functions in SHC expressions, you can use the @expression decorator to turn any function into an ExpressionHandler evaluating that function with the latest values.

Classes for Building Expressions

The following classes are useful for building SHC Expressions. They are Readable and Subscribable objects, some of them already inheriting from ExpressionBuilder for operator support. In addition, if they subscribe to other objects to evaluate their value, these objects are given as constructor arguments, allowing to construct the SHC Expression in a single-line Python expression.

shc.variables.Variable (expression support through .EX property)
IfThenElse: SHC Expression version of a Python inline-if (a if condition else b)
Multiplexer: A multiplexer expression, selecting one of n input values using a integer control value
shc.misc.Hysteresis: Fixed threshold evaluation with hysteresis (expression support through .EX property)
shc.timer.TOn: Power-up delay of bool value (expression support through .EX property)
shc.timer.TOff: Turn-off delay of bool value (expression support through .EX property)
shc.timer.TOnOff: Resettable boolean delay (expression support through .EX property)
shc.timer.TPulse: Fixed-lenght Pulse generator from bool value (expression support through .EX property)
shc.timer.Delay: Universal value delay (expression support through .EX property)

`shc.expressions` Module Reference

class shc.expressions.ExpressionBuilder

An abstract base class for any object that may be used in SHC expressions.

This class defines all the overloaded operators, to create Connectable ExpressionHandler objects when being used in a Python expression. For using this class’s features with any Readable + Subscribable object, use the ExpressionWrapper subclass. Additionally, each ExpressionHandler is also an ExpressionBuilder, allowing to combine Expressions to even larger Expressions.

Currently, the following operators and builtin functions are supported:

+ (add)
- (sub)
* (mul)
/ (truediv)
// (floordiv)
% (mod)
abs()
-() (neg)
ceil()
floor()
round()
== (eq)
!= (ne)
< (lt)
<= (le)
> (gt)
>= (ge)

Additionally, the following methods are provided:

and_(other) → BinaryCastExpressionHandler

or_(other) → BinaryCastExpressionHandler

not_() → UnaryExpressionHandler[bool]

convert(type_: Type[S], converter: Optional[Callable[[T], S]] = None) → UnaryExpressionHandler[S]

Returns an ExpressionHandler that wraps this object to convert values to another type using a given converter function or the default converter for those two types.

Example:

value = shc.Variable(float)
which_one = shc.Variable(bool)
# display_string shall be a str-typed expression object
display_string = expression.IfThenElse(which_one, value.EX.convert(str), "static string")

Parameters:

type – The target type to convert new values to. This will also by the type attribute of the returned ExpressionHandler object, such that it will be used for further
converter – An optional converter function to use instead of the default conversion from this object’s type to the given target type.

class shc.expressions.ExpressionWrapper(wrapped: Subscribable[T]): Wrapper for any Readable + Subscribable object to equip it with expression-building capabilities, inherited from ExpressionBuilder

class shc.expressions.ExpressionHandler(type_: Type[T], operands: Iterable[object])

Base class for expression objects, i.e. the result object of an expression built with ExpressionBuilder’s overloaded operators

The ExpressionHandler object stores the operator (as a callable) and a reference to each operand. Thereby, it can read there current values and evaluate the operation/expression at any time with these values. ExpressionHandler objects are Readable (to evaluate the expression’s current value on demand) and Subscribable (to evaluate and publish the expression’s new value, when any of the operands is updated).

class shc.expressions.IfThenElse(condition: Union[bool, Readable[bool]], then: Union[T, Readable[T]], otherwise: Union[T, Readable[T]])

A ExpressionHandler version of the x if condition else y python syntax

This class takes three connectable objects or static values: condition, then and otherwise. condition must be bool (or a Connectable with value type bool), then and otherwise must be of the same type (reps. have the same value type). If the condition evaluates to True, this object evaluates to the value of then, otherwise it evaluates to the value of otherwise.

See also Multiplexer for switching between more than two values using an integer control value. If you only want enable and disable a single subscription dynamically (e.g. enable/disable the influence of a TimerSwitch on a Variable), take a look at shc.misc.BreakableSubscription instead.

class shc.expressions.Multiplexer(control: Readable[int], *inputs: Union[T, Readable[T]])

A ExpressionHandler that behaves as a multiplexer block with an integer control value and any number of inputs from that the output value is chosen

The control object, the first argument of this class, needs to be a Readable (and optionally Subscribable) object of int type. All other parameters are considered to be inputs. They must either be static values or Readable (and optionally Subscribable) objects. All of them must be of the same value type (resp. be instances of exactly that type). This type will also be the type of the Multiplexer itself.

The control signal selects which of the inputs is passed through and is the provided value of the Multiplexer. E.g., when the control object has the value 0, reading from the Multiplexer returns the first input’s current value (i.e. the value of the object given as the second argument); when the control object has the value 1, the second input’s value is returned, and so on. If the control value is out of range of the available inputs, reading returns an UninitializedError.

The Multiplexer publishes a value update whenever an update from the control object or the currently selected input is received – as long as they are Subscribable.

See also IfThenElse for simple cases with only two values and a boolean control value. If you only want enable and disable a single subscription dynamically (e.g. enable/disable the influence of a TimerSwitch on a Variable), take a look at shc.misc.BreakableSubscription instead.

shc.expressions.and_(a, b) → Union[bool, BinaryCastExpressionHandler[bool]]

Create a ExpressionHandler that wraps two Connectables or static values a,b and evaluates to a and b.

This is the workaround for Python’s limitations on overriding the and operator.

shc.expressions.or_(a, b) → Union[bool, BinaryCastExpressionHandler[bool]]

Create a ExpressionHandler that wraps two Connectables or static values a,b and evaluates to a or b.

This is the workaround for Python’s limitations on overriding the or operator.

shc.expressions.not_(a) → Union[bool, UnaryCastExpressionHandler[bool]]

Create a ExpressionHandler that wraps a Connectable or static value and evaluates to not x for the value x.

This is the workaround for Python’s limitations on overriding the not operator.

@shc.expressions.expression(func: Callable[[...], T]) → Type[ExpressionFunctionHandler[T]]

A decorator to turn any simple function into an ExpressionHandler class to be used in SHC expressions

This way, custom functions can be used in SHC expression in an intuitive way:

@expression
def invert(a: float) -> float:
    return 1 / a

var1 = Variable(float)
var2 = Variable(float).connect( invert(var1) )

The decorator turns the function into a class, such that calling it will return an instance of that class instead of evaluating the function. The resulting object is a Readable and Subscribable object, that evaluates the original function with the current values of the given arguments on every read() and every value update. The arguments passed to each instance must either be Readable (and optionally Subscribable) objects, providing values of the expected argument type, or simply static objects of the expected type.

The wrapped function must be synchronous (not async) and have a type annotation for its return type. The type annotation must refer to a runtime type, which can be used for dynamic type checking (i.e. no subscripted generic type; use list instead of List[int] and similar). This return type annotation is also the type of the resulting ExpressionHandler objects.

Currently only positional arguments (no keyword arguments) are supported for the functions.

Variables and Expressions

Variables

Tuple Field Access

DelayedVariable

Expressions

Classes for Building Expressions

shc.expressions Module Reference

`shc.expressions` Module Reference