Getting Started

Tag API

Overview

The TagManager class is the primary entry point of the Tag API.

When constructing a TagManager, you can pass an HttpConfiguration (like one retrieved from the HttpConfigurationManager), or let TagManager use the default connection. The default connection depends on your SystemLink Client settings.

With a TagManager object, you can:

• Query, create, modify, and delete tags from the server

  • Use open() to get a tag’s TagData from the server when you know the tag’s path.

  • Use query() to get a collection of TagData objects based on the tags’ paths, keywords, and/or properties.

  • Use refresh() to update a list of TagData objects with fresh metadata from the server.

  • Use update() to modify the server metadata for a list of tags, using either TagData objects to overwrite the server’s tag data or TagDataUpdate objects to selectively update specific fields.

  • Use delete() to delete one or more tags from the server.

• Read and write tag values

  • Use read() to get a tag’s current value. Via method parameters, you can also request the timestamp indicating when that value was last written and/or the aggregate data stored for the tag (if the tag’s collect_aggregates attribute is enabled on the server).

  • Use create_writer() to get a BufferedTagWriter that will buffer a set of writes, and automatically send the writes when a given buffer_size is reached or when max_buffer_time has elapsed (or when send_buffered_writes() is called).

• Get a TagSelection that can help perform several of the above operations on several tags at once

  • Use TagManager.create_selection() if you already have a list of TagData objects that you want to perform a set of operations on.

  • Use TagManager.open_selection() if you just have a list of paths – optionally including glob-style wildcards! – with which to create the selection.

If you have a TagSelection, you can use it to create a TagSubscription that will trigger a tag_changed event any time one of the tags’ values is changed.

Examples

Read and write individual tags

from datetime import timedelta

from nisystemlink.clients.tag import DataType, TagManager

mgr = TagManager()
tag = mgr.open("MyTags.Example Tag", DataType.DOUBLE, create=True)

with mgr.create_writer(buffer_size=10, max_buffer_time=timedelta(seconds=3)) as writer:
    writer.write(tag.path, tag.data_type, 3.5)
# Note: Exiting the "with" block automatically calls writer.send_buffered_writes()

read_result = mgr.read(tag.path)
assert read_result is not None
assert read_result.value == 3.5

Subscribe to tag changes

from contextlib import ExitStack
from time import sleep

from nisystemlink.clients.tag import DataType, TagData, TagManager, TagValueReader

SIMULATE_EXTERNAL_TAG_CHANGES = True


def on_tag_changed(tag: TagData, reader: TagValueReader) -> None:
    """Callback for tag_changed events."""
    path = tag.path
    data_type = tag.data_type.name

    if reader is not None:
        read_result = reader.read()
        # A read_result of None means that the tag has no value, but it *must*
        # have a value, because we got a tag_changed event!
        assert read_result is not None
        value = read_result.value
    else:
        value = "???"  # tag has unknown data type

    print(f'Tag changed: "{path}" = {value} ({data_type})')


mgr = TagManager()
if SIMULATE_EXTERNAL_TAG_CHANGES:
    mgr.open("MyTags.Example Tag", DataType.DOUBLE, create=True)
    writer = mgr.create_writer(buffer_size=1)

with ExitStack() as stack:
    # Notes:
    # 1. The tags are assumed to already exist before this example is run, but
    #    setting SIMULATE_EXTERNAL_TAG_CHANGES to True will ensure there is one.
    # 2. Any tags that get added later will NOT automatically appear in the
    #    selection just because the path matches the wildcard used below; you
    #    must call one of the selection's refresh methods to update the tag list
    #    from the server. But even if you do that:
    # 3. The subscription will only contain the tags that were in the selection
    #    when the subscription was created. If you want the subscription to add
    #    new tags that were added to the selection, you must recreate it.
    paths = ["MyTags.*"]
    selection = stack.enter_context(mgr.open_selection(paths))
    if not selection.metadata:
        print(f"Found no tags that match {paths}")
    else:
        print("Matching tags:")
        for path in selection.metadata.keys():
            print(f" - {path}")
        print()

        subscription = stack.enter_context(selection.create_subscription())
        subscription.tag_changed += on_tag_changed

        # Wait forever, until a KeyboardInterrupt (Ctrl+C)
        print("Watching for tag changes; hit Ctrl+C to stop")
        try:
            i = 0
            while True:
                sleep(1)
                if SIMULATE_EXTERNAL_TAG_CHANGES:
                    writer.write("MyTags.Example Tag", DataType.DOUBLE, i)
                    i += 1
        except KeyboardInterrupt:
            pass

