Client Methods

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

Connect

client.Connect(host, port, options)

Connect to the WebSocket server.

Parameters

Name Type Description

host

str

An IP address or a host

port

int

The port number of the Web API

options

Options

Request attributes like authentication or authorization information

Examples

The following example shows how to connect to the system:inmation.

import asyncio
from inmation_api_client import Client, Options


def create_api_client(event_loop=None):
    client = Client(event_loop)

    def connection_changed(conn_info):
        print('Connection state: {}, authenticated: {}'.format(conn_info.state_string, conn_info.authenticated))
    client.OnConnectionChanged(connection_changed)

    def on_error(err):
        if err:
            print("Error: {}".format(err.message))
    client.OnError(on_error)

    client.Connect('127.0.0.1', 8002, Options({
        'auth': {
            'username': 'user',
            'password': 'pass'
        }
    }))

    return client

Disconnect

client.Disconnect()

Disconnect from the WebSocket server.

ExecuteFunction

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

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

The ExecuteFunctionAsync method which has the same signature is available to be used in an asynchronous context.

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

options

Options

Optional options

Examples

import asyncio
import json
from inmation_api_client import

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

# Async example
async def exec_function_async():
    context = Item('/System/Core/APIContext/WebAPI01')
    response = await client.ExecuteFunctionAsync(context, 'testlib','test')
    print(response['data'])

event_loop.run_until_complete(exec_function_async())

# Sync example
def exec_function():
    context = Item('/System/Core/APIContext/WebAPI01')
    response = client.ExecuteFunction(context, 'testlib', 'test')

exec_function()

Mass

client.Mass(entries, [options])

The MassAsync method with the same signature is available to be used in an asynchronous context.

Parameters

Name Type Description

entries

list(dict)

A list of dict (mass entries)

options

Options

Options object with optional fields or batch_flags attributes

The options parameter can be extended with either the fields or batch_flags attributes.

from inmation_api_client import Options

opt = Options()
opt.fields = ['ALL']  # return all fields

# OR

opt.batch_flags = 4  # SUPPRESS_RESULTS

# and then supply the `opt` as a second argument to the client.Mass() method
  • See the Mass endpoint in the Web API docs for more details.

  • Also refer to the Lua inmation.mass() function to get more insight on how to create mass entries.

Examples

import asyncio

event_loop = asyncio.get_event_loop()
client = create_api_client(event_loop)  # See client.Connect() method

mass_entries_1 = [
    {
        'path': '/System/Core/TestFolder1',
        'operation': 'UPSERT',
        'class': 'GenFolder',
        'ObjectName': 'TestFolder1'
    }
]

# Mass Async - create GenFolder items
async def mass_async():
    response = await client.MassAsync(mass_entries_1)
    print(response['data'])
event_loop.run_until_complete(mass_async())

mass_entries_2 = [
    {
        'path': '/System/Core/TestFolder2',
        'operation': 'UPSERT',
        'class': 'GenFolder',
        'ObjectName': 'TestFolder2'
    }
]

# Mass - create GenFolder items
response = client.Mass(mass_entries_2)
print(response['data'])

Read

client.Read(items, [options])

Read the VQT data for the given list of Item items.

The ReadAsync method with the same signature is available to be used in an asynchronous context.

Parameters

Name Type Description

items

list(Item)

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

options

Options

Options object

Examples

import asyncio

from inmation_api_client import Item

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

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

# Read async
async def read_async():
    response = await client.ReadAsync(items)
    print(response['data'])

event_loop.run_until_complete(read_async())


# Read
def read():
    response = client.Read(items)
    print(response['data'])

read()

ReadHistoricalData

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

Read the historical data for the given list of HistoricalDataItem items.

The ReadHistoricalDataAsync method with the same signature is available to be used in an asynchronous context.

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

options

Options

Options object

Examples

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 client.Connect() 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)

# Make sure you have some items with historical data
hdi1 = HistoricalDataItem('Item01', "AGG_TYPE_RAW")
hdi2 = HistoricalDataItem('Item02', "AGG_TYPE_RAW")

items = [hdi1, hdi2]

# Async example
async def read_historical_data_async():
    response = await client.ReadHistoricalDataAsync(items, start_time, end_time, 1)
    print(response['data'])

event_loop.run_until_complete(read_historical_data_async())

# Sync example
def read_historical_data():
    response = client.ReadHistoricalData(items, start_time, end_time, 1)
    print(response['data'])

read_historical_data()

ReadRawHistoricalData

client.ReadRawHistoricalData(queries, [options])

Read raw historical data for the given list of RawHistoricalDataQuery items.

The ReadRawHistoricalDataAsync method with the same signature is available to be used in an asynchronous context.

Parameters

Name Type Description

queries

list(RawHistoricalDataQuery)

A list of RawHistoricalDataQuery

options

Options

Options object

Examples

RunScript

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

Run a Lua code snippet under a specific context.

The RunScriptAsync method with the same signature is available to be used in an asynchronous context.

Parameters

Name Type Description

context

list(Identity)

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

script

str

Lua code snippet

options

Options

Options object

Examples

import asyncio
from inmation_api_client import Item

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

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


# Run script async
async def run_script_async():
    response = await client.RunScriptAsync(context, script)
    print(response['data'])

event_loop.run_until_complete(run_script_async())


# Run script
def run_script():
    response = client.RunScript(context, script)
    print(response['data'])

run_script()

SubscribeAsync

client.SubscribeAsync(items, subscription_type, cbk)

Subscribe an item or a list of items to a specific subscription type.

At the moment, 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

Examples

import asyncio
from inmation_api_client import Item, SubscriptionType

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

# Provide a callback to the client.OnDataChanged() in order to get notified
# when values change for the subscribed items.
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.OnDataChanged(on_data_changed)

async def subscribe_to_data_changes(items):
    def s_cbk(*args):
        '''Subscription callback, executed only once during the subscription'''
        _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.SubscribeAsync(items, SubscriptionType.DataChanged, s_cbk)

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

client.RunAsync([
    subscribe_to_data_changes(items)
])

UnsubscribeAsync

client.UnsubscribeAsync(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

Examples

import asyncio
from inmation_api_client import Item, SubscriptionType

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

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


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


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

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

Write

client.Write(items, [options])

Write VQT data to one or multiple items.

The WriteAsync method with the same signature is available to be used in an asynchronous context.

Parameters

Name Type Description

items

list(ItemValue)

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

options

Options

Options object

Examples

import asyncio
from random import randint
from inmation_api_client import Item, ItemValue

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

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

# Write async
async def write_async():
    response = await client.WriteAsync(items)
    print(response['data'])

io_loop.run_until_complete(write_async())


# Write
def write():
    response = client.Write(items)
    print(response['data'])

write()