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

Product API

Overview

The ProductClient class is the primary entry point of the Product API.

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

With a ProductClient object, you can:

  • Create, update, query, and delete Products

Examples

Create, query, update, and delete some products

 1from nisystemlink.clients.core import HttpConfiguration
 2from nisystemlink.clients.product import ProductClient
 3from nisystemlink.clients.product.models import (
 4    CreateProductRequest,
 5    ProductField,
 6    ProductOrderBy,
 7    QueryProductsRequest,
 8    QueryProductValuesRequest,
 9)
10
11name = "Example Name"
12family = "Example Family"
13
14
15def create_some_products():
16    """Create two example products on your server."""
17    new_products = [
18        CreateProductRequest(
19            part_number="Example 123 AA",
20            name=name,
21            family=family,
22            keywords=["original keyword"],
23            properties={"original property key": "yes"},
24        ),
25        CreateProductRequest(
26            part_number="Example 123 AA1",
27            name=name,
28            family=family,
29            keywords=["original keyword"],
30            properties={"original property key": "original"},
31        ),
32    ]
33    create_response = client.create_products(new_products)
34    return create_response
35
36
37# Setup the server configuration to point to your instance of SystemLink Enterprise
38server_configuration = HttpConfiguration(
39    server_uri="https://yourserver.yourcompany.com",
40    api_key="YourAPIKeyGeneratedFromSystemLink",
41)
42client = ProductClient(configuration=server_configuration)
43
44# Get all the products using the continuation token in batches of 100 at a time.
45response = client.get_products_paged(take=100, return_count=True)
46all_products = response.products
47while response.continuation_token:
48    response = client.get_products_paged(
49        take=100, continuation_token=response.continuation_token, return_count=True
50    )
51    all_products.extend(response.products)
52
53create_response = create_some_products()
54
55# use get for first product created
56created_product = client.get_product(create_response.products[0].id)
57
58# Query products without continuation
59query_request = QueryProductsRequest(
60    filter=f'family="{family}" && name="{name}"',
61    return_count=True,
62    order_by=ProductOrderBy.FAMILY,
63)
64query_response = client.query_products_paged(query_request)
65
66# Update the first product that you just created and replace the keywords
67updated_product = create_response.products[0]
68updated_product.keywords = ["new keyword"]
69updated_product.properties = {"new property key": "new value"}
70update_response = client.update_products([create_response.products[0]], replace=True)
71
72# Query for just the ids of products that match the family
73values_query = QueryProductValuesRequest(
74    filter=f'family="{family}"', field=ProductField.ID
75)
76values_response = client.query_product_values(query=values_query)
77
78# delete each created product individually by id
79for product in create_response.products:
80    client.delete_product(product.id)
81
82# Create some more and delete them with a single call to delete.
83create_response = create_some_products()
84client.delete_products([product.id for product in create_response.products])

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.

  • Get a specification using an Id.

Examples

Create, Get 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
 71created_response = client.create_specs(CreateSpecificationsRequest(specs=spec_requests))
 72
 73# use get for first spec created
 74if created_response.created_specs and len(created_response.created_specs) > 0:
 75    created_spec = client.get_spec(created_response.created_specs[0].id)
 76
 77# You can query specs based on any field using DynamicLinq syntax.
 78# These are just some representative examples.
 79
 80response = client.query_specs(QuerySpecificationsRequest(product_ids=[product]))
 81all_product_specs = response.specs
 82
 83# Query based on spec id
 84response = client.query_specs(
 85    QuerySpecificationsRequest(product_ids=[product], filter='specId == "spec2"')
 86)
 87if response.specs:
 88    spec2 = response.specs[0]
 89
 90# Query based on name
 91response = client.query_specs(
 92    QuerySpecificationsRequest(product_ids=[product], filter='name.Contains("voltage")')
 93)
 94voltage_specs = response.specs
 95
 96# Query based on Category
 97response = client.query_specs(
 98    QuerySpecificationsRequest(
 99        product_ids=[product], filter='category == "Noise Thresholds"'
100    )
101)
102noise_category = response.specs
103print(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 if spec.id])

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)

Upload a File from disk or memory to SystemLink

"""Example to upload a file to SystemLink."""

import io

from nisystemlink.clients.file import FileClient