DataFrame API

Overview

The DataFrameClient class is the primary entry point of the DataFrame API.

When constructing a DataFrameClient, you can pass an HttpConfiguration (like one retrieved from the HttpConfigurationManager), or let DataFrameClient use the default connection. The default connection depends on your environment.

With a DataFrameClient object, you can:

• Create and delete data tables.

• Modify table metadata and query for tables by their metadata.

• Append rows of data to a table, query for rows of data from a table, and decimate table data.

• Export table data in a comma-separated values (CSV) format.

Examples

Create and write data to a table

import random
from datetime import datetime

from nisystemlink.clients.dataframe import DataFrameClient
from nisystemlink.clients.dataframe.models import (
    AppendTableDataRequest,
    Column,
    ColumnType,
    CreateTableRequest,
    DataFrame,
    DataType,
)

client = DataFrameClient()

# Create table
table_id = client.create_table(
    CreateTableRequest(
        name="Example Table",
        columns=[
            Column(name="ix", data_type=DataType.Int32, column_type=ColumnType.Index),
            Column(name="Float_Column", data_type=DataType.Float32),
            Column(name="Timestamp_Column", data_type=DataType.Timestamp),
        ],
    )
)

# Generate example data
frame = DataFrame(
    data=[[i, random.random(), datetime.now().isoformat()] for i in range(100)]
)

# Write example data to table
client.append_table_data(
    table_id, data=AppendTableDataRequest(frame=frame, endOfData=True)
)

Query and read data from a table

from nisystemlink.clients.dataframe import DataFrameClient
from nisystemlink.clients.dataframe.models import (
    DecimationMethod,
    DecimationOptions,
    QueryDecimatedDataRequest,
)

client = DataFrameClient()

# List a table
response = client.list_tables(take=1)
table = response.tables[0]

# Get table metadata by table id
client.get_table_metadata(table.id)

# Query decimated table data
request = QueryDecimatedDataRequest(
    decimation=DecimationOptions(
        x_column="index",
        y_columns=["col1"],
        intervals=1,
        method=DecimationMethod.MaxMin,
    )
)
client.query_decimated_data(table.id, request)

Export data from a table

from shutil import copyfileobj

from nisystemlink.clients.dataframe import DataFrameClient
from nisystemlink.clients.dataframe.models import (
    ColumnFilter,
    ColumnOrderBy,
    ExportFormat,
    ExportTableDataRequest,
    FilterOperation,
)

client = DataFrameClient()

# List a table
response = client.list_tables(take=1)
table = response.tables[0]

# Export table data with query options
request = ExportTableDataRequest(
    columns=["col1"],
    order_by=[ColumnOrderBy(column="col2", descending=True)],
    filters=[
        ColumnFilter(column="col1", operation=FilterOperation.NotEquals, value="0")
    ],
    response_format=ExportFormat.CSV,
)

data = client.export_table_data(id=table.id, query=request)

# Write the export data to a file
with open(f"{table.name}.csv", "wb") as f:
    copyfileobj(data, f)

# Alternatively, load the export data into a pandas dataframe
# import pandas as pd
# df = pd.read_csv(data)

Spec API

Overview

The SpecClient class is the primary entry point of the Specification Compliance API.

When constructing a SpecClient, you can pass an HttpConfiguration (like one retrieved from the HttpConfigurationManager), or let SpecClient use the default connection. The default connection depends on your environment.

With a SpecClient object, you can:

• Create and delete specifications under a product.

• Modify any fields of an existing specification

• Query for specifications on any fields using DynamicLinq syntax.

Examples

Create and Query Specifications

from nisystemlink.clients.core import HttpConfiguration
from nisystemlink.clients.spec import SpecClient
from nisystemlink.clients.spec.models import (
    Condition,
    ConditionRange,
    ConditionType,
    CreateSpecificationsRequest,
    NumericConditionValue,
    QuerySpecificationsRequest,
    SpecificationDefinition,
    SpecificationLimit,
    SpecificationType,
)

# Setup the server configuration to point to your instance of SystemLink Enterprise
server_configuration = HttpConfiguration(
    server_uri="https://yourserver.yourcompany.com",
    api_key="YourAPIKeyGeneratedFromSystemLink",
)
client = SpecClient(configuration=server_configuration)

