Client methods

On this page, you’ll find all the inmation_api_client.Client class methods, including their description, parameters and example code.

connect_ws

client.connect_ws(url, options)

Connect to the WebSocket server.

Parameters

Name Type Description

url

str

An URL to the WebSocket server

options

Options

Request attributes like username and password

Example

import asyncio
from inmation_api_client import Client, Options

event_loop = ansyncio.get_event_loop()

def create_api_client(io_loop=None):
    client = Client(io_loop)
    event_loop.run_until_complete(client.connect_ws('ws://127.0.0.1:8000/ws', Options('user','pass')))

    return client

client = create_api_client(event_loop)

disconnect_ws

client.disconnect()

Disconnect from the WebSocket server.

exec_function

client.exec_function(context, library_name, function_name, function_arg, cbk, [options])

Execute an existing Lua library function on the system by providing all the required information through the function parameters.

Parameters

Name Type Description

context

dict

Specify the system context, e.g. {"p": "/System/Core/Test/Item1"}

library_name

str

Library Script name

function_name

str

Library function name

function_arg

dict

Library function arguments packed in a dictionary

cbk

function

A callback function

options

Options

Optional options

Example

import asyncio
import json

event_loop = asyncio.get_event_loop()
client = create_api_client(event_loop) # see connect_ws method

