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

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

from nisystemlink.clients.core import HttpConfiguration
from nisystemlink.clients.product import ProductClient
from nisystemlink.clients.product.models import (
    CreateProductRequest,
    ProductField,
    ProductOrderBy,
    QueryProductsRequest,
    QueryProductValuesRequest,
)

name = "Example Name"
family = "Example Family"


def create_some_products():
    """Create two example products on your server."""
    new_products = [
        CreateProductRequest(
            part_number="Example 123 AA",
            name=name,
            family=family,
            keywords=["original keyword"],
            properties={"original property key": "yes"},
        ),
        CreateProductRequest(
            part_number="Example 123 AA1",
            name=name,
            family=family,
            keywords=["original keyword"],
            properties={"original property key": "original"},
        ),
    ]
    create_response = client.create_products(new_products)
    return create_response


# 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 = ProductClient(configuration=server_configuration)

# Get all the products using the continuation token in batches of 100 at a time.
response = client.get_products_paged(take=100, return_count=True)
all_products = response.products
while response.continuation_token:
    response = client.get_products_paged(
        take=100, continuation_token=response.continuation_token, return_count=True
    )
    all_products.extend(response.products)

create_response = create_some_products()

# use get for first product created
created_product = client.get_product(create_response.products[0].id)

# Query products without continuation
query_request = QueryProductsRequest(
    filter=f'family="{family}" && name="{name}"',
    return_count=True,
    order_by=ProductOrderBy.FAMILY,
)
query_response = client.query_products_paged(query_request)

# Update the first product that you just created and replace the keywords
updated_product = create_response.products[0]
updated_product.keywords = ["new keyword"]
updated_product.properties = {"new property key": "new value"}
update_response = client.update_products([create_response.products[0]], replace=True)

# Query for just the ids of products that match the family
values_query = QueryProductValuesRequest(
    filter=f'family="{family}"', field=ProductField.ID
)
values_response = client.query_product_values(query=values_query)

# delete each created product individually by id
for product in create_response.products:
    client.delete_product(product.id)

# Create some more and delete them with a single call to delete.
create_response = create_some_products()
client.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

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.

• Get a specification using an Id.

Examples

Create, Get 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
created_response = client.create_specs(CreateSpecificationsRequest(specs=spec_requests))

# use get for first spec created
if created_response.created_specs and len(created_response.created_specs) > 0:
    created_spec = client.get_spec(created_response.created_specs[0].id)

# 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 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.

"""Example to download a file from SystemLink."""

from shutil import copyfileobj

from nisystemlink.clients.file import FileClient

client = FileClient()

file_id = "a55adc7f-5068-4202-9d70-70ca6a06bee9"

# Fetch the file metadata to get the name
files = client.get_files(ids=[file_id])

if not files.available_files:
    raise Exception(f"File ID {file_id} not found.")


file_name = "Untitled"

file_properties = files.available_files[0].properties

if file_properties:
    file_name = file_properties["Name"]

# Download the file using FileId with content inline
content = client.download_file(id=file_id)

# Write the content to a file
with open(file_name, "wb") as f:
    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.

"""Functionality of creating feeds APIs."""

from nisystemlink.clients.core import ApiException, HttpConfiguration
from nisystemlink.clients.feeds._feeds_client import FeedsClient
from nisystemlink.clients.feeds.models import (
    CreateFeedRequest,
    Platform,
)

# Update the constants.
FEED_NAME = ""
FEED_DESCRIPTION = ""
PLATFORM = Platform.WINDOWS
WORKSPACE_ID = (
    None  # None uses Default workspace. Replace with Systemlink workspace id.
)

server_url = ""  # SystemLink API URL
server_api_key = ""  # SystemLink API key

# Provide the valid API key and API URL for client initialization.
client = FeedsClient(HttpConfiguration(api_key=server_api_key, server_uri=server_url))

# Creating Feeds.
try:
    feed_request = CreateFeedRequest(
        name=FEED_NAME,
        description=FEED_DESCRIPTION,
        platform=PLATFORM,
        workspace=WORKSPACE_ID,
    )
    feed_details = client.create_feed(feed=feed_request)

    print("Feed created Successfully.")
    print(f"Created feed details: {feed_details}")

except ApiException as exp:
    print(exp)
except Exception as exp:
    print(exp)

Query feeds and upload a package to feed.

