API reference

SDK

class evaics.sdk.ACI(aci_payload)

ACI (API Call Info) helper class

is_writable()

Check is the current session writable or read-only

class evaics.sdk.Action(event)

Item action from bus event

class evaics.sdk.Controller(bus)

Action handler helper class for controllers

class evaics.sdk.EvaLogHandler(bus)

Initializes the instance - basically setting the formatter to None and the filter list to empty.

emit(record)

Do whatever it takes to actually log the specified logging record.

This version is intended to be implemented by subclasses and so raises a NotImplementedError.

class evaics.sdk.LocalProxy

Simple proxy for threading.local namespace

clear(attr)

Clear (delete) thread-local attribute

Parameters

attr – attribute name

get(attr, default=None)

Get thread-local attribute

Parameters
  • attr – attribute name

  • default – default value if attribute is not set

Returns

attribute value or default value

has(attr)

Check if thread-local attribute exists

Parameters

attr – attribute name

Returns

True if attribute exists, False if not

set(attr, value)

Set thread-local attribute

Parameters
  • attr – attribute name

  • value – attribute value to set

class evaics.sdk.OID(s, from_path=False)

Base item OID class

Constructs a new OID from string

Parameters

from_path – construct OID from a path (kind/group(s)/id)

to_path()

Converts OID to path

class evaics.sdk.Service

The primary service class

block(prepare=True)

Block the service until terminated

Automatically calls register_signals, mark_ready, mark_terminating (after receiving a termination signal/event)

Optional:

prepare: default: True, if False, register_signals, mark_ready and

mark_terminating must be called manually

create_items(oids)

Create items one-by one

Must be called after the node core is ready

Ignores errors if an item already exists

Parameters

oids – list of item OIDs to create

deploy_items(items)

Create items using deployment payloads in bulk

Must be called after the node core is ready

Payloads are equal to item.deploy eva.core EAPI call

See also https://info.bma.ai/en/actual/eva4/iac.html#items

Parameters

oids – list of items to create

drop_privileges()

Drop service process privileges

get_config()

Get service configuration

init(info=None, on_frame=None, on_rpc_call=None)

Init the service

Automatically calls init_bus, drop_privileges, init_logs and init_rpc (if info specified)

Optional:

info: RPC info

on_frame: bus frame handler

init_bus()

Init the local bus

init_logs()

Initialize service logs

init_rpc(svc_info)

Init bus RPC layer

is_active()

Check is the service active

is_mode_normal()

Is service started in normal mode

is_mode_rtf()

Is service started in react-to-fail mode

is_shutdown_requested()

Check is the service shutdown requested

mark_ready()

Mark the service ready

Automatically logs the service is started if logs are initialized

mark_terminating()

Mark the service terminating

need_ready()

Raises an exception if not ready

RPC helper method which raises an exception if the service is not ready

register_signals()

Register service process system signals

report_accounting_event(u=None, src=None, svc=None, subj=None, oid=None, data=None, note=None, code=None, err=None)

Reports an event into accounting system

subscribe_oids(oids, event_kind='any')

subscribe bus to OID events

Parameters
  • oids – list of OIDs or strings

  • event_kind – any, remote, remote_archive or local

wait_core(timeout=None, wait_forever=True)

Wait until the EVA ICS core is started

class evaics.sdk.ServiceInfo(author='', description='', version='')

Service info helper class

Parameters
  • author – service author

  • description – service description

  • version – service version

add_method(method, description='', required=[], optional=[])

Add a method to service info help

Parameters
  • method – method name

  • description – method description

  • required – list of required param names (strings)

  • optional – list of optional param names

class evaics.sdk.XCall(payload)

HMI X calls helper class

check_op(op)

Check if the session ACL has rights for the operation

Parameters

op – operation code (e.g. “supervisor”)

get_items_allow_deny_reading()

Get allow and deny item list from ACL

is_admin()

Check if the session ACL has admin rights

is_item_readable(oid)

Check if the session ACL has rights to read an item

is_item_writable(oid)

Check if the session ACL has rights to write an item

is_pvt_readable(path)

Check if the session ACL has rights to read a pvt path

is_writable()