client = FileClient()

workspace_id = None  # Upload to default workspace of the auth key

# Upload file from disk
file_path = "path/to/your/file"
with open(file_path, "rb") as fp:
    file_id = client.upload_file(file=fp, workspace=workspace_id)
    print(f"Uploaded file from {file_path} to SystemLink with FileID - {file_id}")

# Upload file-like object from memory
test_file = io.BytesIO(b"This is an example file content.")
test_file.name = "File_From_Memory.txt"  # assign a name to the file object
file_id = client.upload_file(file=test_file)
print(f"Uploaded file from memory to SystemLink with FileID - {file_id}")

Feeds API

Overview

The FeedsClient class is the primary entry point of the Feeds API.

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

With a FeedsClient object, you can:

  • Get the list of feeds, create feed, upload package to feed and delete feed.

Examples

Create a new feed.

 1"""Functionality of creating feeds APIs."""
 2
 3from nisystemlink.clients.core import ApiException, HttpConfiguration
 4from nisystemlink.clients.feeds._feeds_client import FeedsClient
 5from nisystemlink.clients.feeds.models import (
 6    CreateFeedRequest,
 7    Platform,
 8)
 9
10# Update the constants.
11FEED_NAME = ""
12FEED_DESCRIPTION = ""
13PLATFORM = Platform.WINDOWS
14WORKSPACE_ID = (
15    None  # None uses Default workspace. Replace with Systemlink workspace id.
16)
17
18server_url = ""  # SystemLink API URL
19server_api_key = ""  # SystemLink API key
20
21# Provide the valid API key and API URL for client initialization.
22client = FeedsClient(HttpConfiguration(api_key=server_api_key, server_uri=server_url))
23
24# Creating Feeds.
25try:
26    feed_request = CreateFeedRequest(
27        name=FEED_NAME,
28        description=FEED_DESCRIPTION,
29        platform=PLATFORM,
30        workspace=WORKSPACE_ID,
31    )
32    feed_details = client.create_feed(feed=feed_request)
33
34    print("Feed created Successfully.")
35    print(f"Created feed details: {feed_details}")
36
37except ApiException as exp:
38    print(exp)
39except Exception as exp:
40    print(exp)

Query feeds and upload a package to feed.

 1"""Functionality of uploading & querying feeds APIs."""
 2
 3from nisystemlink.clients.core import ApiException, HttpConfiguration
 4from nisystemlink.clients.feeds._feeds_client import FeedsClient
 5from nisystemlink.clients.feeds.models import Platform
 6from nisystemlink.clients.feeds.utilities._get_feed_details import get_feed_by_name
 7
 8# Update the constants.
 9FEED_NAME = ""
10PLATFORM = None
11FEED_DESCRIPTION = ""
12PLATFORM = Platform.WINDOWS
13WORKSPACE_ID = (
14    None  # None uses Default workspace. Replace with Systemlink workspace id.
15)
16PACKAGE_PATH = ""
17
18server_url = ""  # SystemLink API URL
19server_api_key = ""  # SystemLink API key
20
21# Provide the valid API key and API URL for client initialization.
22client = FeedsClient(HttpConfiguration(api_key=server_api_key, server_uri=server_url))
23
24# To upload a package to feed.
25try:
26    # Get ID of the Feed to upload by name
27    feeds = client.query_feeds(platform=PLATFORM, workspace=WORKSPACE_ID)
28    feed = get_feed_by_name(feeds=feeds, name=FEED_NAME)
29    feed_id = feed.id if feed else None
30
31    # Upload the package to Feed by ID
32    if feed_id:
33        client.upload_package(
34            feed_id=feed_id,
35            overwrite=True,
36            package_file_path=PACKAGE_PATH,
37        )
38        print("Package uploaded sucessfully.")
39
40except ApiException as exp:
41    print(exp)
42
43except Exception as exp:
44    print(exp)

Delete a feed.

 1"""Functionality of deleting feed API."""
 2
 3from nisystemlink.clients.core import ApiException, HttpConfiguration
 4from nisystemlink.clients.feeds._feeds_client import FeedsClient
 5from nisystemlink.clients.feeds.models import Platform
 6from nisystemlink.clients.feeds.utilities._get_feed_details import get_feed_by_name
 7
 8# Update the constants.
 9FEED_NAME = ""