"""Functionality of uploading & querying feeds APIs."""

from nisystemlink.clients.core import ApiException, HttpConfiguration
from nisystemlink.clients.feeds._feeds_client import FeedsClient
from nisystemlink.clients.feeds.models import Platform
from nisystemlink.clients.feeds.utilities._get_feed_details import get_feed_by_name

# Update the constants.
FEED_NAME = ""
PLATFORM = None
FEED_DESCRIPTION = ""
PLATFORM = Platform.WINDOWS
WORKSPACE_ID = (
    None  # None uses Default workspace. Replace with Systemlink workspace id.
)
PACKAGE_PATH = ""

server_url = ""  # SystemLink API URL
server_api_key = ""  # SystemLink API key

# Provide the valid API key and API URL for client initialization.
client = FeedsClient(HttpConfiguration(api_key=server_api_key, server_uri=server_url))

# To upload a package to feed.
try:
    # Get ID of the Feed to upload by name
    feeds = client.query_feeds(platform=PLATFORM, workspace=WORKSPACE_ID)
    feed = get_feed_by_name(feeds=feeds, name=FEED_NAME)
    feed_id = feed.id if feed else None

    # Upload the package to Feed by ID
    if feed_id:
        client.upload_package(
            feed_id=feed_id,
            overwrite=True,
            package_file_path=PACKAGE_PATH,
        )
        print("Package uploaded sucessfully.")

except ApiException as exp:
    print(exp)

except Exception as exp:
    print(exp)

Delete a feed.

"""Functionality of deleting feed API."""

from nisystemlink.clients.core import ApiException, HttpConfiguration
from nisystemlink.clients.feeds._feeds_client import FeedsClient
from nisystemlink.clients.feeds.models import Platform
from nisystemlink.clients.feeds.utilities._get_feed_details import get_feed_by_name

# Update the constants.
FEED_NAME = ""
PLATFORM = Platform.WINDOWS
WORKSPACE_ID = (
    None  # None uses Default workspace. Replace with Systemlink workspace id.
)

server_url = ""  # SystemLink API URL
server_api_key = ""  # SystemLink API key

# Provide the valid API key and API URL for client initialization.
client = FeedsClient(HttpConfiguration(api_key=server_api_key, server_uri=server_url))

# Deleting Feed.
try:
    # Get ID of the Feed to delete by name
    feeds = client.query_feeds(platform=PLATFORM, workspace=WORKSPACE_ID)
    feed = get_feed_by_name(feeds=feeds, name=FEED_NAME)
    feed_id = feed.id if feed else None

    # Delete the Feed by ID
    if feed_id:
        client.delete_feed(id=feed_id)
        print("Feed deleted successfully.")

except ApiException as exp:
    print(exp)
except Exception as exp:
    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

from nisystemlink.clients.testmonitor import TestMonitorClient
from nisystemlink.clients.testmonitor.models import (
    CreateResultRequest,
    QueryResultsRequest,
    QueryResultValuesRequest,
    ResultField,
    Status,
    StatusType,
    UpdateResultRequest,
)

program_name = "Example Name"
host_name = "Example Host"
status_type = StatusType.PASSED


def create_some_results():
    """Create two example results on your server."""
    new_results = [
        CreateResultRequest(
            part_number="Example 123 AA",
            program_name=program_name,
            host_name=host_name,
            status=Status.PASSED(),
            keywords=["original keyword"],
            properties={"original property key": "yes"},
        ),
        CreateResultRequest(
            part_number="Example 123 AA1",
            program_name=program_name,
            host_name=host_name,
            status=Status(status_type=StatusType.CUSTOM, status_name="Custom"),
            keywords=["original keyword"],
            properties={"original property key": "original"},
        ),
    ]
    create_response = client.create_results(new_results)
    return create_response


# Server configuration is not required when used with SystemLink Client or run through Jupyter on SystemLink
server_configuration = None

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

client = TestMonitorClient(configuration=server_configuration)

create_response = create_some_results()

# Get all the results using the continuation token in batches of 100 at a time.
response = client.get_results(take=100, return_count=True)
all_results = response.results
while response.continuation_token:
    response = client.get_results(
        take=100, continuation_token=response.continuation_token, return_count=True
    )
    all_results.extend(response.results)

# use get for first result created
created_result = client.get_result(create_response.results[0].id)

