API reference

Contents

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.Registry(service)

Service registry helper class. All key names are relative to eva/svc_data/<service_id>

key_decrement(key)

Decrement key value

Parameters: key – key name
Returns: new key value

key_delete(key)

Delete key

Parameters: key – key name

key_delete_recursive(key)

Delete key and all subkeys

Parameters: key – key name

key_get(key)

Get key value

Parameters: key – key name
Returns: key value

key_get_recursive(key)

Get all key-value pairs

Parameters: key – key name
Returns: list of [key, value] pairs

key_increment(key)

Increment key value

Parameters: key – key name
Returns: new key value

key_set(key, value)

Set key value

Parameters

key – key name
value – key value

class evaics.sdk.Service

The primary service class

Useful varibles:

self.data_path - service data path (None for nobody user)

self.registry - service registry manager (available after init)

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

call(method: str, params: dict | None = None, target='eva.core')

Call BUS/RT EAPI method

Parameters: method – API method

Optional:

params: API method parameters (dict)

target: target service (default: eva.core)

Returns: API response payload

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

Parameters: items – 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: str | None = 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: dict | None = 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: str | None = None, password: str | None = None, key: str | None = 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: dict | None = 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: dict | None = 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: dict | None = 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

Deploy

class evaics.deploy.ACL(id)

ACL class is used to represent an access control list

Parameters: ID (id - the ACL) –

class evaics.deploy.Alarm(group: str, id: str, level: int)

Alarm class is used to represent an alarm

Parameters

group (group - the alarm) –
ID (id - the alarm) –
level (level - the alarm) –

class evaics.deploy.DataObject(name)

DataObject class is used to represent a data object

Parameters: name (name - the data object) –

field(name: str, type: str): Add a field to the data object

fields(fields: list): Specify the fields for the data object

class evaics.deploy.Deploy

Deploy class is used to represent the deploy file

add(node: Node): Add a node to the deploy file

clone(): Clone the Deploy object

merge(other)

Merge another deploy with the current one

Parameters: file (other - the other deploy object or a path to a deploy) –

print(*args, **kwargs): Print the Deploy object to the console (requires PyYAML)

save(path: str): Save the Deploy object to a file (requires PyYAML)

to_dict(): Convert the Deploy object to a dictionary

to_json(): Convert the Deploy object to a JSON string

to_yaml(): Convert the Deploy object to a YAML string (requires PyYAML)

class evaics.deploy.EAPICall(method: str, on='deploy', stage='after'): EAPICall class is used to represent an API call

class evaics.deploy.Element

Basic abstract node element class

clone(): Clone the Element object

set(key: str, value): Set data to the element by key

to_dict(): Convert the Element object to a dictionary

class evaics.deploy.ExtraComamand(on='deploy', stage='after')

class evaics.deploy.Function(function: str, *args, on='deploy', stage='after')

Function class is used to represent a function call

Parameters

name (function - the function) –
arguments (args - the function) –
as-is (passed) –
(default (before/after) – ‘deploy’)
value (stage - the stage) – ‘after’)
(default – ‘after’)

class evaics.deploy.GeneratorSource(name: str, sampling: int)

GeneratorSource class is used to represent a generator source

Parameters

name (name - the generator source) –
frequency (sampling - the generator source sampling) –

params(params: dict): Specify the parameters for the generator source

target(target): Add a target item to the generator source

targets(targets: list): Specify the target items for the generator source

class evaics.deploy.Item(oid)

Item class is used to represent a node item

Parameters: OID (oid - the item) –

action_svc(svc): Specify the action service

action_timeout(timeout): Specify the action timeout

class evaics.deploy.Key(id)

Key class is used to represent an API key

Parameters: ID (id - the key) –

class evaics.deploy.Node(name: str = '.local')

Node class is used to represent a single node in the deploy file.

Parameters: (default (name - the name of the node) – ‘.local’)

add(key: str, value): Add data to the node

add_element(element): Add an element to the node

add_from_export(path: str): Add data from an export file

clone(): Clone the Node object

param(key: str, value): Specify a parameter for the node

set(key: str, value): Set data to the node by key

to_dict(): Convert the Node object to a dictionary

class evaics.deploy.Service(id: str, command: str)

Service class is used to represent a service

Parameters

ID (id - the service) –
command (command - the service) –

config(config: dict): Specify the configuration for the service

user(user: str): Specify the user for the service

workers(workers: int): Specify the number of workers for the service

class evaics.deploy.Upload(src=None, text=None, target=None)

Upload class is used to represent a file upload element

Parameters

URL (src - the source file path or) –
path (target - the target file) –

class evaics.deploy.User(login)

User class is used to represent a user

Parameters: login (login - the user) –

evaics.deploy.load_deploy(path: str): Load a deployment configuration from a path

evaics.deploy.service_from_tpl(id: str, tpl_path): Create a service from a template file (requires PyYAML)