Core plug-ins

Developing core plug-ins is the most advanced way to extend EVA ICS functionality. You can create own API methods, functions for Logic control macros and SFA Templates plus much and much more.

Guide by example

Note

All plugin objects and custom API functions are always registered as x_{plugin}_{name}

"""
EVA ICS plugin example

To load the plugin, put its name to controller config, section [server]

e.g. edit etc/uc.ini:

...........................................
[server]
plugins = my ; plugin list, comma separated
...........................................

Plugins can be one-file, in this case they can be just put into plugins
(/opt/eva/plugins) directory and named e.g. "my.py".

If your plugin is more complex or you want to redistribute it e.g. via PyPi -
create Python module called "evacontrib.<yourpluginname>" and install it either
globally or in EVA ICS venv.

Plugin configuration

Put a section [<pluginname>] in controller config, e.g. for this plugin:

......
[my]
var1 = value1
var2 = value2
......
"""

# plugin header, required
__author__ = 'Altertech, https://www.altertech.com/'
__copyright__ = 'Copyright (C) 2012-2020 Altertech'
__license__ = 'Apache License 2.0'
__version__ = '0.0.1'

# import EVA ICS Plugin API library
import eva.pluginapi as pa
"""
"flags" namespace can be defined, EVA ICS uses this namespace to determine
plugin status:

flags.ready = True # means plugin is correctly initialized and works properly
"""
from types import SimpleNamespace
flags = SimpleNamespace(ready=False)

# init plugin logger
logger = pa.get_logger()


def init(config, **kwargs):
    """
    Called by EVA ICS core when the initial configuration is loaded. All
    methods are optional, if the method doesn't exist in plugin, no exception
    is raised

    All methods should have **kwargs in argument list to accept extra arguments
    from future Plugin API versions

    Args:
        config: plugin configuration (comes as key/value dict)
    """
    # require Plugin API version 1+
    pa.check_version(1)
    # this feature is only for LM PLC
    try:
        pa.check_product('lm')
        # register new lmacro function "mytest"
        pa.register_lmacro_object('mytest', mytest)
        # register lmacro object "weekend"
        pa.register_lmacro_object('weekend', weekend)
    except:
        pass
    # this feature is only for SFA
    try:
        pa.check_product('sfa')
        # register SFA Templates function "weekend_days"
        pa.register_sfatpl_object('weekend_days', get_weekend)
    except:
        pass
    """
    register API extension blueprint

    currently only JSON RPC and direct API calling methods can be registered
    """
    pa.register_apix(MyAPIFuncs(), sys_api=False)
    flags.ready = True


def before_start(**kwargs):
    """
    Called right before controller start
    """
    logger.info('plugin my before start called')


def start(**kwargs):
    """
    Called after controller start
    """
    logger.info('plugin my start called')


def before_stop(**kwargs):
    """
    Called right before controller stop
    """
    logger.info('plugin my before stop called')


def stop(**kwargs):
    """
    Called after controller stop
    """
    logger.info('plugin my stop called')


def dump(**kwargs):
    """
    Called after controller stop
    """
    return 'something'


def handle_state_event(source, data, **kwargs):
    """
    Called when any state event is received

    Args:
        source: event source item (object)
        data: serialized item state dict
    """
    logger.info(f'event from {source.oid}')
    logger.info(data)


# custom plugin code

weekend = ['Sat', 'Sun']


# the function we registered for SFA TPL
def get_weekend():
    return ','.join(weekend)


# the function we registered for LM PLC macros
def mytest():
    logger.info('something')


# APIX blueprint to implement new API functions
class MyAPIFuncs(pa.APIX):

    # log API call as DEBUG
    @pa.api_log_d
    # require master key
    @pa.api_need_master
    def my_square(self, **kwargs):
        # parse incoming params
        x = pa.parse_api_params(kwargs, 'x', 'N')
        if x < 0:
            raise pa.InvalidParameter('x < 0')
        # return some result
        # if API method produces no result, it SHOULD return True
        return {
            'result': x * x,
            'you': pa.get_aci('key_id'),
            'me': [pa.get_directory('eva'),
                   pa.get_product().build]
        }

    # log API call as INFO
    @pa.api_log_i
    # require master key
    @pa.api_need_master
    def my_test(self, **kwargs):
        # let's return the result of test_phi API function
        return pa.api_call('test_phi', i='ct1', c='self')

Resources

class eva.pluginapi.APIX

API blueprint extension class

exception eva.pluginapi.AccessDenied(msg='')

raised when call has no access to the resource

__str__()

Return str(self).

exception eva.pluginapi.FunctionFailed(msg='')

raised with function failed with any reason

__str__()

Return str(self).

