Predicates

Predicate are a way for the device to report any anomalies on it. It works by means of the userland querying smok-client about value of given sensors, and responding properly, by opening an event (alarm condition) or closing it.

Predicates

You work with predicates in such a way, that you register a bunch of classes to handle provided statistics. A predicate is defined on-server, and smok-client will take care to instantiate your classes with the proper data. Let’s see how a predicate is built:

class smok.predicate.BaseStatistic(device: SMOKDevice, predicate_id: str, verbose_name: str, silencing: List[DisabledTime], configuration: Optional[dict], statistic: Optional[str] = None, group: str = 'B', state=None, **kwargs)

Base class for your own predicates.

Variables

• device – a weak reference to the SMOKDevice

• predicate_id – ID of the predicate

• verbose_name – Human-readable name of this predicate

• silencing – periods during which the predicate shouldn’t generate alerts

• configuration – a dictionary containing the predicate’s configuration

• group – notification group

• state – state of the predicate, persisted between calls (picklable)

• statistic_name – name of this statistic

close_event(event: Event) → None

Close an event

Parameters

event – event to close

on_configuration_changed(new_config: dict) → None

Called upon configuration changing. This should assign the changed configuration.

Called by communicator thread.

Parameters

new_config – new configuration

on_group_changed(new_group: str) → None

Called upon group changing. This should assign the changed verbose name.

Called by communicator thread.

Parameters

new_group – new verbose name

on_offline() → None

Called when the predicate is disabled or deleted. After that, this class will be destroyed, and if the predicate gets enabled again, a new instance will be created.

Called by communicator thread.

on_silencing_changed(new_silencing: List[DisabledTime]) → None

Called upon silencing rules changing. This should assign the changed silencing.

Called by communicator thread.

Parameters

new_silencing – new silencing

abstract on_tick() → None

Called about each 60 seconds by the communicator thread. This should commence any required analyses.

state is loaded before this call and persisted after it finishes

on_verbose_name_changed(new_verbose_name: str) → None

Called upon verbose name changing. This should assign the changed verbose name.

Called by communicator thread.

Parameters

new_verbose_name – new verbose name

open_event(msg: str, color: Color) → Optional[Event]

Open an event.

This automatically checks for current silencing effect, and will return None if current time indicates that the event should be silenced.

Parameters

• msg – extra message for the event

• color – color of the event

Returns

an Event if silencing is not in effect, else None

class smok.predicate.Event(uuid_: Optional[str], started_on: Optional[int], ended_on: Optional[int], color: Color, is_point: bool, token: str, group: str, message: str, handled_by: Optional[str], metadata: Dict[str, str])

An object representing a single event in the SMOK system.

Variables

• uuid – event UUID (str) after being synced with the server

• provisional_uuid – UUID (str) assigned by the client after it’s opening

• started_on – timestamp of event beginning, in seconds (int)

• ended_on – timestamp of event ending in seconds, or None if not ended (tp.Optional[int])

• color – event color (Color)

• is_point – is the event a point one? (bool)

• token – token (str)

• group – event group (str)

• message – human-readable message (str)

• handled_by – user that handles this event (tp.Optional[str])

• metadata – event metadata (tp.Dict[str, str])

classmethod from_json(dct: dict) → Event

Restore self from a JSON representation

classmethod from_pickle(y: bytes) → Event

Load an event from a pickle

Parameters

y (bytes) – pickled Event

Returns

unpickled Event

Return type

Event

is_closed() → bool

Returns

is given event closed?

to_json() → dict

Convert self to JSON representation

to_pickle() → bytes

Returns

pickled self

class smok.predicate.Color(value)

Event severity

RED = 2

most severe event

WHITE = 0

least severe event

YELLOW = 1

event of medium severity

Example:

class MyStatistic(BaseStatistic):
    statistic_name = 'my'

    def on_tick(self):
        sen = self.device.get_sensor('value')
        if sen.get()[1] > 0 and self.state is None:
            self.state = self.open_event('Hello world!', Color.RED)
        elif self.state is not None and sen.get()[1] == 0:
            self.close_event(self.state)
            self.state = None

sd.register_statistic(MyStatistic, lambda stat, cfg: stat == 'my_statistic')

Silencing

During specified times, the user does not want to bother him with the predicate’s alerts. Following classes are given as arguments to your constructor:

class smok.predicate.Time(day_of_week: int, hour: int, minute: int)

A representation of a time during a weekday

Variables

• day_of_week – day of week, as per ISO 8601

• hour – a hour, according to a 24-hour clock

• minute – a minute

class smok.predicate.DisabledTime(start: Time, stop: Time)

Class marking a period during a week

Variables

• start – when this period starts

• stop – when this period stops

is_in_time(t: datetime) → bool

Check whether provided time is inside the range of this silencing period.

Returns

True if the time is inside

Opening, closing events and state

Every predicate has a magic property of state. It will be restored between calls to on_tick() and saved after it. You best store the Event that you’re created via open_event().

You open new events via open_event() and close them with close_event(). Example code could look like:

from satella.coding import silence_excs
from smok.predicate import BaseStatistic, Color, Event
from smok.exceptions import OperationFailedError

class CustomPredicate(BaseStatistic):
    """
    A predicate that watches for
    """
    statistic_name = 'test'

    @silence_excs(KeyError, OperationFailedError)
    def on_tick(self) -> None:
        sensor = self.device.get_sensor('value')
        self.device.execute(sensor.read())
        ts, v = sensor.get()
        if v == 10 and self.state is None:
            self.state = self.open_event('Value is equal to 10', Color.RED)     # type: Event
        elif v != 10 and self.state is not None:
            self.close_event(self.state)
            self.state = None

sd.register_statistic(CustomPredicate, lambda stat_name, cfg: stat_name == 'my_statistic')

Beware, :term:`point event`s cannot be closed as they do not span a period and are created closed.

Registrations

A register_statistic() returns objects of following type:

class smok.predicate.registration.StatisticRegistration(matcher: Callable[[str, Dict], bool], stat_class: Type[BaseStatistic])

A proof that given statistic has been registered

cancel()

Cancel this registration.

Note that this won’t update existing predicates.