# Query results without continuation
query_request = QueryResultsRequest(
    filter=f'status.statusType="{status_type.value}"', return_count=True
)
response = client.query_results(query_request)

# Update the first result that you just created and replace the keywords
updated_result = create_response.results[0]
updated_result = UpdateResultRequest(
    id=create_response.results[0].id,
    keywords=["new keyword"],
    properties={"new property key": "new value"},
)
update_response = client.update_results([updated_result], replace=True)

# Query for just the ids of results that match the family
values_query = QueryResultValuesRequest(
    filter=f'programName="{program_name}"', field=ResultField.ID
)
values_response = client.query_result_values(query=values_query)

# delete each created result individually by id
for result in create_response.results:
    client.delete_result(result.id)

# Create some more and delete them with a single call to delete.
create_response = create_some_results()
client.delete_results([result.id for result in create_response.results])

Create, update, query, and delete steps

from nisystemlink.clients.testmonitor import TestMonitorClient
from nisystemlink.clients.testmonitor.models import (
    CreateResultRequest,
    CreateStepRequest,
    NamedValue,
    QueryStepsRequest,
    QueryStepValuesRequest,
    StepField,
    StepIdResultIdPair,
    UpdateStepRequest,
)
from nisystemlink.clients.testmonitor.models._status import Status
from nisystemlink.clients.testmonitor.models._step_data import Measurement, StepData


def create_test_result():
    """Create example result on your server."""
    new_results = [
        CreateResultRequest(
            part_number="Example 123 AA",
            program_name="Example Name",
            host_name="Example Host",
            status=Status.PASSED(),
            keywords=["original keyword"],
            properties={"original property key": "yes"},
        )
    ]
    create_response = client.create_results(new_results)
    return create_response


# Server configuration is not required when used with Systemlink Client or run throught Jupyter on SLE
server_configuration = None

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

client = TestMonitorClient(configuration=server_configuration)

# create a result to attach the steps to
create_response = create_test_result()

# Create the step requests
result_id = create_response.results[0].result_id
step_requests = [
    CreateStepRequest(
        step_id="step1",
        name="step1",
        result_id=result_id,
        inputs=[
            NamedValue(name="Temperature", value="35"),
            NamedValue(name="Voltage", value="5"),
        ],
    ),
    CreateStepRequest(
        step_id="step2",
        name="step2",
        result_id=result_id,
    ),
]

# Create the steps
create_response = client.create_steps(steps=step_requests)
created_steps = create_response.steps
print(create_response)

# You can query steps based on any field using DynamicLinq syntax.
# These are just some representative examples.
# Query based on result id
query_response = client.query_steps(
    QueryStepsRequest(filter=f'resultId == "{result_id}"')
)
queried_steps = query_response.steps

# query step name using query step values
query_values_response = client.query_step_values(
    QueryStepValuesRequest(
        filter=f'resultId == "{result_id}"',
        field=StepField.NAME,
    )
)

# update the data of the step
# extra properties of the measurements will be converted to string if not already a string
update_response = client.update_steps(
    steps=[
        UpdateStepRequest(
            step_id=step.step_id,
            result_id=step.result_id,
            data=StepData(
                text="My output string",
                parameters=[
                    Measurement(
                        name="Temperature",
                        status="Passed",
                        measurement="35",
                        lowLimit="30",
                        highLimit="40",
                        units="C",
                        comparisonType="Numeric",
                        spec_id="spec1",
                        spec_info={
                            "specKey": 10
                        },  # will be converted to string as '{"specKey": 10}'
                    )
                ],
            ),
        )
        for step in created_steps
    ]
)

# delete all steps at once
delete_response = client.delete_steps(
    steps=[
        StepIdResultIdPair(step_id=step.step_id, result_id=step.result_id)
        for step in queried_steps
    ]
)

create_response = client.create_steps(steps=step_requests)
created_steps = create_response.steps

# delete steps one by one
for step in created_steps:
    if step.step_id and step.result_id:
        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.

from nisystemlink.clients.core import HttpConfiguration
from nisystemlink.clients.notebook import NotebookClient
from nisystemlink.clients.notebook.models import NotebookMetadata, QueryNotebookRequest

# 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 = NotebookClient(configuration=server_configuration)

# Create a notebook with metadata and content
metadata = NotebookMetadata(
    name="Example Notebook",
    parameters={"param1": "value1"},
    properties={"property1": "value1"},
)

