uun_iot.modules package

Submodules

uun_iot.modules.BaseHealthCheck module

class uun_iot.modules.BaseHealthCheck.BaseHealthCheck(gconfig, on_stop=True, on_update=True)

Bases: uun_iot.modules.Module.Module

Base class for creating runtime consistency checks and debug diagnostics.

HealthCheck modules can log incoming events and act correspondingly, ie. a HealthCheck module can restart application if the app is not working correctly and log more debugging information for further analysis.

This can be useful for detecting anomalous cases when the app is not working correctly due to an unknown bug, typically when the bug occurs at weird edge cases in long running apps. At the same time, it allows to recover from the case (by restarting) and to collect diagnostic info.

HealthCheck can monitor multiple modules at the same time. This is done using @on("tick", "moduleId") timers.

The monitored modules need to explicitly send information to the HealthCheck module as this cannot always be automatized.

Event informations are submitted by monitored modules via notify() method in a form of uun_iot.diagnostic.GeneralDiagnosticEvents.

BaseHealthCheck only provides infrastructure for storing these events and it is upto app-specific HealthCheck to implement neccessary logic for determining malfunctioning app states. Although, some helper functions for usual cases are already present in BaseHealthCheck (_act_if_data_not_sent()).

BaseHealthCheck can register basic @on methods automatically. This is useful, as most of derived HealthCheck modules will need save_all_diagnostics registered as @on(stop) and _init_module_config registered as @on(update) for updating module-specific leeways etc.

Sample gconfig:

 {
    "moduleTimers": {
        "healthCheck": {
            "weatherConditions": 300
        }
    }

    "healthCheck": {
        "diagnosticPath": "diagnostics/",
        "modules": {
            "weatherConditions": {
                "leeway": 120
            }
        }
    }
}

Leeways should be sufficiently large (larger than watched module’s timers), so that the module can do its function well. Only module-level configuration is dynamically updateable: leeways,

Parameters:

  • gconfig (dict) – gateway configuration

  • on_stop (bool) – default True, assign @on("stop") decorator to _save_all_diagnostics()

  • on_update (bool) – default True, assign @on(“update”) decorator to _init_module_config()

_abc_impl = <_abc._abc_data object>
_act_if_data_not_sent(mid, send_leeway, now)

Check if module (usually its uucmd send handler) sent data in time interval (now-send_leeway[mid], now).

If yes, save any current running diagnostic for mid. If not, run diagnostic - or - shutdown the application, if diagnostic is already running.

Note

Even if the module has no data to send, this will flag it as malfunctioning and restart the app.

Data send event is considered to be DiagnosticEvent.DATA_SEND_ATTEMPTED. No internet server communication is needed, only local information.

                   <--------> send_leeway
-*-----------------*----------------------| now
 ^ attempted sends ^
Parameters:

  • mid (str) – module id

  • send_leeway (datetime.timedelta) – specified time window (relative to now) to check for sent data

  • now (datetime.datetime) – current time

_diagnostic_active: Dict[str, threading.Event]
_diagnostic_lock: Dict[str, _thread.allocate_lock]
_diagnostic_log: Dict[str, List]
_diagnostics_save_path: str
_event_dates: Dict[str, Dict[uun_iot.diagnostic.GeneralDiagnosticEvent, datetime.datetime]]
_exit_app()
_init_module_config()

Initialize module-specific configuration of monitored modules from configuration. Ie. everything under modules.<moduleId>.* configuration. Currently only leeways.

_leeways: Dict[str, datetime.timedelta]
_start_date: datetime.datetime
notifier(mid)

Return a notify() method with prefilled mid argument.

Parameters:

mid (str) – module ID

notify(mid, event, *args, **kwargs)

Notify HealthCheck about watched modules’ events.

Optionally, add more custom data about the event in form of additional *args or **kwargs.

Parameters:

  • mid (str) – watched module ID

  • event (uun_iot.diagnostic.GeneralDiagnosticEvent) – GeneralDiagnosticEvent representing what action was done

  • *args – other positional arguments for logging purposes

  • **kwargs – other keyword arguments for logging purposes

save_all_diagnostics()

Stop all diagnostics and save them to files.

save_diagnostic(mid)

Stop logging incoming events for mid and save them to file.

start_diagnostic(mid)

Start logging incoming events for module mid.

uun_iot.modules.Heartbeat module

class uun_iot.modules.Heartbeat.Heartbeat(uucmd)

Bases: uun_iot.typing.IModule

Ping server in short periods to let server know if the gateway is online and send a little info about the gateway.

Parameters:

uucmd (Callable[[Dict], requests.models.Response]) – function (dto_in) -> requests.Response, the function takes an argument with data to be sent to the heartbeat uuCmd endpoint. It returns the reponse formed using requests.Response.

_abc_impl = <_abc._abc_data object>
id: str = 'heartbeat'
on_tick()

Determine online status and send a little information about gateway to uuApp.

The online status is determined using self._uucmd uuCmd. The information sent includes

  • boot timestamp (in ISO timestamp)

  • CPU usage in percent

  • RAM usage in percent

Executed on each timer tick.

uun_iot.modules.Module module