# Create the spec requests
product = "Amplifier"
spec_requests = [
    SpecificationDefinition(
        product_id=product,
        spec_id="spec1",
        type=SpecificationType.PARAMETRIC,
        category="Parametric Specs",
        name="output voltage",
        limit=SpecificationLimit(min=1.2, max=1.5),
        unit="mV",
    ),
    SpecificationDefinition(
        product_id=product,
        spec_id="spec2",
        type=SpecificationType.PARAMETRIC,
        category="Parametric Specs",
        name="input voltage",
        limit=SpecificationLimit(min=0.02, max=0.15),
        unit="mV",
        conditions=[
            Condition(
                name="Temperature",
                value=NumericConditionValue(
                    condition_type=ConditionType.NUMERIC,
                    range=[ConditionRange(min=-25, step=20, max=85)],
                    unit="C",
                ),
            ),
            Condition(
                name="Supply Voltage",
                value=NumericConditionValue(
                    condition_type=ConditionType.NUMERIC,
                    discrete=[1.3, 1.5, 1.7],
                    unit="mV",
                ),
            ),
        ],
    ),
    SpecificationDefinition(
        product_id=product,
        spec_id="spec3",
        type=SpecificationType.FUNCTIONAL,
        category="Noise Thresholds",
        name="noise",
    ),
]

# Create the specs on the server
client.create_specs(CreateSpecificationsRequest(specs=spec_requests))

# You can query specs based on any field using DynamicLinq syntax.
# These are just some representative examples.

response = client.query_specs(QuerySpecificationsRequest(product_ids=[product]))
all_product_specs = response.specs

# Query based on spec id
response = client.query_specs(
    QuerySpecificationsRequest(product_ids=[product], filter='specId == "spec2"')
)
if response.specs:
    spec2 = response.specs[0]

# Query based on name
response = client.query_specs(
    QuerySpecificationsRequest(product_ids=[product], filter='name.Contains("voltage")')
)
voltage_specs = response.specs

# Query based on Category
response = client.query_specs(
    QuerySpecificationsRequest(
        product_ids=[product], filter='category == "Noise Thresholds"'
    )
)
noise_category = response.specs
print(noise_category)

Update and Delete Specifications

from nisystemlink.clients.core import HttpConfiguration
from nisystemlink.clients.spec import SpecClient
from nisystemlink.clients.spec.models import (
    QuerySpecificationsRequest,
    SpecificationDefinition,
    SpecificationType,
    UpdateSpecificationsRequest,
)

# Setup the server configuration to point to your instance of SystemLink Enterprise
server_configuration = HttpConfiguration(
    server_uri="https://yourserver.yourcompany.com",
    api_key="YourAPIKeyGeneratedFromSystemLink",
)
client = SpecClient(configuration=server_configuration)

# The query and delete examples assume you have created the specs from the query_specs example
product = "Amplifier"

# update spec1 to change the block to "modifiedBlock"
# query the original spec
response = client.query_specs(
    QuerySpecificationsRequest(product_ids=[product], filter='specId == "spec1"')
)
if response.specs:
    original_spec1 = response.specs[0]
    print(f"Original spec1 block: {original_spec1.block}")
    print(f"Original spec1 version: {original_spec1.version}")

# make the modifications
modified_spec = SpecificationDefinition(
    id=original_spec1.id,
    product_id=original_spec1.product_id,
    spec_id=original_spec1.spec_id,
    type=SpecificationType.FUNCTIONAL,
    keywords=["work", "reviewed"],
    block="modifiedBlock",
    version=original_spec1.version,
    workspace=original_spec1.workspace,
)
update_response = client.update_specs(
    specs=UpdateSpecificationsRequest(specs=[modified_spec])
)
if update_response and update_response.updated_specs:
    print(f"New spec1 version: {update_response.updated_specs[0].version}")

# query again to see new version
response = client.query_specs(
    QuerySpecificationsRequest(product_ids=[product], filter='specId == "spec1"')
)
if response.specs:
    original_spec1 = response.specs[0]
    print(f"Modified spec1 block: {original_spec1.block}")

# delete all the specs for the product
# query all specs
response = client.query_specs(QuerySpecificationsRequest(product_ids=[product]))
if response.specs:
    client.delete_specs(ids=[spec.id for spec in response.specs])