with open("example.ipynb", "rb") as file:
    notebook_response = client.create_notebook(metadata=metadata, content=file)

# Get the notebook by ID
notebook = client.get_notebook("your_notebook_id")

# Update the notebook with new metadata and content
metadata = NotebookMetadata(
    name="Updated Example Notebook",
    parameters={"param1": "value2"},
    properties={"property1": "value2"},
)

with open("example_updated.ipynb", "rb") as file:
    notebook_response = client.update_notebook(
        id="your_notebook_id",
        metadata=metadata,
        content=file,
    )

# Get notebook content by ID
notebook_content = client.get_notebook_content("your_notebook_id")

# Query notebook by name
query_request = QueryNotebookRequest(
    filter='name="Example Notebook"',
)

query_response = client.query_notebooks(query_request)

# Query notebooks by take
query_request = QueryNotebookRequest(take=2)
query_response = client.query_notebooks(query_request)

query_request = QueryNotebookRequest(
    continuation_token=query_response.continuation_token,
    take=1,
)
query_response = client.query_notebooks(query_request)

# Delete the notebook by ID
client.delete_notebook("your_notebook_id")

Create, query, retry, and cancel notebook executions.

from nisystemlink.clients.core import HttpConfiguration
from nisystemlink.clients.notebook import NotebookClient
from nisystemlink.clients.notebook.models import (
    CreateExecutionRequest,
    ExecutionField,
    ExecutionPriority,
    ExecutionResourceProfile,
    ExecutionSortField,
    ExecutionStatus,
    QueryExecutionsRequest,
    ReportSettings,
    ReportType,
)

# 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 = NotebookClient(configuration=server_configuration)


# Create a notebook execution
execution_request = CreateExecutionRequest(
    notebook_id="your_notebook_id",
    parameters={"param1": "value1"},
    workspace_id="your_workspace_id",
    timeout=300,
    result_cache_period=3600,
    report_settings=ReportSettings(
        format=ReportType.HTML,
        exclude_code=False,
    ),
    client_requests_id="your_client_request_id",
    priority=ExecutionPriority.HIGH,
    resource_profile=ExecutionResourceProfile.DEFAULT,
)

# Pass the list of execution requests to the create_executions method
create_execution_response = client.create_executions([execution_request])

# Get the execution by ID
execution = client.get_execution_by_id("your_execution_id")

# Query executions
query_request = QueryExecutionsRequest(
    filter=f"(status = {ExecutionStatus.FAILED.value}))",
    order_by=ExecutionSortField.COMPLETED_AT,
    descending=True,
    projection=[
        ExecutionField.ID,
        ExecutionField.NOTEBOOK_ID,
        ExecutionField.STATUS,
    ],
)

query_executions_response = client.query_executions(query_request)

# Cancel execution
cancel_execution_response = client.cancel_executions(["your_execution_id"])

# Retry execution
retry_execution_response = client.retry_executions(["your_execution_id"])

# Create execution from existing one
run_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.

from nisystemlink.clients.assetmanagement import AssetManagementClient
from nisystemlink.clients.assetmanagement.models import (
    AssetBusType,
    AssetDiscoveryType,
    AssetLocationForCreate,
    AssetPresence,
    AssetPresenceStatus,
    AssetType,
    CreateAssetRequest,
    ExternalCalibration,
    QueryAssetsRequest,
    SelfCalibration,
    TemperatureSensor,
)
from nisystemlink.clients.core._http_configuration import HttpConfiguration


server_configuration = HttpConfiguration(
    server_uri="https://yourserver.yourcompany.com",
    api_key="YourAPIKeyGeneratedFromSystemLink",
)
client = AssetManagementClient(configuration=server_configuration)

