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

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

 1from datetime import timedelta
 2
 3from nisystemlink.clients.tag import DataType, TagManager
 4
 5mgr = TagManager()
 6tag = mgr.open("MyTags.Example Tag", DataType.DOUBLE, create=True)
 7
 8with mgr.create_writer(buffer_size=10, max_buffer_time=timedelta(seconds=3)) as writer:
 9    writer.write(tag.path, tag.data_type, 3.5)
10# Note: Exiting the "with" block automatically calls writer.send_buffered_writes()
11
12read_result = mgr.read(tag.path)
13assert read_result is not None
14assert read_result.value == 3.5

Subscribe to tag changes

 1from contextlib import ExitStack
 2from time import sleep
 3
 4from nisystemlink.clients.tag import DataType, TagData, TagManager, TagValueReader
 5
 6SIMULATE_EXTERNAL_TAG_CHANGES = True
 7
 8
 9def on_tag_changed(tag: TagData, reader: TagValueReader) -> None:
10    """Callback for tag_changed events."""
11    path = tag.path
12    data_type = tag.data_type.name
13
14    if reader is not None:
15        read_result = reader.read()
16        # A read_result of None means that the tag has no value, but it *must*
17        # have a value, because we got a tag_changed event!
18        assert read_result is not None
19        value = read_result.value
20    else:
21        value = "???"  # tag has unknown data type
22
23    print(f'Tag changed: "{path}" = {value} ({data_type})')
24
25
26mgr = TagManager()
27if SIMULATE_EXTERNAL_TAG_CHANGES:
28    mgr.open("MyTags.Example Tag", DataType.DOUBLE, create=True)
29    writer = mgr.create_writer(buffer_size=1)
30
31with ExitStack() as stack:
32    # Notes:
33    # 1. The tags are assumed to already exist before this example is run, but
34    #    setting SIMULATE_EXTERNAL_TAG_CHANGES to True will ensure there is one.
35    # 2. Any tags that get added later will NOT automatically appear in the
36    #    selection just because the path matches the wildcard used below; you
37    #    must call one of the selection's refresh methods to update the tag list
38    #    from the server. But even if you do that:
39    # 3. The subscription will only contain the tags that were in the selection
40    #    when the subscription was created. If you want the subscription to add
41    #    new tags that were added to the selection, you must recreate it.
42    paths = ["MyTags.*"]
43    selection = stack.enter_context(mgr.open_selection(paths))
44    if not selection.metadata:
45        print(f"Found no tags that match {paths}")
46    else:
47        print("Matching tags:")
48        for path in selection.metadata.keys():
49            print(f" - {path}")
50        print()
51
52        subscription = stack.enter_context(selection.create_subscription())
53        subscription.tag_changed += on_tag_changed
54
55        # Wait forever, until a KeyboardInterrupt (Ctrl+C)
56        print("Watching for tag changes; hit Ctrl+C to stop")
57        try:
58            i = 0
59            while True:
60                sleep(1)
61                if SIMULATE_EXTERNAL_TAG_CHANGES:
62                    writer.write("MyTags.Example Tag", DataType.DOUBLE, i)
63                    i += 1
64        except KeyboardInterrupt:
65            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

 1import random
 2from datetime import datetime
 3
 4from nisystemlink.clients.dataframe import DataFrameClient
 5from nisystemlink.clients.dataframe.models import (
 6    AppendTableDataRequest,
 7    Column,
 8    ColumnType,
 9    CreateTableRequest,
10    DataFrame,
11    DataType,
12)
13
14client = DataFrameClient()
15
16# Create table
17table_id = client.create_table(
18    CreateTableRequest(
19        name="Example Table",
20        columns=[
21            Column(name="ix", data_type=DataType.Int32, column_type=ColumnType.Index),
22            Column(name="Float_Column", data_type=DataType.Float32),
23            Column(name="Timestamp_Column", data_type=DataType.Timestamp),
24        ],
25    )
26)
27
28# Generate example data
29frame = DataFrame(
30    data=[[i, random.random(), datetime.now().isoformat()] for i in range(100)]
31)
32
33# Write example data to table
34client.append_table_data(
35    table_id, data=AppendTableDataRequest(frame=frame, endOfData=True)
36)