10PLATFORM = Platform.WINDOWS
11WORKSPACE_ID = (
12    None  # None uses Default workspace. Replace with Systemlink workspace id.
13)
14
15server_url = ""  # SystemLink API URL
16server_api_key = ""  # SystemLink API key
17
18# Provide the valid API key and API URL for client initialization.
19client = FeedsClient(HttpConfiguration(api_key=server_api_key, server_uri=server_url))
20
21# Deleting Feed.
22try:
23    # Get ID of the Feed to delete by name
24    feeds = client.query_feeds(platform=PLATFORM, workspace=WORKSPACE_ID)
25    feed = get_feed_by_name(feeds=feeds, name=FEED_NAME)
26    feed_id = feed.id if feed else None
27
28    # Delete the Feed by ID
29    if feed_id:
30        client.delete_feed(id=feed_id)
31        print("Feed deleted successfully.")
32
33except ApiException as exp:
34    print(exp)
35except Exception as exp:
36    print(exp)

TestMonitor API (Results and Steps)

Overview

The TestMonitorClient class is the primary entry point of the Test Monitor API used to interact with test results (Results) and test steps (Steps).

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

With a TestMonitorClient object, you can:

  • Create, update, query, and delete results

  • Create, update, query, and delete steps

Examples

Create, query, update, and delete some results

 1from nisystemlink.clients.testmonitor import TestMonitorClient
 2from nisystemlink.clients.testmonitor.models import (
 3    CreateResultRequest,
 4    QueryResultsRequest,
 5    QueryResultValuesRequest,
 6    ResultField,
 7    Status,
 8    StatusType,
 9    UpdateResultRequest,
10)
11
12program_name = "Example Name"
13host_name = "Example Host"
14status_type = StatusType.PASSED
15
16
17def create_some_results():
18    """Create two example results on your server."""
19    new_results = [
20        CreateResultRequest(
21            part_number="Example 123 AA",
22            program_name=program_name,
23            host_name=host_name,
24            status=Status.PASSED(),
25            keywords=["original keyword"],
26            properties={"original property key": "yes"},
27        ),
28        CreateResultRequest(
29            part_number="Example 123 AA1",
30            program_name=program_name,
31            host_name=host_name,
32            status=Status(status_type=StatusType.CUSTOM, status_name="Custom"),
33            keywords=["original keyword"],
34            properties={"original property key": "original"},
35        ),
36    ]
37    create_response = client.create_results(new_results)
38    return create_response
39
40
41# Server configuration is not required when used with SystemLink Client or run through Jupyter on SystemLink
42server_configuration = None
43
44# # Example of setting up the server configuration to point to your instance of SystemLink Enterprise
45# server_configuration = HttpConfiguration(
46#     server_uri="https://yourserver.yourcompany.com",
47#     api_key="YourAPIKeyGeneratedFromSystemLink",
48# )
49
50client = TestMonitorClient(configuration=server_configuration)
51
52create_response = create_some_results()
53
54# Get all the results using the continuation token in batches of 100 at a time.
55response = client.get_results(take=100, return_count=True)
56all_results = response.results
57while response.continuation_token:
58    response = client.get_results(
59        take=100, continuation_token=response.continuation_token, return_count=True
60    )
61    all_results.extend(response.results)
62
63# use get for first result created
64created_result = client.get_result(create_response.results[0].id)
65
66# Query results without continuation
67query_request = QueryResultsRequest(
68    filter=f'status.statusType="{status_type.value}"', return_count=True
69)
70response = client.query_results(query_request)
71
72# Update the first result that you just created and replace the keywords
73updated_result = create_response.results[0]
74updated_result = UpdateResultRequest(
75    id=create_response.results[0].id,
76    keywords=["new keyword"],
77    properties={"new property key": "new value"},
78)
79update_response = client.update_results([updated_result], replace=True)
80
81# Query for just the ids of results that match the family
82values_query = QueryResultValuesRequest(
83    filter=f'programName="{program_name}"', field=ResultField.ID
84)
85values_response = client.query_result_values(query=values_query)
86
87# delete each created result individually by id
88for result in create_response.results:
89    client.delete_result(result.id)
90
91# Create some more and delete them with a single call to delete.
92create_response = create_some_results()
93client.delete_results([result.id for result in create_response.results])