Check is the current session writable or read-only

class evaics.sdk.XCallDefault(aci=None, acl=None)

HMI X calls mocker for no ACI/ACL

evaics.sdk.log_traceback()

Log an exception traceback

evaics.sdk.no_rpc_method()

Raise an exception on invalid RPC method

exception evaics.exceptions.AccessDenied(msg='')

raised when a call has no access to the resource

exception evaics.exceptions.FunctionFailed(msg='')

raised when a function failed is failed with any reason

exception evaics.exceptions.GenericException(msg='')
exception evaics.exceptions.InvalidParameter
exception evaics.exceptions.MethodNotImplemented(msg='')

raised when the requested method exists but requested functionality is not implemented

exception evaics.exceptions.ResourceAlreadyExists(msg='')

raised when the requested resource already exists

exception evaics.exceptions.ResourceBusy(msg='')

raised when the requested resource is busy or can not be modified

exception evaics.exceptions.ResourceNotFound(msg='')

raised when the requested resource is not found

exception evaics.exceptions.TimeoutException(msg='')

raised when a call is timed out

Client wrapper

class evaics.client.Client(path='/opt/eva4/var/bus.ipc', **kwargs)

BUS/RT HTTP client class

Automatically connects either via BUS/RT or HTTP

Optional:

path: BUS/RT path or HTTP URI (default: /opt/eva4/var/bus.ipc’)

kwargs passed to client as-is

BUS/RT client (EAPI)

evaics.client.BusClient

alias of Client

class evaics.client.bus.Client(path: str = '/opt/eva4/var/bus.ipc', name: Optional[str] = None, timeout: float = 120)

BUS/RT client for EVA ICS (EAPI)

Create a new BUS/RT client instance

Parameters
  • path – BUS/RT socket (default: /opt/eva4/var/bus.ipc)

  • name – client name (default: PROGRAM.PID)

authenticate()

Authenticate the client

Blank method

bus_call(method: str, params: Optional[dict] = None, target='eva.core')

Call BUS/RT EAPI method

Requires admin permissions

Parameters

method – API method

Optional:

params: API method parameters (dict)

target: target service (default: eva.core)

Returns

API response payload

call(*args, **kwargs)

Alias for bus_call

connect()

Connects the client

test()

Call eva.core test method

Returns

API response payload object

HTTP client

evaics.client.HttpClient

alias of Client

class evaics.client.http.Client(url: str = 'http://localhost:7727', user: Optional[str] = None, password: Optional[str] = None, key: Optional[str] = None, timeout: float = 120)

HTTP client for EVA ICS (HMI HTTP API)

Requires “requests” Python module to be manually installed

Create a new HTTP client instance

Parameters

url – HMI URL (proto://host:port)

api_key(api_key: str)

Authenticate with API key

Parameters

api_key – API key

authenticate()

Authenticate the client

Authenticates the client and stores the authentication token. The method may be called manually but is not mandatory to use

bus_call(method: str, params: Optional[dict] = None, target='eva.core')

Call BUS/RT EAPI method

Requires admin permissions

Parameters

method – API method

Optional:

params: API method parameters (dict)

target: target service (default: eva.core)

Returns

API response payload

call(method: str, params: Optional[dict] = None)

Call server API method

Parameters

method – API method

Optional:

params: API method parameters (dict)

Returns

API response payload

connect()

Blank method, tests HTTP connection only

credentials(user: str, password: str)

Set authentication credentials

Parameters
  • user – user name

  • password – user password

test()

Call server test method

Returns

API response payload object (server and session info)

Cloud client

evaics.client.CloudClient

alias of Client

class evaics.client.cloud.Client(client)

Cloud client for EVA ICS

Create a new Cloud client instance

Parameters

client – HTTP or BUS/RT client

bus_call(method: str, params: Optional[dict] = None, target='eva.core', node=None)

Call BUS/RT EAPI method

Requires admin permissions

Parameters

method – API method

Optional:

params: API method parameters (dict)

target: target service (default: eva.core)

node: target node (.local can be used for the local)

Returns

API response payload

prepare()

Prepare the client

Load node map from the local core, required to be called at least once