Query and read data from a table

 1from nisystemlink.clients.dataframe import DataFrameClient
 2from nisystemlink.clients.dataframe.models import (
 3    DecimationMethod,
 4    DecimationOptions,
 5    QueryDecimatedDataRequest,
 6)
 7
 8client = DataFrameClient()
 9
10# List a table
11response = client.list_tables(take=1)
12table = response.tables[0]
13
14# Get table metadata by table id
15client.get_table_metadata(table.id)
16
17# Query decimated table data
18request = QueryDecimatedDataRequest(
19    decimation=DecimationOptions(
20        x_column="index",
21        y_columns=["col1"],
22        intervals=1,
23        method=DecimationMethod.MaxMin,
24    )
25)
26client.query_decimated_data(table.id, request)

Export data from a table

 1from shutil import copyfileobj
 2
 3from nisystemlink.clients.dataframe import DataFrameClient
 4from nisystemlink.clients.dataframe.models import (
 5    ColumnFilter,
 6    ColumnOrderBy,
 7    ExportFormat,
 8    ExportTableDataRequest,
 9    FilterOperation,
10)
11
12client = DataFrameClient()
13
14# List a table
15response = client.list_tables(take=1)
16table = response.tables[0]
17
18# Export table data with query options
19request = ExportTableDataRequest(
20    columns=["col1"],
21    order_by=[ColumnOrderBy(column="col2", descending=True)],
22    filters=[
23        ColumnFilter(column="col1", operation=FilterOperation.NotEquals, value="0")
24    ],
25    response_format=ExportFormat.CSV,
26)
27
28data = client.export_table_data(id=table.id, query=request)
29
30# Write the export data to a file
31with open(f"{table.name}.csv", "wb") as f:
32    copyfileobj(data, f)
33
34# Alternatively, load the export data into a pandas dataframe
35# import pandas as pd
36# 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

 1from nisystemlink.clients.core import HttpConfiguration
 2from nisystemlink.clients.spec import SpecClient
 3from nisystemlink.clients.spec.models import (
 4    Condition,
 5    ConditionRange,
 6    ConditionType,
 7    CreateSpecificationsRequest,
 8    NumericConditionValue,
 9    QuerySpecificationsRequest,
10    SpecificationDefinition,
11    SpecificationLimit,
12    SpecificationType,
13)
14
15# Setup the server configuration to point to your instance of SystemLink Enterprise
16server_configuration = HttpConfiguration(
17    server_uri="https://yourserver.yourcompany.com",
18    api_key="YourAPIKeyGeneratedFromSystemLink",
19)
20client = SpecClient(configuration=server_configuration)
21
22# Create the spec requests
23product = "Amplifier"
24spec_requests = [
25    SpecificationDefinition(
26        product_id=product,
27        spec_id="spec1",
28        type=SpecificationType.PARAMETRIC,
29        category="Parametric Specs",
30        name="output voltage",
31        limit=SpecificationLimit(min=1.2, max=1.5),
32        unit="mV",
33    ),
34    SpecificationDefinition(
35        product_id=product,
36        spec_id="spec2",
37        type=SpecificationType.PARAMETRIC,
38        category="Parametric Specs",
39        name="input voltage",
40        limit=SpecificationLimit(min=0.02, max=0.15),
41        unit="mV",
42        conditions=[
43            Condition(
44                name="Temperature",
45                value=NumericConditionValue(
46                    condition_type=ConditionType.NUMERIC,
47                    range=[ConditionRange(min=-25, step=20, max=85)],
48                    unit="C",
49                ),
50            ),
51            Condition(
52                name="Supply Voltage",
53                value=NumericConditionValue(
54                    condition_type=ConditionType.NUMERIC,
55                    discrete=[1.3, 1.5, 1.7],
56                    unit="mV",
57                ),
58            ),
59        ],
60    ),
61    SpecificationDefinition(
62        product_id=product,
63        spec_id="spec3",
64        type=SpecificationType.FUNCTIONAL,
65        category="Noise Thresholds",
66        name="noise",
67    ),
68]
69
70# Create the specs on the server
71client.create_specs(CreateSpecificationsRequest(specs=spec_requests))
72
73# You can query specs based on any field using DynamicLinq syntax.
74# These are just some representative examples.
75
76response = client.query_specs(QuerySpecificationsRequest(product_ids=[product]))
77all_product_specs = response.specs
78
79# Query based on spec id
80response = client.query_specs(
81    QuerySpecificationsRequest(product_ids=[product], filter='specId == "spec2"')
82)
83if response.specs:
84    spec2 = response.specs[0]
85
86# Query based on name
87response = client.query_specs(
88    QuerySpecificationsRequest(product_ids=[product], filter='name.Contains("voltage")')
89)
90voltage_specs = response.specs
91
92# Query based on Category
93response = client.query_specs(
94    QuerySpecificationsRequest(
95        product_ids=[product], filter='category == "Noise Thresholds"'
96    )
97)
98noise_category = response.specs
99print(noise_category)