exception eva.pluginapi.InvalidParameter
__str__()

Return str(self).

__weakref__

list of weak references to the object (if defined)

class eva.pluginapi.MQTT(notifier_id)

MQTT helper class

Parameters:notifier_id – MQTT notifier to use (default: eva_1)
__init__(notifier_id)
Parameters:notifier_id – MQTT notifier to use (default: eva_1)
__weakref__

list of weak references to the object (if defined)

register(topic, func, qos=1)

Register MQTT topic handler

send(topic, data, retain=None, qos=1)

Send MQTT message

unregister(topic, func)

Unregister MQTT topic handler

exception eva.pluginapi.MethodNotFound

raised when requested method is not found

__str__()

Return str(self).

__weakref__

list of weak references to the object (if defined)

exception eva.pluginapi.MethodNotImplemented(msg='')

raised when requested method exists but requested functionality is not implemented

__str__()

Return str(self).

exception eva.pluginapi.ResourceAlreadyExists(msg='')

raised when requested resource already exists

__str__()

Return str(self).

exception eva.pluginapi.ResourceBusy(msg='')

raised when requested resource is busy (e.g. can’t be changed)

__str__()

Return str(self).

exception eva.pluginapi.ResourceNotFound(msg='')

raised when requested resource is not found

__str__()

Return str(self).

exception eva.pluginapi.TimeoutException(msg='')

raised when call is timed out

eva.pluginapi.api_call(method, key_id=None, **kwargs)

Call controller API method

Parameters:
  • key_id – API key ID. If key_id is None, masterkey is used
  • other – passed to API method as-is
Returns:

API function result

Raises:

eva.exceptions

eva.pluginapi.api_log_d(f)

API method decorator to log API call as DEBUG

eva.pluginapi.api_log_i(f)

API method decorator to log API call as INFO

eva.pluginapi.api_log_w(f)

API method decorator to log API call as WARNING

eva.pluginapi.api_need_cmd(f)

API method decorator to pass if API key has “cmd” allowed

eva.pluginapi.api_need_file_management(f)

API method decorator to pass if file management is allowed in server config

eva.pluginapi.api_need_lock(f)

API method decorator to pass if API key has “lock” allowed

eva.pluginapi.api_need_master(f)

API method decorator to pass if API key is masterkey

eva.pluginapi.api_need_rpvt(f)

API method decorator to pass if rpvt is allowed in server config

eva.pluginapi.api_need_sysfunc(f)

API method decorator to pass if API key has “sysfunc” allowed

eva.pluginapi.check_product(code)

Check controller type

Parameters:code – required controller type (uc, lm or sfa)
Raises:RuntimeError – if current controller type is wrong
eva.pluginapi.check_version(min_version)

Check plugin API version

Parameters:min_version – min Plugin API version required
Raises:RuntimeError – if Plugin API version is too old
eva.pluginapi.critical()

Send critical event

eva.pluginapi.get_aci(field, default=None)

get API call info field

Parameters:
  • field – ACI field
  • default – default value if ACI field isn’t set
Returns:

None if ACI field isn’t set

eva.pluginapi.get_db()

get SQLAlchemy connection to primary DB

eva.pluginapi.get_directory(tp)

Get path to EVA ICS directory

Parameters:tp – directory type: eva, runtime, ui, pvt or xc
Raises:LookupError – if directory type is invalid
eva.pluginapi.get_item(i)

Get controller item

Parameters:i – item oid
Returns:None if item is not found
eva.pluginapi.get_logger()

Get plugin logger

Returns:logger object
eva.pluginapi.get_masterkey()

get master API key

Returns:master API key
eva.pluginapi.get_polldelay()

Get core poll delay

eva.pluginapi.get_product()

Get product object

Returns:namespace(name, code, build)
eva.pluginapi.get_sleep_step()

Get core sleep step

eva.pluginapi.get_system_name()

Get system name (host name)

eva.pluginapi.get_timeout()

Get default timeout

eva.pluginapi.get_userdb()

get SQLAlchemy connection to user DB

eva.pluginapi.get_version()

Get Plugin API version

eva.pluginapi.key_check(k, item=None, allow=[], pvt_file=None, rpvt_uri=None, ip=None, master=False, sysfunc=False, ro_op=False)

check API key access

Arguments are ACL which can be combined

Parameters:
  • items – item objects
  • allow – check allows
  • pvt_file – access to pvt resource
  • pvt_file – access to rpvt resource
  • ip – caller IP
  • master – is master access required
  • sysfunc – is sysfunc required
  • ro_op – is item operation read-only
eva.pluginapi.key_check_master(k)

check is given key a masterkey

eva.pluginapi.key_id(k)

get key ID by API key

Returns:API key ID
eva.pluginapi.log_traceback()

Log traceback

eva.pluginapi.parse_api_params(params, names='', types='', defaults=None)

calls parse_function_params but omits API key

eva.pluginapi.parse_function_params(params, names, types='', defaults=None, e=<class 'eva.tools.InvalidParameter'>, ignore_extra=False)
Parameters:
  • names – parameter names (list or string if short) S: equal to ‘save’ Y: equal to ‘full’ J: equal to ‘_j’ F: equal to ‘force’
  • values – parameter values R: required, any not null and non-empty string r: required, but empty strings are possible s: required, should be string S: required, should be non-empty string b: boolean (or 0/1 or boolean-like strings) B: boolean (or 0/1 or boolean-like strings), required i: integer, can be None f or n: float(number), can be None I: integer, required F or N: float(number), required D: dict, required T: tuple, required X: set, required L: list, required . (dot): optional o: oid, can be null O: OID required
  • params – dict
  • defaults – dict (name/value)
  • e – exception to raise
class eva.pluginapi.partial

partial(func, *args, **keywords) - new function with partial application of the given arguments and keywords.

__call__()

Call self as a function.

__delattr__()

Implement delattr(self, name).

__getattribute__()

Return getattr(self, name).

__new__()

Create and return a new object. See help(type) for accurate signature.

__reduce__()

Helper for pickle.

__repr__()

Return repr(self).

__setattr__()

Implement setattr(self, name, value).

args

tuple of arguments to future partial calls

func

function object to use in future partial calls

keywords

dictionary of keyword arguments to future partial calls

eva.pluginapi.register_apix(o, sys_api=False)

Register API extension (APIX) object

All object methods (except internal and private) are automatically exposed as API functions

Functions are registered as x_{plugin}_{fn}

Parameters:
  • o – APIX object
  • sys_api – if True, object functions are registered as SYS API
eva.pluginapi.register_lmacro_object(n, o)

Register custom object for LM PLC macros

Object is registered as x_{plugin}_{n}

Parameters:
  • n – object name
  • o – object itself
eva.pluginapi.register_sfatpl_object(n, o)

Register custom object for SFA Templates

Object is registered as x_{plugin}_{n}

Parameters:
  • n – object name
  • o – object itself
eva.pluginapi.set_aci(field, value)

set API call info field

Parameters:
  • field – ACI field
  • value – field value
Returns:

True if value is set, False for error (e.g. ACI isn’t initialized)

eva.pluginapi.snmp_get(oid, host, port=161, community='public', timeout=0, retries=0, rf=<class 'str'>, snmp_ver=2, walk=False)
Parameters:
  • oid – SNMP OID or MIB name
  • host – target host
  • port – target port (default: 161)
  • community – SNMP community (default: public)
  • timeout – max SNMP timeout
  • retries – max retry count (default: 0)
  • rf – return format: str, float, int or None
  • snmp_ver – SNMP version (default: 2)
  • walk – if True, SNMP walk will be performed
Returns:

If rf is set to None, raw pysnmp object is returned, otherwise parsed to float, int or str

If walk is requested, list of pysnmp objects is returned

eva.pluginapi.snmp_set(oid, value, host, port=161, community='private', timeout=0, retries=0, snmp_ver=2)
Parameters:
  • oid – SNMP OID or MIB name
  • value – value to set
  • host – target host
  • port – target port (default: 161)
  • community – SNMP community (default: public)
  • timeout – max SNMP timeout
  • retries – max retry count (default: 0)
  • snmp_ver – SNMP version (default: 2)
Returns:

True if value is set, False if not

eva.pluginapi.spawn(f, *args, **kwargs)

Run function as a thread in EVA ICS thread pool

Parameters:
  • f – callable
  • args/kwargs – passed to function as-is
Returns:

concurrent.futures Future object

eva.pluginapi.upnp_discover(st, ip='239.255.255.250', port=1900, mx=True, interface=None, trailing_crlf=True, parse_data=True, discard_headers=['Cache-control', 'Host'], timeout=None)

discover uPnP equipment

Parameters:
  • st – service type
  • ip – multicast ip
  • port – multicast port
  • mx – use MX header (=timeout)
  • interface – network interface (None - scan all)
  • trailing_crlf – put trailing CRLF at the end of msg
  • parse_data – if False, raw data will be returned
  • discard_headers – headers to discard (if parse_data is True)
  • timeout – socket timeout (for a single interface)
Returns:

list of dicts, where IP=equipment IP, otherwise dict, where key=equipment IP addr, value=raw ssdp reply. Note: if data is parsed, all variables are converted to lowercase and capitalized.

Return type:

if data is parsed