Create, update, query, and delete steps

  1from nisystemlink.clients.testmonitor import TestMonitorClient
  2from nisystemlink.clients.testmonitor.models import (
  3    CreateResultRequest,
  4    CreateStepRequest,
  5    NamedValue,
  6    QueryStepsRequest,
  7    QueryStepValuesRequest,
  8    StepField,
  9    StepIdResultIdPair,
 10    UpdateStepRequest,
 11)
 12from nisystemlink.clients.testmonitor.models._status import Status
 13from nisystemlink.clients.testmonitor.models._step_data import Measurement, StepData
 14
 15
 16def create_test_result():
 17    """Create example result on your server."""
 18    new_results = [
 19        CreateResultRequest(
 20            part_number="Example 123 AA",
 21            program_name="Example Name",
 22            host_name="Example Host",
 23            status=Status.PASSED(),
 24            keywords=["original keyword"],
 25            properties={"original property key": "yes"},
 26        )
 27    ]
 28    create_response = client.create_results(new_results)
 29    return create_response
 30
 31
 32# Server configuration is not required when used with Systemlink Client or run throught Jupyter on SLE
 33server_configuration = None
 34
 35# # Example of setting up the server configuration to point to your instance of SystemLink Enterprise
 36# server_configuration = HttpConfiguration(
 37#     server_uri="https://yourserver.yourcompany.com",
 38#     api_key="YourAPIKeyGeneratedFromSystemLink",
 39# )
 40
 41client = TestMonitorClient(configuration=server_configuration)
 42
 43# create a result to attach the steps to
 44create_response = create_test_result()
 45
 46# Create the step requests
 47result_id = create_response.results[0].result_id
 48step_requests = [
 49    CreateStepRequest(
 50        step_id="step1",
 51        name="step1",
 52        result_id=result_id,
 53        inputs=[
 54            NamedValue(name="Temperature", value="35"),
 55            NamedValue(name="Voltage", value="5"),
 56        ],
 57    ),
 58    CreateStepRequest(
 59        step_id="step2",
 60        name="step2",
 61        result_id=result_id,
 62    ),
 63]
 64
 65# Create the steps
 66create_response = client.create_steps(steps=step_requests)
 67created_steps = create_response.steps
 68print(create_response)
 69
 70# You can query steps based on any field using DynamicLinq syntax.
 71# These are just some representative examples.
 72# Query based on result id
 73query_response = client.query_steps(
 74    QueryStepsRequest(filter=f'resultId == "{result_id}"')
 75)
 76queried_steps = query_response.steps
 77
 78# query step name using query step values
 79query_values_response = client.query_step_values(
 80    QueryStepValuesRequest(
 81        filter=f'resultId == "{result_id}"',
 82        field=StepField.NAME,
 83    )
 84)
 85
 86# update the data of the step
 87# extra properties of the measurements will be converted to string if not already a string
 88update_response = client.update_steps(
 89    steps=[
 90        UpdateStepRequest(
 91            step_id=step.step_id,
 92            result_id=step.result_id,
 93            data=StepData(
 94                text="My output string",
 95                parameters=[
 96                    Measurement(
 97                        name="Temperature",
 98                        status="Passed",
 99                        measurement="35",
100                        lowLimit="30",
101                        highLimit="40",
102                        units="C",
103                        comparisonType="Numeric",
104                        spec_id="spec1",
105                        spec_info={
106                            "specKey": 10
107                        },  # will be converted to string as '{"specKey": 10}'
108                    )
109                ],
110            ),
111        )
112        for step in created_steps
113    ]
114)
115
116# delete all steps at once
117delete_response = client.delete_steps(
118    steps=[
119        StepIdResultIdPair(step_id=step.step_id, result_id=step.result_id)
120        for step in queried_steps
121    ]
122)
123
124create_response = client.create_steps(steps=step_requests)
125created_steps = create_response.steps
126
127# delete steps one by one
128for step in created_steps:
129    if step.step_id and step.result_id:
130        client.delete_step(result_id=step.result_id, step_id=step.step_id)

Notebook API

Overview

The NotebookClient class is the primary entry point of the Notebook API.

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

With a NotebookClient object, you can:

  • Create, update, query, and delete Notebooks

  • Create, get and query Notebook Executions

Examples

Create, query, update, and delete some notebooks.

 1from nisystemlink.clients.core import HttpConfiguration
 2from nisystemlink.clients.notebook import NotebookClient
 3from nisystemlink.clients.notebook.models import NotebookMetadata, QueryNotebookRequest
 4
 5# Setup the server configuration to point to your instance of SystemLink Enterprise
 6server_configuration = HttpConfiguration(
 7    server_uri="https://yourserver.yourcompany.com",
 8    api_key="YourAPIKeyGeneratedFromSystemLink",
 9)
10client = NotebookClient(configuration=server_configuration)
11
12# Create a notebook with metadata and content
13metadata = NotebookMetadata(
14    name="Example Notebook",
15    parameters={"param1": "value1"},
16    properties={"property1": "value1"},
17)
18
19with open("example.ipynb", "rb") as file:
20    notebook_response = client.create_notebook(metadata=metadata, content=file)
21
22# Get the notebook by ID
23notebook = client.get_notebook("your_notebook_id")
24
25# Update the notebook with new metadata and content
26metadata = NotebookMetadata(
27    name="Updated Example Notebook",
28    parameters={"param1": "value2"},
29    properties={"property1": "value2"},
30)
31
32with open("example_updated.ipynb", "rb") as file:
33    notebook_response = client.update_notebook(
34        id="your_notebook_id",
35        metadata=metadata,
36        content=file,
37    )
38
39# Get notebook content by ID
40notebook_content = client.get_notebook_content("your_notebook_id")
41
42# Query notebook by name
43query_request = QueryNotebookRequest(
44    filter='name="Example Notebook"',
45)
46
47query_response = client.query_notebooks(query_request)
48
49# Query notebooks by take
50query_request = QueryNotebookRequest(take=2)
51query_response = client.query_notebooks(query_request)
52
53query_request = QueryNotebookRequest(
54    continuation_token=query_response.continuation_token,
55    take=1,
56)
57query_response = client.query_notebooks(query_request)
58
59# Delete the notebook by ID
60client.delete_notebook("your_notebook_id")

