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
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 |
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 syslib.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.
|
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 {syslib.getcorepath(), syslib.gettime(syslib.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()