create_assets_request = [
    CreateAssetRequest(
        model_number=4000,
        model_name="NI PXIe-6368",
        serial_number="01BB877A",
        vendor_name="NI",
        vendor_number="4244",
        bus_type=AssetBusType.ACCESSORY,
        name="PCISlot2",
        asset_type=AssetType.DEVICE_UNDER_TEST,
        firmware_version="A1",
        hardware_version="12A",
        visa_resource_name="vs-3144",
        temperature_sensors=[TemperatureSensor(name="Sensor0", reading=25.8)],
        supports_self_calibration=True,
        supports_external_calibration=True,
        custom_calibration_interval=24,
        self_calibration=SelfCalibration(
            temperature_sensors=[TemperatureSensor(name="Sensor0", reading=25.8)],
            is_limited=False,
            date="2022-06-07T18:58:05.000Z",
        ),
        is_NI_asset=True,
        workspace="your-workspace-id",
        location=AssetLocationForCreate(
            state=AssetPresence(asset_presence=AssetPresenceStatus.PRESENT)
        ),
        external_calibration=ExternalCalibration(
            temperature_sensors=[TemperatureSensor(name="Sensor0", reading=25.8)],
            date="2022-06-07T18:58:05.000Z",
            recommended_interval=10,
            next_recommended_date="2023-11-14T20:42:11.583Z",
            next_custom_due_date="2024-11-14T20:42:11.583Z",
            resolved_due_date="2022-06-07T18:58:05.000Z",
        ),
        properties={"Key1": "Value1"},
        keywords=["Keyword1"],
        discovery_type=AssetDiscoveryType.MANUAL,
        file_ids=["608a5684800e325b48837c2a"],
        supports_self_test=True,
        supports_reset=True,
        partNumber="A1234 B5",
    )
]

# Create an asset.
create_assets_response = client.create_assets(assets=create_assets_request)

created_asset_id = None
if create_assets_response.assets and len(create_assets_response.assets) > 0:
    created_asset_id = str(create_assets_response.assets[0].id)

# Query assets using id.
query_asset_request = QueryAssetsRequest(
    ids=[created_asset_id],
    skip=0,
    take=1,
    descending=False,
    calibratable_only=False,
    returnCount=False,
)
client.query_assets(query=query_asset_request)

# Link files to the created asset.
file_ids = ["sameple-file-id"]
if created_asset_id:
    link_files_response = client.link_files(
        asset_id=created_asset_id, file_ids=file_ids
    )

# Delete the created asset.
if created_asset_id is not None:
    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.

from nisystemlink.clients.core._http_configuration import HttpConfiguration
from nisystemlink.clients.systems._systems_client import SystemsClient
from nisystemlink.clients.systems.models._create_virtual_systems_request import (
    CreateVirtualSystemRequest,
)
from nisystemlink.clients.systems.models._query_systems_request import (
    QuerySystemsRequest,
)

# 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 = SystemsClient(configuration=server_configuration)

# Systems request metadata.
create_virtual_system_request: CreateVirtualSystemRequest = CreateVirtualSystemRequest(
    alias="Python integration virtual system",
    workspace="your-workspace-id",
)

# Create a virtual system.
create_virtual_system_response = client.create_virtual_system(
    create_virtual_system_request=create_virtual_system_request
)

minion_id = None

if create_virtual_system_response and create_virtual_system_response.minionId:
    minion_id = create_virtual_system_response.minionId

# Query systems using id.
query_systems_request = QuerySystemsRequest(
    filter=f'id="{minion_id}"', projection="new(id, alias)"
)

client.query_systems(query=query_systems_request)

# Delete the created systems.
if minion_id is not None:
    remove_systems = [minion_id]
    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

from datetime import datetime

from nisystemlink.clients.core._http_configuration import HttpConfiguration
from nisystemlink.clients.test_plan import TestPlanClient
from nisystemlink.clients.test_plan.models import (
    CreateTestPlanRequest,
    Dashboard,
    Job,
    JobExecution,
    ManualExecution,
    QueryTestPlansRequest,
    ScheduleTestPlanRequest,
    ScheduleTestPlansRequest,
    UpdateTestPlanRequest,
    UpdateTestPlansRequest,
)

# 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 = TestPlanClient(configuration=server_configuration)

create_test_plans_request = [
    CreateTestPlanRequest(
        name="Python integration test plan",
        state="NEW",
        description="Test plan for verifying integration flow",
        assigned_to="test.user@example.com",
        estimated_duration_in_seconds=86400,
        properties={"env": "staging", "priority": "high"},
        part_number="px40482",
        dut_id="Sample-Dut_Id",
        test_program="TP-Integration-001",
        system_filter="os:linux AND arch:x64",
        workspace="your_workspace_id",
        file_ids_from_template=["file1", "file2"],
        dashboard=Dashboard(
            id="DashBoardId", variables={"product": "PXIe-4080", "location": "Lab1"}
        ),
        execution_actions=[
            ManualExecution(action="boot", type="MANUAL"),
            JobExecution(
                action="run",
                type="JOB",
                jobs=[
                    Job(
                        functions=["run_test_suite"],
                        arguments=[["test_suite.py"]],
                        metadata={"env": "staging"},
                    )
                ],
                systemId="system-001",
            ),
        ],
    )
]