class uun_iot.modules.Module.ConfigScopeEnum(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: enum.Enum

Enumeration for configuration scopes.

GATEWAY = 1

Use the root gateway key for the module configuration.

SELF = 2

Use the gateway.moduleId key for the module configuration.

class uun_iot.modules.Module.Module(config={}, uucmd=None)

Bases: uun_iot.typing.IModule, abc.ABC

Provides utility functions for modules.

Class Module implements interface uun_iot.typing.IModule for derived classes automatically. The id attribute is derived from the class name using uun_iot.utils.module_id_from_str(). If the class already has the id attribute, keep it.

The derived classes must call the Module’s init function using super(). Example:

from uun_iot import Module
class TestModule(Module):
    def __init__(self, config, uucmd):
        super().__init__(config=config, uucmd=uucmd)
        self.attribute1 = "value1"

The utility functions include:

  • configuration management

  • storage management

  • backup and restore of storage to filesystem (useful for power failures)

  • automation for sending storage to uuApp

  • handling of failed send storage actions (useful for unstable internet connection)

Configuration management:

Access configuration in easy way as self._c("key/subkey/subsubkey") instead of cumbersome self._config["key"]["subkey"]["subsubkey"] and more. See configuration manager _config_manager() or its alias _c().

Storage management:

An instanced thread-safe Storage is available as _storage. The class provides backup and restore to JSON files and limiting the storage in number of its entries. If enabled, oldest entries will be removed for the sake of newer ones.

If the derived class does not have the _config attribute, set it to the constructor argument config.

Parameters:

  • uucmd (Callable[[Any], List]) – function(storage)->failed_send_data to send data to uuApp server

  • config (dict) – gateway configuration

__backup: bool

is backup to file active?

_abc_impl = <_abc._abc_data object>
_c(key=None, scope=ConfigScopeEnum.SELF)

Return configuration key.

Examples

Following example JSON configuration file with Python outputs:

{
    "gateway": {
        "testModule1": {
            "option1": "value1",
            "structure": {
                "field1": 1,
                "field2": {
                    "nested": 2
                }
            }
        },

        "testModule2": {
            "option2": "value2"
        }
    }

from uun_iot import ConfigScopeEnum

# inside of a method of class TestModule1 derived from Module
print(self._c("option1"))
>>> value1

print(self._c("structure/field1/nested"))
>>> 2

print(self._c("option1", scope=ConfigScopeEnum.SELF))
>>> value1

print(self._c("testModule1/option1", scope=ConfigScopeEnum.GATEWAY))
>>> value1

print(self._c("testModule2/option2", scope=ConfigScopeEnum.GATEWAY))
>>> value2
Parameters:

  • key (str | None) – specify value of configuration key to read. Set to None to return whole dictionary of corresponding configuration given by scope. delimit subdictionaries with / symbol, ie. self._c("color/bright/red") is the same as self._c("color")["bright"]["red"]

  • scope (uun_iot.modules.Module.ConfigScopeEnum) – the root of configuration scope can be any of ConfigScopeEnum. Defaults to ConfigScopeEnum.SELF, ie. key is taken as a subkey of gateway.<moduleId>. Alternatively, specify ConfigScopeEnum.GATEWAY to take key as subkey of gateway key

Returns:

the configuration entry

Return type:

Any

Raises:

KeyError – when a given key does not exist inside the configuration

_config: dict

gateway configuration

_config_manager(key=None, scope=ConfigScopeEnum.SELF)

Return configuration key.

Examples

Following example JSON configuration file with Python outputs:

{
    "gateway": {
        "testModule1": {
            "option1": "value1",
            "structure": {
                "field1": 1,
                "field2": {
                    "nested": 2
                }
            }
        },

        "testModule2": {
            "option2": "value2"
        }
    }

from uun_iot import ConfigScopeEnum

# inside of a method of class TestModule1 derived from Module
print(self._c("option1"))
>>> value1

print(self._c("structure/field1/nested"))
>>> 2

print(self._c("option1", scope=ConfigScopeEnum.SELF))
>>> value1

print(self._c("testModule1/option1", scope=ConfigScopeEnum.GATEWAY))
>>> value1

print(self._c("testModule2/option2", scope=ConfigScopeEnum.GATEWAY))
>>> value2
Parameters:

  • key (str | None) – specify value of configuration key to read. Set to None to return whole dictionary of corresponding configuration given by scope. delimit subdictionaries with / symbol, ie. self._c("color/bright/red") is the same as self._c("color")["bright"]["red"]

  • scope (uun_iot.modules.Module.ConfigScopeEnum) – the root of configuration scope can be any of ConfigScopeEnum. Defaults to ConfigScopeEnum.SELF, ie. key is taken as a subkey of gateway.<moduleId>. Alternatively, specify ConfigScopeEnum.GATEWAY to take key as subkey of gateway key

Returns:

the configuration entry

Return type:

Any

Raises:

KeyError – when a given key does not exist inside the configuration

_send_data(data, usecase=None)

Logging wrapper for sending data with usecase function.

Log a message with INFO level with information about the data, then send the data.

Parameters:

  • data (List | Any) – data to be sent by usecase as usecase(data)

  • usecase (Callable) – if None, use stored _uucmd

Raises:

TypeError – if usecase is not specified and self._uucmd is also not specified

_send_storage(usecase=None)

Sends entire storage of module to uuApp via usecase or stored uucmd.

The usecase takes data and outputs items which failed to be sent. This method saves these failed attempts to file and retries again together with new data on next envocation. Only failed entries will be saved to the backup file. This functionality is independent on the backup functionality.

Parameters:

  • usecase (Callable[[List], List] | None) – function List->List with one argument taking the data

  • If (to be sent and outputting list with items which were not sent.) –

  • None –

  • self._uucmd (use) –

Raises:

TypeError – if usecase is not specified and self._uucmd is also not specified

_storage: uun_iot.utils.Storage

storage facility

_uucmd: Callable[[...], List] | None

optional uuCmd function for data sending

id: str = None

is set in __init__() dynamically according to derived class

Module contents

Initialize modules.

uun_iot.modules.init(config, uuclient)
Parameters: