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


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:

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

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:

var1 = value1
var2 = value2

# plugin header, required
__author__ = 'Altertech,'
__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

        config: plugin configuration (comes as key/value dict)
    # require Plugin API version 1+
    # this feature is only for LM PLC
        # register new lmacro function "mytest"
        pa.register_lmacro_object('mytest', mytest)
        # register lmacro object "weekend"
        pa.register_lmacro_object('weekend', weekend)
    # this feature is only for SFA
        # register SFA Templates function "weekend_days"
        pa.register_sfatpl_object('weekend_days', get_weekend)
    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
    """'plugin my before start called')

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

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

def stop(**kwargs):
    Called after controller stop
    """'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

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

# 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():'something')

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

    # log API call as DEBUG
    # require master key
    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'),

    # log API call as INFO
    # require master key
    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')


class eva.pluginapi.APIX

API blueprint extension class

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

raised when call has no access to the resource


Return str(self).

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

raised with function failed with any reason


Return str(self).

exception eva.pluginapi.InvalidParameter

Return str(self).


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)
Parameters:notifier_id – MQTT notifier to use (default: eva_1)

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


Return str(self).


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


Return str(self).

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

raised when requested resource already exists


Return str(self).

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

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


Return str(self).

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

raised when requested resource is not found


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

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

API function result




API method decorator to log API call as DEBUG


API method decorator to log API call as INFO


API method decorator to log API call as WARNING


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


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


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


API method decorator to pass if API key is masterkey


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


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


Check controller type

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

Check plugin API version

Parameters:min_version – min Plugin API version required
Raises:RuntimeError – if Plugin API version is too old

Send critical event

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

get API call info field

  • field – ACI field
  • default – default value if ACI field isn’t set

None if ACI field isn’t set


get SQLAlchemy connection to primary DB


Get path to EVA ICS directory

Parameters:tp – directory type: eva, runtime, ui, pvt or xc
Raises:LookupError – if directory type is invalid

Get controller item

Parameters:i – item oid
Returns:None if item is not found

Get plugin logger

Returns:logger object

get master API key

Returns:master API key

Get core poll delay


Get product object

Returns:namespace(name, code, build)

Get core sleep step


Get system name (host name)


Get default timeout


get SQLAlchemy connection to user DB


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

  • 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

check is given key a masterkey


get key ID by API key

Returns:API key ID

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 ''>, ignore_extra=False)
  • 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 self as a function.


Implement delattr(self, name).


Return getattr(self, name).


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


Helper for pickle.


Return repr(self).


Implement setattr(self, name, value).


tuple of arguments to future partial calls


function object to use in future partial calls


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}

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

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

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

set API call info field

  • field – ACI field
  • value – field value

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

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

True if value is set, False if not

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

Run function as a thread in EVA ICS thread pool

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

concurrent.futures Future object

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

discover uPnP equipment

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

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