Client Methods

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

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()

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

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'])

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()

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()

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()

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)
])

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
])

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()