# create a test plan
created_test_plans_response = client.create_test_plans(
    test_plans=create_test_plans_request
)

if created_test_plans_response.created_test_plans:
    created_test_plan_id = created_test_plans_response.created_test_plans[0].id

# Query test plan using id.
query_test_plans_request = QueryTestPlansRequest(
    skip=0, take=1, descending=False, returnCount=False
)
client.query_test_plans(query_request=query_test_plans_request)

# Get test plan
get_test_plan = client.get_test_plan(test_plan_id=created_test_plan_id)

# Update test plan
update_test_plans_request = UpdateTestPlansRequest(
    test_plans=[
        UpdateTestPlanRequest(
            id=created_test_plan_id,
            name="Updated Test Plan",
        )
    ]
)
updated_test_plan = client.update_test_plans(update_request=update_test_plans_request)

# Schedule the test plan
schedule_test_plans_request = ScheduleTestPlansRequest(
    test_plans=[
        ScheduleTestPlanRequest(
            id=created_test_plan_id,
            planned_start_date_time=datetime.strptime(
                "2025-05-20T15:07:42.527Z", "%Y-%m-%dT%H:%M:%S.%fZ"
            ),
            estimated_end_date_time=datetime.strptime(
                "2025-05-22T15:07:42.527Z", "%Y-%m-%dT%H:%M:%S.%fZ"
            ),
            system_id="fake-system",
        )
    ]
)
schedule_test_plan_response = client.schedule_test_plans(
    schedule_request=schedule_test_plans_request
)

# Delete test plan
client.delete_test_plans(ids=[created_test_plan_id])

Create, query and delete test plan templates.

from nisystemlink.clients.core._http_configuration import HttpConfiguration
from nisystemlink.clients.test_plan import TestPlanClient
from nisystemlink.clients.test_plan.models import (
    CreateTestPlanTemplateRequest,
    Dashboard,
    Job,
    JobExecution,
    ManualExecution,
    QueryTestPlanTemplatesRequest,
)


# 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 = TestPlanClient(configuration=server_configuration)

# Test plan template request metadata
create_test_plan_template_request = [
    CreateTestPlanTemplateRequest(
        name="Python integration test plan template",
        template_group="sample template group",
        product_families=["FamilyA", "FamilyB"],
        part_numbers=["PN-1001", "PN-1002"],
        summary="Template for running integration test plans",
        description="This template defines execution steps for integration workflows.",
        test_program="TP-INT-002",
        estimated_duration_in_seconds=86400,
        system_filter="os:linux AND arch:x64",
        execution_actions=[
            ManualExecution(action="boot", type="MANUAL"),
            JobExecution(
                action="run",
                type="JOB",
                jobs=[
                    Job(
                        functions=["run_test_suite"],
                        arguments=[["test_suite.py"]],
                        metadata={"env": "staging"},
                    )
                ],
                systemId="system-001",
            ),
        ],
        file_ids=["file1", "file2"],
        workspace="your_workspace_id",
        properties={"env": "staging", "priority": "high"},
        dashboard=Dashboard(
            id="DashBoardId", variables={"product": "PXIe-4080", "location": "Lab1"}
        ),
    )
]

# Create a test plan template
create_test_plan_template_response = client.create_test_plan_templates(
    test_plan_templates=create_test_plan_template_request
)

create_test_plan_template_id = None

if (
    create_test_plan_template_response.created_test_plan_templates
    and create_test_plan_template_response.created_test_plan_templates[0].id
):
    create_test_plan_template_id = str(
        create_test_plan_template_response.created_test_plan_templates[0].id
    )

# Query test plan templates using id
query_test_plan_template_request = QueryTestPlanTemplatesRequest(
    filter=f'id="{create_test_plan_template_id}"', take=1
)

client.query_test_plan_templates(
    query_test_plan_templates=query_test_plan_template_request
)

# Delete the created test plan template.
if create_test_plan_template_id is not None:
    client.delete_test_plan_templates(ids=[create_test_plan_template_id])