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.


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.

  • 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


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.


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.


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.


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.


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.

  • msg – extra message for the event

  • color – color of the event


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.

  • 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


y (bytes) – pickled Event


unpickled Event

Return type


is_closed() bool

is given event closed?

to_json() dict

Convert self to JSON representation

to_pickle() bytes

pickled self

class smok.predicate.Color(value)

Event severity

RED = 2

most severe event


least severe event


event of medium severity


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.state = None

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


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

  • 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

  • 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.


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')
        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.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.


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 this registration.

Note that this won’t update existing predicates.