Miscellaneous Helper Classes

The shc.misc Module provides a number of classes to help with connecting Connectable objects.

In addition to the classes provided by this module, you may want to take a look at the helper classes in the timer module.

class shc.misc.BreakableSubscription(wrapped: Subscribable[T], control: Readable[bool])

A transparent wrapper for Subscribable objects, that allows to dynamically switch forwarding of new values on and off.

A BreakableSubscription object can wrap any Subscribable object and is Subscribable itself to transparently re-publish all published values of the wrapped object. However, the re-publishing can be switched on and off through a Readable boolean control object.

This can be used to dynamically disable automatic control rules, e.g. control of variables through expressions. In the following example, the fan is automatically controlled via an expression, based on room_temperature, but only as long as the automatic_fan_control is on. Otherwise, it will not get value updates from that comparison expression and thus keep its current value:

fan = shc.Variable(bool)
room_temperature = shc.Variable(float)
automatic_fan_control = shc.Variable(bool)

fan.connect(shc.misc.BreakableSubscription(room_temperature.EX > 25.0,
                                           automatic_fan_control))

If the wrapped object is also Readable and the control object is also Subscribable, the current value of the wrapped object is read and published when the subscription is enabled (the control object changes to True).

If you want to select a value from two or more Readable and Subscribable objects dynamically, take a look at shc.expressions.IfThenElse or shc.expressions.Multiplexer.

Parameters:

  • wrapped – The Subscribable object to be wrapped

  • control – The Readable control object

class shc.misc.ConvertSubscription(wrapped: Subscribable[S], type_: Type[T], convert: Optional[Callable[[S], T]] = None)

An adapter, wrapping a subscribable object of value type S and converting it to a subscribable object of value type T, using the default converter or a given converter function.

Typically, the following methods should be used instead of this class:

However, they may be situations, where neither of these methods is applicable, e.g. when the subscribable object is not readable (so it should not used in an expression) and it is not used with .subscribe() or .connect() directly (e.g. it is passed to another object’s constructor).

class shc.misc.FadeStepAdapter(wrapped: Subscribable[FadeStep])

An adapter to connect a Subscribable object which publishes shc.datatypes.FadeStep values to a shc.datatypes.RangeFloat1 Variable (or another Readable + Wrtiable object of type RangeFloat1), such that the FadeSteps are applied to the current value of that object.

Example usage:

dim_up_button = shc.web.widgets.StatelessButton(FadeStep(0.1), "up")
dim_down_button = shc.web.widgets.StatelessButton(FadeStep(-0.1), "down")

dimmer_value = shc.Variable(RangeFloat1)            .connect(FadeStepAdapter(dim_up_button))            .connect(FadeStepAdapter(dim_down_button))

To apply FadeSteps as dynamic ramps of a fixed duration instead of a sudden jump, take a look at shc.timer.FadeStepRamp (see description in the documentation of shc.timer.AbstractRamp).

Parameters:

wrapped – The Subscribable object which shall be wrapped to apply its published FadeStep values to connected objects

is_reading_optional: bool = False
type

alias of RangeFloat1

class shc.misc.Hysteresis(wrapped: Subscribable[T], lower: T, upper: T, inverted: bool = False, initial_value: bool = False)

A Hysteresis function wrapper for Subscribable objects of any comparable type.

The Hysteresis object is a Subscribable and Readable object of boolean type, which wraps any Subscribable object of a type which is comparable with <. On any value update of the wrapped object, its current value is compared to two fixed bounds. When the value exceeds the upper bound, the Hysteresis publishes a True value; when it falls below the lower bound, the Hysteresis publishes a False value. As long as the value stays between the two bounds, the Hysteresis keeps its previous value (and does not publish anything). The output boolean value of the Hysteresis may be inverted via the inverted parameter.

Parameters:

  • wrapped – The Subscribable object to apply the hysteresis on

  • lower – The lower bound of the hysteresis

  • upper – The lower bound of the hysteresis

  • inverted – If True, the boolean output is inverted (False when value is above upper bound, True when value is below lower bound)

  • initial_value – The initial value of the Hysteresis object, which is returned on read() requests, until the first value outside of the bounds is received. Attention: If inverted is True, the initial value is inverted, too.

property EX: ExpressionWrapper[bool]
async read() → bool

Get the current value of this Connectable object.

Raises:

UninitializedError – If the value is not (yet) specified

type

alias of bool

class shc.misc.PeriodicReader(wrapped: Readable[T], interval: timedelta)

Wraps a Readable object to turn it into Subscribable object by periodically reading and publishing its value.

Parameters:

  • wrapped – The Readable object, which shall be wrapped and read periodically

  • interval – The interval in which the object’s .read() coroutine is called and the result is published, e.g. datetime.timedelta(seconds=15).

async do_read(_value, origin) → None
async read() → T

Get the current value of this Connectable object.

Raises:

UninitializedError – If the value is not (yet) specified

class shc.misc.SimpleInputConnector(type_: Type[T], callback: Optional[Callable[[List[Any]], Awaitable[None]]] = None)

A generic Reading object to be used as a simple input connector for “function block” classes