Update and Delete Specifications

 1from nisystemlink.clients.core import HttpConfiguration
 2from nisystemlink.clients.spec import SpecClient
 3from nisystemlink.clients.spec.models import (
 4    QuerySpecificationsRequest,
 5    SpecificationDefinition,
 6    SpecificationType,
 7    UpdateSpecificationsRequest,
 8)
 9
10# Setup the server configuration to point to your instance of SystemLink Enterprise
11server_configuration = HttpConfiguration(
12    server_uri="https://yourserver.yourcompany.com",
13    api_key="YourAPIKeyGeneratedFromSystemLink",
14)
15client = SpecClient(configuration=server_configuration)
16
17# The query and delete examples assume you have created the specs from the query_specs example
18product = "Amplifier"
19
20# update spec1 to change the block to "modifiedBlock"
21# query the original spec
22response = client.query_specs(
23    QuerySpecificationsRequest(product_ids=[product], filter='specId == "spec1"')
24)
25if response.specs:
26    original_spec1 = response.specs[0]
27    print(f"Original spec1 block: {original_spec1.block}")
28    print(f"Original spec1 version: {original_spec1.version}")
29
30# make the modifications
31modified_spec = SpecificationDefinition(
32    id=original_spec1.id,
33    product_id=original_spec1.product_id,
34    spec_id=original_spec1.spec_id,
35    type=SpecificationType.FUNCTIONAL,
36    keywords=["work", "reviewed"],
37    block="modifiedBlock",
38    version=original_spec1.version,
39    workspace=original_spec1.workspace,
40)
41update_response = client.update_specs(
42    specs=UpdateSpecificationsRequest(specs=[modified_spec])
43)
44if update_response and update_response.updated_specs:
45    print(f"New spec1 version: {update_response.updated_specs[0].version}")
46
47# query again to see new version
48response = client.query_specs(
49    QuerySpecificationsRequest(product_ids=[product], filter='specId == "spec1"')
50)
51if response.specs:
52    original_spec1 = response.specs[0]
53    print(f"Modified spec1 block: {original_spec1.block}")
54
55# delete all the specs for the product
56# query all specs
57response = client.query_specs(QuerySpecificationsRequest(product_ids=[product]))
58if response.specs:
59    client.delete_specs(ids=[spec.id for spec in response.specs])

File API

Overview

The FileClient class is the primary entry point of the File API.

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

With a FileClient object, you can:

  • Get the list of files, download and delete files

Examples

Get the metadata of a File using its Id and download it.

 1"""Example to download a file from SystemLink."""
 2
 3from shutil import copyfileobj
 4
 5from nisystemlink.clients.file import FileClient
 6
 7client = FileClient()
 8
 9file_id = "a55adc7f-5068-4202-9d70-70ca6a06bee9"
10
11# Fetch the file metadata to get the name
12files = client.get_files(ids=[file_id])
13
14if not files.available_files:
15    raise Exception(f"File ID {file_id} not found.")
16
17
18file_name = "Untitled"
19
20file_properties = files.available_files[0].properties
21
22if file_properties:
23    file_name = file_properties["Name"]
24
25# Download the file using FileId with content inline
26content = client.download_file(id=file_id)
27
28# Write the content to a file
29with open(file_name, "wb") as f:
30    copyfileobj(content, f)