Create, query, retry, and cancel notebook executions.

 1from nisystemlink.clients.core import HttpConfiguration
 2from nisystemlink.clients.notebook import NotebookClient
 3from nisystemlink.clients.notebook.models import (
 4    CreateExecutionRequest,
 5    ExecutionField,
 6    ExecutionPriority,
 7    ExecutionResourceProfile,
 8    ExecutionSortField,
 9    ExecutionStatus,
10    QueryExecutionsRequest,
11    ReportSettings,
12    ReportType,
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 = NotebookClient(configuration=server_configuration)
21
22
23# Create a notebook execution
24execution_request = CreateExecutionRequest(
25    notebook_id="your_notebook_id",
26    parameters={"param1": "value1"},
27    workspace_id="your_workspace_id",
28    timeout=300,
29    result_cache_period=3600,
30    report_settings=ReportSettings(
31        format=ReportType.HTML,
32        exclude_code=False,
33    ),
34    client_requests_id="your_client_request_id",
35    priority=ExecutionPriority.HIGH,
36    resource_profile=ExecutionResourceProfile.DEFAULT,
37)
38
39# Pass the list of execution requests to the create_executions method
40create_execution_response = client.create_executions([execution_request])
41
42# Get the execution by ID
43execution = client.get_execution_by_id("your_execution_id")
44
45# Query executions
46query_request = QueryExecutionsRequest(
47    filter=f"(status = {ExecutionStatus.FAILED.value}))",
48    order_by=ExecutionSortField.COMPLETED_AT,
49    descending=True,
50    projection=[
51        ExecutionField.ID,
52        ExecutionField.NOTEBOOK_ID,
53        ExecutionField.STATUS,
54    ],
55)
56
57query_executions_response = client.query_executions(query_request)
58
59# Cancel execution
60cancel_execution_response = client.cancel_executions(["your_execution_id"])
61
62# Retry execution
63retry_execution_response = client.retry_executions(["your_execution_id"])
64
65# Create execution from existing one
66run_new = client.create_executions_from_existing(["your_execution_id"])

Asset Management API

Overview

The AssetManagementClient class is the primary entry point of the Asset Management API.

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

With a AssetManagementClient object, you can:

  • Create, delete, get the list of assets and link files to assets.

Examples

create, delete, query asset and link files to assets.

 1from nisystemlink.clients.assetmanagement import AssetManagementClient
 2from nisystemlink.clients.assetmanagement.models import (
 3    AssetBusType,
 4    AssetDiscoveryType,
 5    AssetLocationForCreate,
 6    AssetPresence,
 7    AssetPresenceStatus,
 8    AssetType,
 9    CreateAssetRequest,
10    ExternalCalibration,
11    QueryAssetsRequest,
12    SelfCalibration,
13    TemperatureSensor,
14)
15from nisystemlink.clients.core._http_configuration import HttpConfiguration
16
17
18server_configuration = HttpConfiguration(
19    server_uri="https://yourserver.yourcompany.com",
20    api_key="YourAPIKeyGeneratedFromSystemLink",
21)
22client = AssetManagementClient(configuration=server_configuration)
23
24create_assets_request = [
25    CreateAssetRequest(
26        model_number=4000,
27        model_name="NI PXIe-6368",
28        serial_number="01BB877A",
29        vendor_name="NI",
30        vendor_number="4244",
31        bus_type=AssetBusType.ACCESSORY,
32        name="PCISlot2",
33        asset_type=AssetType.DEVICE_UNDER_TEST,
34        firmware_version="A1",
35        hardware_version="12A",
36        visa_resource_name="vs-3144",
37        temperature_sensors=[TemperatureSensor(name="Sensor0", reading=25.8)],
38        supports_self_calibration=True,
39        supports_external_calibration=True,
40        custom_calibration_interval=24,
41        self_calibration=SelfCalibration(
42            temperature_sensors=[TemperatureSensor(name="Sensor0", reading=25.8)],
43            is_limited=False,
44            date="2022-06-07T18:58:05.000Z",
45        ),
46        is_NI_asset=True,
47        workspace="your-workspace-id",
48        location=AssetLocationForCreate(
49            state=AssetPresence(asset_presence=AssetPresenceStatus.PRESENT)
50        ),
51        external_calibration=ExternalCalibration(
52            temperature_sensors=[TemperatureSensor(name="Sensor0", reading=25.8)],
53            date="2022-06-07T18:58:05.000Z",
54            recommended_interval=10,
55            next_recommended_date="2023-11-14T20:42:11.583Z",
56            next_custom_due_date="2024-11-14T20:42:11.583Z",
57            resolved_due_date="2022-06-07T18:58:05.000Z",
58        ),
59        properties={"Key1": "Value1"},
60        keywords=["Keyword1"],
61        discovery_type=AssetDiscoveryType.MANUAL,
62        file_ids=["608a5684800e325b48837c2a"],
63        supports_self_test=True,
64        supports_reset=True,
65        partNumber="A1234 B5",
66    )
67]
68
69# Create an asset.
70create_assets_response = client.create_assets(assets=create_assets_request)
71
72created_asset_id = None
73if create_assets_response.assets and len(create_assets_response.assets) > 0:
74    created_asset_id = str(create_assets_response.assets[0].id)
75
76# Query assets using id.
77query_asset_request = QueryAssetsRequest(
78    ids=[created_asset_id],
79    skip=0,
80    take=1,
81    descending=False,
82    calibratable_only=False,
83    returnCount=False,
84)
85client.query_assets(query=query_asset_request)
86
87# Link files to the created asset.
88file_ids = ["sameple-file-id"]
89if created_asset_id:
90    link_files_response = client.link_files(
91        asset_id=created_asset_id, file_ids=file_ids
92    )
93
94# Delete the created asset.
95if created_asset_id is not None:
96    client.delete_assets(ids=[created_asset_id])

Systems API

Overview

The SystemsClient class is the primary entry point of the Systems API.

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

With a SystemsClient object, you can:

  • Create, query, and remove systems.

Examples

Create, query, and remove some systems.

 1from nisystemlink.clients.core._http_configuration import HttpConfiguration
 2from nisystemlink.clients.systems._systems_client import SystemsClient
 3from nisystemlink.clients.systems.models._create_virtual_systems_request import (
 4    CreateVirtualSystemRequest,
 5)
 6from nisystemlink.clients.systems.models._query_systems_request import (
 7    QuerySystemsRequest,
 8)
 9
10# Setup the server configuration to point to your instance of
11# SystemLink Enterprise.
12server_configuration = HttpConfiguration(
13    server_uri="https://yourserver.yourcompany.com",
14    api_key="YourAPIKeyGeneratedFromSystemLink",
15)
16client = SystemsClient(configuration=server_configuration)
17
18# Systems request metadata.
19create_virtual_system_request: CreateVirtualSystemRequest = CreateVirtualSystemRequest(
20    alias="Python integration virtual system",
21    workspace="your-workspace-id",
22)
23
24# Create a virtual system.
25create_virtual_system_response = client.create_virtual_system(
26    create_virtual_system_request=create_virtual_system_request
27)
28
29minion_id = None
30
31if create_virtual_system_response and create_virtual_system_response.minionId:
32    minion_id = create_virtual_system_response.minionId
33
34# Query systems using id.
35query_systems_request = QuerySystemsRequest(
36    filter=f'id="{minion_id}"', projection="new(id, alias)"
37)
38
39client.query_systems(query=query_systems_request)
40
41# Delete the created systems.
42if minion_id is not None:
43    remove_systems = [minion_id]
44    client.remove_systems(tgt=remove_systems)

TestPlan API

Overview

The TestPlanClient class is the primary entry point of the TestPlan API.

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

With a TestPlanClient object, you can:

  • Create, query, get, update, schedule and delete TestPlans

  • Create, query and delete test plan templates

Examples

Create, query, get, update, schedule and delete TestPlans

  1from datetime import datetime
  2
  3from nisystemlink.clients.core._http_configuration import HttpConfiguration
  4from nisystemlink.clients.test_plan import TestPlanClient
  5from nisystemlink.clients.test_plan.models import (
  6    CreateTestPlanRequest,
  7    Dashboard,
  8    Job,
  9    JobExecution,
 10    ManualExecution,
 11    QueryTestPlansRequest,
 12    ScheduleTestPlanRequest,
 13    ScheduleTestPlansRequest,
 14    UpdateTestPlanRequest,
 15    UpdateTestPlansRequest,
 16)
 17
 18# Setup the server configuration to point to your instance of SystemLink Enterprise
 19server_configuration = HttpConfiguration(
 20    server_uri="https://yourserver.yourcompany.com",
 21    api_key="YourAPIKeyGeneratedFromSystemLink",
 22)
 23client = TestPlanClient(configuration=server_configuration)
 24
 25create_test_plans_request = [
 26    CreateTestPlanRequest(
 27        name="Python integration test plan",
 28        state="NEW",
 29        description="Test plan for verifying integration flow",
 30        assigned_to="test.user@example.com",
 31        estimated_duration_in_seconds=86400,
 32        properties={"env": "staging", "priority": "high"},
 33        part_number="px40482",
 34        dut_id="Sample-Dut_Id",
 35        test_program="TP-Integration-001",
 36        system_filter="os:linux AND arch:x64",
 37        workspace="your_workspace_id",
 38        file_ids_from_template=["file1", "file2"],
 39        dashboard=Dashboard(
 40            id="DashBoardId", variables={"product": "PXIe-4080", "location": "Lab1"}
 41        ),
 42        execution_actions=[
 43            ManualExecution(action="boot", type="MANUAL"),
 44            JobExecution(
 45                action="run",
 46                type="JOB",
 47                jobs=[
 48                    Job(
 49                        functions=["run_test_suite"],
 50                        arguments=[["test_suite.py"]],
 51                        metadata={"env": "staging"},
 52                    )
 53                ],
 54                systemId="system-001",
 55            ),
 56        ],
 57    )
 58]
 59
 60# create a test plan
 61created_test_plans_response = client.create_test_plans(
 62    test_plans=create_test_plans_request
 63)
 64
 65if created_test_plans_response.created_test_plans:
 66    created_test_plan_id = created_test_plans_response.created_test_plans[0].id
 67
 68# Query test plan using id.
 69query_test_plans_request = QueryTestPlansRequest(
 70    skip=0, take=1, descending=False, returnCount=False
 71)
 72client.query_test_plans(query_request=query_test_plans_request)
 73
 74# Get test plan
 75get_test_plan = client.get_test_plan(test_plan_id=created_test_plan_id)
 76
 77# Update test plan
 78update_test_plans_request = UpdateTestPlansRequest(
 79    test_plans=[
 80        UpdateTestPlanRequest(
 81            id=created_test_plan_id,
 82            name="Updated Test Plan",
 83        )
 84    ]
 85)
 86updated_test_plan = client.update_test_plans(update_request=update_test_plans_request)
 87
 88# Schedule the test plan
 89schedule_test_plans_request = ScheduleTestPlansRequest(
 90    test_plans=[
 91        ScheduleTestPlanRequest(
 92            id=created_test_plan_id,
 93            planned_start_date_time=datetime.strptime(
 94                "2025-05-20T15:07:42.527Z", "%Y-%m-%dT%H:%M:%S.%fZ"
 95            ),
 96            estimated_end_date_time=datetime.strptime(
 97                "2025-05-22T15:07:42.527Z", "%Y-%m-%dT%H:%M:%S.%fZ"
 98            ),
 99            system_id="fake-system",
100        )
101    ]
102)
103schedule_test_plan_response = client.schedule_test_plans(
104    schedule_request=schedule_test_plans_request
105)
106
107# Delete test plan
108client.delete_test_plans(ids=[created_test_plan_id])

Create, query and delete test plan templates.

 1from nisystemlink.clients.core._http_configuration import HttpConfiguration
 2from nisystemlink.clients.test_plan import TestPlanClient
 3from nisystemlink.clients.test_plan.models import (
 4    CreateTestPlanTemplateRequest,
 5    Dashboard,
 6    Job,
 7    JobExecution,
 8    ManualExecution,
 9    QueryTestPlanTemplatesRequest,
10)
11
12
13# Setup the server configuration to point to your instance of SystemLink Enterprise
14server_configuration = HttpConfiguration(
15    server_uri="https://yourserver.yourcompany.com",
16    api_key="YourAPIKeyGeneratedFromSystemLink",
17)
18client = TestPlanClient(configuration=server_configuration)
19
20# Test plan template request metadata
21create_test_plan_template_request = [
22    CreateTestPlanTemplateRequest(
23        name="Python integration test plan template",
24        template_group="sample template group",
25        product_families=["FamilyA", "FamilyB"],
26        part_numbers=["PN-1001", "PN-1002"],
27        summary="Template for running integration test plans",
28        description="This template defines execution steps for integration workflows.",
29        test_program="TP-INT-002",
30        estimated_duration_in_seconds=86400,
31        system_filter="os:linux AND arch:x64",
32        execution_actions=[
33            ManualExecution(action="boot", type="MANUAL"),
34            JobExecution(
35                action="run",
36                type="JOB",
37                jobs=[
38                    Job(
39                        functions=["run_test_suite"],
40                        arguments=[["test_suite.py"]],
41                        metadata={"env": "staging"},
42                    )
43                ],
44                systemId="system-001",
45            ),
46        ],
47        file_ids=["file1", "file2"],
48        workspace="your_workspace_id",
49        properties={"env": "staging", "priority": "high"},
50        dashboard=Dashboard(
51            id="DashBoardId", variables={"product": "PXIe-4080", "location": "Lab1"}
52        ),
53    )
54]
55
56# Create a test plan template
57create_test_plan_template_response = client.create_test_plan_templates(
58    test_plan_templates=create_test_plan_template_request
59)
60
61create_test_plan_template_id = None
62
63if (
64    create_test_plan_template_response.created_test_plan_templates
65    and create_test_plan_template_response.created_test_plan_templates[0].id
66):
67    create_test_plan_template_id = str(
68        create_test_plan_template_response.created_test_plan_templates[0].id
69    )
70
71# Query test plan templates using id
72query_test_plan_template_request = QueryTestPlanTemplatesRequest(
73    filter=f'id="{create_test_plan_template_id}"', take=1
74)
75
76client.query_test_plan_templates(
77    query_test_plan_templates=query_test_plan_template_request
78)
79
80# Delete the created test plan template.
81if create_test_plan_template_id is not None:
82    client.delete_test_plan_templates(ids=[create_test_plan_template_id])