The connector is also Writable to invoke a value update, but it does not use the written value. Instead, the value is read from the configured default provider even then.

Parameters:

  • type – The value type of this connector

  • callback –

    A coroutine to be called when a value is written to this connector write(). It must take a single argument: The origin list of the value update.

    Important: When the coroutine triggers subsequent value updates which are propagated to other connectable objects, make sure to pass the origin along when publishing these value updates and to return from this coroutine only as soon as the subsequent value updates have been published/written.

async get_value()

Internal method to be called by the “function block” object, this connector is part of, to read the attached provider’s value.

is_reading_optional: bool = False
class shc.misc.SimpleOutputConnector(type_: Type[T], initial_value: Optional[T] = None)

A generic readable + subscribable object with value caching to be used as a simple output connector for “function block” classes

Note: Under “pure functional” conditions, when exactly one output value is calculated from one or more other values, without internal state or side effects, creating an SHC expression is typically a better choice.

Parameters:

  • type – The value type of this connector

  • initial_value – Initial value for the value attribute.

property EX: ExpressionWrapper[T]
async read() → T

Get the current value of this Connectable object.

Raises:

UninitializedError – If the value is not (yet) specified

set_generated_value(value: T) → None

Set and publish the value of this output connector from an internally generated value update

The value is only published when it changes (similar to shc.Variable).

If the value update is the direct consequence of an incoming value update, use set_value() instead and await its completion to ensure correct detection of concurrent value updates via this connector.

async set_value(value: T, origin: List[Any]) → None

Set and publish the value of this output connector as a result of any received value update

The value is only published when it changes (similar to shc.Variable).

Make sure to await the completion of this coroutine before returning from the _write() method, which received the incoming value update to ensure correct detection of concurrent value updates via this connector. I.e. don’t call this method in a new asyncio Task. In addition, don’t

class shc.misc.TwoWayPipe(type_: Type[T])

A helper to connect two sets of Writable+Subscribable objects, without connecting the objects within each set. This object can be thought of as a bidirectional pipe: All write-events send to the left end of the pipe are forwarded to all objects on the right side and vice versa. This can be especially useful to connect a variable to multiple addresses of a home automation bus (like a KNX bus) without all mirroring incoming values on one address (e.g. a central or feedback datapoint) to the other addresses.

To connect objects to one side of the pipe, use connect_left() resp. connect_right(). The additional method connect() is an alias for connect_left to allow using a TwoWayPipe object as an argument for Connectable.connect() itself with the result of connecting the pipe’s left end to the object.

The following example demonstrates, how to connect two interface connectors to a Variable, such that the Variable will interact with both of them, without forwarding events/values from one connector to the other:

shc.Variable(bool)\
    .connect(TwoWayPipe(bool)
        .connect_right(some_interface.connector(1))
        .connect_right(some_interface.connector(2)))
Parameters:

type – The type of the values to be forwarded. This is used as the type of the two pipe-end connectable objects.

connect(*args, **kwargs) → TwoWayPipe

Subscribe self to other and set as default_provider and vice versa (depending on the two object’s capabilities, optionalities and given parameters).

Parameters:

  • other – The other Connectable object to connect with

  • send – Send value updates to other (i.e. subscribe other to self). Requires self to be Subscribable and other to be Writable. In this case, defaults to True if not specified.

  • receive – Receive value updates from other (i.e. subscribe self to other). Requires self to be Writable and other to be Subscribable. In this case, defaults to True if not specified.

  • read – Read values from other (i.e. set other as default provider for self). Requires self to be Reading and other to be Readable. If not specified, defaults to False unless self.is_reading_optional is set to False.

  • provide – Provide values to other for reading (i.e. set self as default provider for other). Requires self to be Readable and other to be Reading. If not specified, defaults to False unless other.is_reading_optional is set to False.

  • convert – Enable built-in type conversion for the created subscriptions/default_providers: Either a boolean or a tuple of two conversion functions. Defaults to False, i.e. a type mismatch will result in a TypeError instead of implicit conversion. Set to True, to choose the appropriate conversion function automatically. Raises a TypeError if no default conversion for one of the two directions is available and that direction is used. For custom conversion functions, pass a tuple (to_other, to_self), where to_other is a (non-async) callable for converting self’s value type into other’s value type and to_self does the conversion the other way round.

Returns:

Returns self to allow functional-style chaining

connect_left(*args, **kwargs) → TwoWayPipe
connect_right(*args, **kwargs) → TwoWayPipe
class shc.misc.UpdateExchange(type_: Type[T])

A “message exchange” for distributing value updates. Similar to shc.variables.Variable but stateless.

In contrast to a Variable, an UpdateExchange does not store the latest value. Thus, it is not Readable and it forwards every single value update to all subscribers, even if the value is equal to the previous value update.

Similar to Variable, an UpdateExchange of a NamedTuple-based value type has additional connectors for each field of the NamedTuple, which can retrieved via the field() method. This can be used for splitting up tuple-typed value updates, i.e. subscribing other Connectable objects to one specific field of the values published by the UpdateExchange. In contrast to Variable’s fields, the UpdateExchangeField objects are not Writable; only the UpdateExchange itself can receive value updates.

field(name: str) → _UpdateExchangeField