Extending HMI with X calls in Python

Contents

• Extending HMI with X calls in Python

  • X call basics

  • Service example

  • Service template

  • HTTP API call example

X call basics

HMI service can be extended with additional calls, called “X” calls.

When a client calls method x::svc_id::method, the following happens:

• the target service gets EAPI calls to the method “x”

• the call contains the following payload:

field	type	description
method	String	the target x-method of the service
params	Any	the target x-method params
aci	Map	ACI (API Call Info) data
acl	Map	the session ACL

• the implemented method must check access control to resources manually if required

• the result (or error) is returned to the client as-is

Service example

Let us create virtual sensors and allow clients to set item data using HTTP RPC calls to HMI, but for sensors only.

Create a virtual sensor, not assigned to real equipment:

eva item create sensor:tests/sensor1

Here is the service code, guided with comments in “X” call processing section. For the general information about services structure in Python, read A simple service in Python:

#!/opt/eva4/venv/bin/python

__version__ = '0.0.1'

import evaics.sdk as sdk
import busrt

from types import SimpleNamespace
from evaics.sdk import pack, unpack, OID, RAW_STATE_TOPIC, XCall

# define a global namespace
_d = SimpleNamespace(service=None)


# RPC calls handler
def handle_rpc(event):
    # handle X calls from HMI
    if event.method == b'x':
        try:
            xp = XCall(unpack(event.get_payload()))
            if xp.method == 'set':
                oid = OID(xp.params['i'])
                status = xp.params['status']
                # check if the item is a sensor
                if oid.kind != 'sensor':
                    raise busrt.rpc.RpcException(
                        'can set state for sensors only',
                        sdk.ERR_CODE_ACCESS_DENIED)
                # check if the caller's session is not a read-only one
                xp.require_writable()
                # check if the caller's ACL has write access to the provided OID
                xp.require_item_write(oid)
                # set the sensor state
                # in this example the service does not check does the sensor
                # really exist in the core or not
                event = dict(status=status)
                if 'value' in xp.params:
                    event['value'] = xp.params['value']
                topic = f'{RAW_STATE_TOPIC}{oid.to_path()}'
                _d.service.bus.send(
                    topic,
                    busrt.client.Frame(pack(event), tp=busrt.client.OP_PUBLISH))
                return
            else:
                sdk.no_rpc_method()
        except busrt.rpc.RpcException as e:
            raise e
        except Exception as e:
            raise busrt.rpc.RpcException(str(e), sdk.ERR_CODE_FUNC_FAILED)
    else:
        sdk.no_rpc_method()


def run():
    info = sdk.ServiceInfo(author='Bohemia Automation',
                           description='Sensor state manipulations',
                           version=__version__)
    service = sdk.Service()
    _d.service = service
    service.init(info, on_rpc_call=handle_rpc)
    service.block()


run()

Service template

The following template can be used to quickly create a service instance with eva-shell:

eva svc create my.svc.sensor_set svc-tpl.yml

command: path/to/eva-svc-example-sensor-set
bus:
  path: var/bus.ipc
config: {}
user: nobody
workers: 1

HTTP API call example

The service responds to the following API calls (httpie call example):

(
cat <<EOF
{
    "jsonrpc": "2.0",
    "id": 1,
    "method": "x::my.svc.sensor_set::set",
    "params": {
        "k": "mykey",
        "i": "sensor:tests/sensor1",
        "status": 1,
        "value": 25
        }
}
EOF
) | http :7727

If using EVA ICS WebEngine, the call can be made as:

eva.call(
    'x::my.svc.sensor_set::set',
    'sensor:tests/sensor1',
    { status: 1, value: 25 }
);