async def exec_test():
    context = Item('/System/Core/APIContext/WebAPI01')

    def exec_function_cbk(*args):
        err = args[0]
        if err:
            print(err.message)
        data = args[1]
        print("Result: {}".format(json.dumps(data))

    await client.exec_function(context, 'testlib','test', {'name': 'test'}, exec_function_cbk)

event_loop.run_until_complete(exec_test())

get_ioloop

client.get_ioloop()

Get the asyncio event loop which is attached to this instance.

on_children_count_changed

client.on_children_count_changed(cbk)

Not supported yet.

on_config_version_changed

client.on_config_version_changed(cbk)

Not supported yet.

on_connection

client.on_connection(cbk)

Provide a callback which will be executed when connecting to the WebSocket.

on_data_changed

client.on_data_changed(cbk)

Provide a callback which will be executed every time when data changes for all of the subscribed items.

Parameters

Name Type Description

cbk

function

A callback function

Example

import asyncio

event_loop = asyncio.get_event_loop()
client = create_api_client(event_loop) # see connect_ws method

def on_data_changed(*args):
    _items = args[1]

    if isinstance(_items, list):
        for item in _items:
            item_val = item['v'] if 'v' in item else 'No Value'
            print("{} - {}".format(item['p'], item_val))

client.on_data_changed(on_data_changed)

on_message

client.on_message(cbk)

Provide a callback which will executed when a new message is being received.

on_user_state_changed

client.on_user_state_changed(cbk)

Not supported yet.

on_ws_connection_changed

client.on_ws_connection_changed(cbk)

Provide a callback which will be executed when the connection information changes.

read

client.read(items, cbk, [, options])

Read the VQT data for the given list of items.

Parameters

Name Type Description

items

list(Item)

A list of Item {"p": "/System/Core/Item1"}

cbk

function

A callback function

options

Options

Options object

Example

import asyncio

event_loop = asyncio.get_event_loop()
client = create_api_client(event_loop) # see connect_ws method

async def read_items_at_once(items):
    def read_cbk(*args):
        """read items callback"""
        _items = args[1]
        if isinstance(_items, list):
            print("{} - {}".format('Item', 'Value'))
            for item in _items:
                item_val = item['v'] if 'v' in item else 'No Value'
                print("{} - {}".format(item['p'], item_val))
    await client.read(items, read_cbk)

items_path = "/System/Core/"
items = [Item(items_path + n) for n in ['Item01', 'Item02', 'Item03']]

# Read items
event_loop.run_until_complete(read_items_at_once(items))

read_historical_data

client.read_historical_data(items, start_time, end_time, number_of_intervals, cbk, [options])

Read the historical data for the given list of items.

Parameters

Name Type Description

items

list(HistoricalDataItem)

A list of HistoricalDataItem {"p": "/System/Core/Test/Item1", "aggregate": "AGG_TYPE_RAW"}

start_time

str

Start time in UTC format (ISO 8601)

end_time

str

Start time in UTC format (ISO 8601)

number_of_intervals

int

Number of intervals

cbk

function

A callback function

options

Options

Options object

Example

import asyncio
from datetime import datetime, timedelta
from inmation_api_client import HistoricalDataItem

event_loop = asyncio.get_event_loop()
client = create_api_client(event_loop) # see connect_ws method

now = datetime.now()
now_minus_month = now + timedelta(-30)
format = '%Y-%m-%dT%H:%M:%S.000Z'

start_time = now_minus_month.strftime(format)
end_time = now.strftime(format)

hdi1 = HistoricalDataItem(ITEM1, "AGG_TYPE_RAW")
hdi2 = HistoricalDataItem(ITEM2, "AGG_TYPE_RAW")

def rhd_cbk(*args):
    data = args[1]
    print(data)

event_loop.run_until_complete(client.read_historical_data([hdi1, hdi2], start_time, end_time, 1, rhd_cbk))

read_raw_historical_data

client.read_raw_historical_data(items, start_time, end_time, page_limit, cbk, [options])

Read raw historical data for the given list of items.

Parameters

Name Type Description

items

list(Item)

A list of Item {"p": "/System/Core/Test/Item1"}

start_time

str

Start time in UTC format (ISO 8601)

end_time

str

Start time in UTC format (ISO 8601)

page_limit

int

Number of intervals

cbk

function

A callback function

options

Options

Options object

run_async

client.run_async(tasks)

Run the given list of tasks/coroutines asynchronously.

Parameters

Name Type Description

tasks

list

A list of tasks or coroutines which will be run asynchronously

run_script

client.(context, script, cbk, [options])

Run a Lua code snippet under a specific context.

Parameters

Name Type Description

context

list(Identity)

A list of Identity {"p": "/System/Core/Test/Item1"}

script

str

Lua code snippet

cbk

function

A callback function

options

Options

Options object

Example

import asyncio
from inmation_api_client import Item

event_loop = asyncio.get_event_loop()
client = create_api_client(event_loop) # see connect_ws method

script = "return {inmation.getcorepath(), inmation.gettime(inmation.currenttime())}"
context = Item('/System/Core')
event_loop.run_until_complete(client.run_script(context, script, None))

subscribe

client.subscribe(items, subscription_type, cbk)

Subscribe an item or a list of items to a specific subscription type. Currently, only the SubscriptionType.DataChanged is supported.

Parameters

Name Type Description

items

list(Item)

A list of Item {"p": "/System/Core/Test/Item1"}

subscription_type

str

One of the subscription types, e.g. SubscriptionType.DataChanged

cbk

function

A callback function

Example

import asyncio
from inmation_api_client import Item, SubscriptionType

event_loop = asyncio.get_event_loop()
client = create_api_client(event_loop) # see connect_ws method

async def subscribe_to_data_changes(items):
    def s_cbk(*args):
        _items = args[1]
        if isinstance(_items, list):
            print("{} - {}".format('Item', 'Value'))
            for item in _items:
                item_val = item['v'] if 'v' in item else 'No Value'
                print("{} - {}".format(item['p'], item_val))
    await client.subscribe(items, SubscriptionType.DataChanged, s_cbk)

items_path = "/System/Core/"
items = [Item(items_path + n) for n in ['Item01', 'Item02', 'Item03']]

client.run_async([
    subscribe_to_data_changes(items)
])

unsubscribe

client.unsubscribe(items, subscription_type, cbk)

Unsubscribe an item or a list of items from a specific subscription type. Currently, only the SubscriptionType.DataChanged is supported.

Parameters

Name Type Description

items

list(Item)

A list of Item {"p": "/System/Core/Test/Item1"}

subscription_type

str

One of the subscription types, e.g. SubscriptionType.DataChanged

cbk

function

A callback function

Example

import asyncio
from inmation_api_client import Item, SubscriptionType

io_loop = asyncio.get_event_loop()
client = create_api_client(io_loop) # see connect_ws method

items_path = "/System/Core/"
items = [Item(items_path + i) for i in ['Item01', 'Item02', 'Item03']]

async def subscribe_to_data_changes():
    await client.subscribe(items, SubscriptionType.DataChanged)

async def unsubscribe_from_data_changes():
    await asyncio.sleep(5)
    await client.unsubscribe(items, SubscriptionType.DataChanged)

client.run_async([
    subscribe_to_data_changes(),
    unsubscribe_from_data_changes(), # unsubscribe in 5 seconds
])

write

client.write(items, cbk, [options])

Write VQT data for multiple items.

Parameters

Name Type Description

items

list(ItemValue)

A list of ItemValue {"p": "/System/Core/Test/Item1", "v": 10.5}

cbk

function

A callback function

options

Options

Options object

Example

import asyncio
from random import randint
from inmation_api_client import ItemValue

io_loop = asyncio.get_event_loop()
client = create_api_client(io_loop) # see connect_ws method

async def write_items_at_once(items):
    def write_cbk(*args):
        _items = args[1]
        if isinstance(_items, list):
            print("{} - {}".format('Item', 'Value'))
            for item in _items:
                print("{} - {}".format(item['p'], item['v']))
    await client.write(items, write_cbk)

items_path = "/System/Core/"
items = [Item(items_path + i) for i in ['Item01', 'Item02', 'Item03']]

# Write the items
io_loop.run_until_complete(write_items_at_once(items))

ws_connection_info

client.ws_connection_info()

Provides the following connection information for the WebSocket connection:

  • Session ID

  • Authentication status

  • Connection state (numerical)

  • Connection state (meaning)

  • Ready state