Getting Started

Alarm API

Overview

The AlarmClient class is the primary entry point of the Alarm API.

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

With an AlarmClient object, you can:

• Create and update alarm instances using create_or_update_alarm()

  • Alarms have two key identifiers:

    • alarm_id: A user-defined identifier for the alarm type

    • instance_id: A server-generated unique identifier for each alarm occurrence

  • Create alarm transitions (SET, CLEAR) to track alarm state changes

• Query alarms with query_alarms()

  • Filter alarms using Dynamic LINQ expressions

  • Control which transitions are returned (most recent only or all)

  • Sort and paginate results

• Get a specific alarm by its instance_id using get_alarm()

• Acknowledge alarms by its instance_id using acknowledge_alarms()

  • Optionally force-clear alarms when acknowledging

• Delete alarms using delete_alarm() or delete_alarms()

Examples

Create, query, acknowledge, and delete alarms

import uuid
from datetime import datetime

from nisystemlink.clients.alarm import AlarmClient
from nisystemlink.clients.alarm.models import (
    AlarmOrderBy,
    AlarmSeverityLevel,
    ClearAlarmTransition,
    CreateOrUpdateAlarmRequest,
    QueryAlarmsWithFilterRequest,
    SetAlarmTransition,
    TransitionInclusionOption,
)
from nisystemlink.clients.core import ApiException, HttpConfiguration

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

# To set up the server configuration to point to your instance of SystemLink Enterprise, uncomment
# the following lines and provide your server URI and API key.
# server_configuration = HttpConfiguration(
#     server_uri="https://yourserver.yourcompany.com",
#     api_key="",
# )

client = AlarmClient(configuration=server_configuration)

# Create a unique alarm ID for this example
# alarm_id is a user-defined identifier for this alarm
alarm_id = f"example_alarm_{uuid.uuid1().hex}"

# Create an alarm with a SET transition
create_request = CreateOrUpdateAlarmRequest(
    alarm_id=alarm_id,
    transition=SetAlarmTransition(
        occurred_at=datetime.now(),
        severity_level=AlarmSeverityLevel.HIGH,
        value="85",
        condition="Greater than 80",
        short_text="Temperature is high",
        detail_text="Temperature sensor reading is 85°C (higher than the configured threshold of 80°C)",
    ),
)
# Returns instance_id - a server-generated unique identifier for this specific alarm occurrence
id = client.create_or_update_alarm(create_request)

# Get the alarm by its instance ID (the unique occurrence identifier)
alarm = client.get_alarm(instance_id=id)
print(f"Retrieved alarm: {alarm.alarm_id}, Condition: {alarm.condition}")

# Update the alarm with a higher severity (same alarm_id, updates the same instance)
update_request = CreateOrUpdateAlarmRequest(
    alarm_id=alarm_id,
    transition=SetAlarmTransition(
        occurred_at=datetime.now(),
        severity_level=AlarmSeverityLevel.CRITICAL,
        value="95",
        condition="Greater than 90",
        short_text="Temperature is critical",
        detail_text="Temperature sensor reading is 95°C (higher than the configured threshold of 90°C)",
    ),
)
client.create_or_update_alarm(update_request)

# Query alarms with a filter (can filter by alarm_id to find all instances)
# Include all transitions to see the full alarm history
query_request = QueryAlarmsWithFilterRequest(
    filter="alarmId=@0",
    substitutions=[alarm_id],
    order_by=AlarmOrderBy.UPDATED_AT,
    order_by_descending=True,
    transition_inclusion_option=TransitionInclusionOption.ALL,
    return_count=True,
)
query_response = client.query_alarms(query_request)

# Display query results
print(f"Total alarms found: {query_response.total_count}")
for alarm in query_response.alarms:
    print(f"  Alarm ID: {alarm.alarm_id}, Transitions: {len(alarm.transitions)}")
    for transition in alarm.transitions:
        print(f"- {transition.transition_type}: {transition.condition}")

# Acknowledge the alarm
client.acknowledge_alarms(instance_ids=[id])

# Clear the alarm with 409 conflict handling - Method 1: Manual exception handling
# A 409 Conflict response indicates that the requested transition would not change the alarm's state.
# This allows stateless applications to simply attempt state transitions without first checking
# the current state. For example, a monitoring system can repeatedly try to CLEAR an alarm
# when conditions return to normal, and the API will return 409 if already cleared.
clear_request = CreateOrUpdateAlarmRequest(
    alarm_id=alarm_id,
    transition=ClearAlarmTransition(
        occurred_at=datetime.now(),
        condition="Temperature returned to normal",
    ),
)
try:
    client.create_or_update_alarm(clear_request)
    print("Alarm cleared successfully")
except ApiException as e:
    if e.http_status_code == 409:
        print("Alarm is already in the requested state (409 Conflict)")
    else:
        raise

# Clear the alarm with 409 conflict handling - Method 2: Using ignore_conflict parameter
# This approach is cleaner for stateless applications that don't care about 409 errors.
# Returns None if the alarm is already in the requested state.
result = client.create_or_update_alarm(clear_request, ignore_conflict=True)
if result is None:
    print("No state change needed (alarm already in requested state)")
else:
    print(f"Alarm cleared successfully: {result}")

# Delete the alarm by its instance ID
client.delete_alarm(instance_id=id)

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


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

# To set up the server configuration to point to your instance of SystemLink Enterprise, uncomment
# the following lines and provide your server URI and API key.
# server_configuration = HttpConfiguration(
#     server_uri="https://yourserver.yourcompany.com",
#     api_key="",
# )

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

try:
    import pyarrow as pa  # type: ignore
except Exception:
    pa = None
try:
    import pandas as pd  # type: ignore
except Exception:
    pd = None
from nisystemlink.clients.core import HttpConfiguration
from nisystemlink.clients.dataframe import DataFrameClient
from nisystemlink.clients.dataframe.models import (
    AppendTableDataRequest,
    Column,
    ColumnType,
    CreateTableRequest,
    DataFrame,
    DataType,
)

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

# To set up the server configuration to point to your instance of SystemLink Enterprise, uncomment
# the following lines and provide your server URI and API key.
# server_configuration = HttpConfiguration(
#     server_uri="https://yourserver.yourcompany.com",
#     api_key="",
# )

client = DataFrameClient(configuration=server_configuration)

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


print(f"Created table with ID: {table_id}")

print("Appending data to table...")

# Append via explicit AppendTableDataRequest (JSON)
frame_request = DataFrame(
    data=[[str(i), str(random.random()), datetime.now().isoformat()] for i in range(3)]
)
client.append_table_data(table_id, AppendTableDataRequest(frame=frame_request))

# Append via DataFrame model directly (JSON)
frame_direct = DataFrame(
    data=[
        [str(i + 3), str(random.random()), datetime.now().isoformat()] for i in range(3)
    ]
)
client.append_table_data(table_id, frame_direct)

if pa is not None:
    print("Appending data to table via Arrow RecordBatches...")
    # Append via single RecordBatch (Arrow)
    batch_single = pa.record_batch(
        [
            pa.array([6, 7, 8], type=pa.int32()),
            pa.array([0.1, 0.2, 0.3], type=pa.float32()),
            pa.array([datetime.now() for _ in range(3)], pa.timestamp("ms")),
        ],
        names=["ix", "Float_Column", "Timestamp_Column"],
    )
    client.append_table_data(table_id, batch_single)

    # Append via iterable of RecordBatches (Arrow)
    batch_list = [
        pa.record_batch(
            [
                pa.array([9, 10], type=pa.int32()),
                pa.array([0.4, 0.5], type=pa.float32()),
                pa.array([datetime.now() for _ in range(2)], pa.timestamp("ms")),
            ],
            names=["ix", "Float_Column", "Timestamp_Column"],
        )
    ]
    client.append_table_data(table_id, batch_list)

if pa is not None and pd is not None:
    print("Appending data to table via Pandas DataFrame...")
    # Append via DataFrame (Pandas)
    df = pd.DataFrame(
        {
            "ix": [11, 12, 13],
            "Float_Column": [0.6, 0.7, 0.8],
            "Timestamp_Column": [datetime.now() for _ in range(3)],
        }
    )

    # Optional - coerce df types to the dataframe table schema
    df = df.astype(
        {
            "ix": "Int32",
            "Float_Column": "float32",
            "Timestamp_Column": "datetime64[ns]",
        }
    )

    # convert Pandas DataFrame to Arrow RecordBatch and append
    record_batch = pa.RecordBatch.from_pandas(df)
    client.append_table_data(table_id, record_batch)

# Mark end_of_data for the table
# Supply `None` and `end_of_data=True`
print("Finished appending data.")
client.append_table_data(table_id, None, end_of_data=True)

Query and read data from a table

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

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

# To set up the server configuration to point to your instance of SystemLink Enterprise, uncomment
# the following lines and provide your server URI and API key.
# server_configuration = HttpConfiguration(
#     server_uri="https://yourserver.yourcompany.com",
#     api_key="",
# )

client = DataFrameClient(configuration=server_configuration)

# List a table
response = client.list_tables(take=1)
table = response.tables[0]
y_column = next(
    (col for col in table.columns if col.column_type != ColumnType.Index),
    table.columns[0],
)

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

# Query decimated table data
request = QueryDecimatedDataRequest(
    decimation=DecimationOptions(
        x_column=None,
        y_columns=[y_column.name],
        intervals=1,
        method=DecimationMethod.MaxMin,
    )
)
rows = client.query_decimated_data(table.id, request)
print(rows.frame.model_dump())

Export data from a table

from shutil import copyfileobj

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

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

# To set up the server configuration to point to your instance of SystemLink Enterprise, uncomment
# the following lines and provide your server URI and API key.
# server_configuration = HttpConfiguration(
#     server_uri="https://yourserver.yourcompany.com",
#     api_key="",
# )

client = DataFrameClient(configuration=server_configuration)

# List a table
response = client.list_tables(take=1)
table = response.tables[0]
index_column = next(
    (col for col in table.columns if col.column_type == ColumnType.Index)
)
order_column = next(
    (col for col in table.columns if col.name != index_column.name), index_column
)
filter_value = (
    "0001-01-01T00:00:00Z" if index_column.data_type == DataType.Timestamp else "0"
)

# Export the first 100,000 rows of table data with query options
request = ExportTableDataRequest(
    columns=[index_column.name],
    order_by=[ColumnOrderBy(column=order_column.name, descending=True)],
    filters=[
        ColumnFilter(
            column=index_column.name,
            operation=FilterOperation.NotEquals,
            value=filter_value,
        )
    ],
    take=100000,
    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,
    CreateSpecificationsRequestObject,
    NumericConditionValue,
    QuerySpecificationsRequest,
    SpecificationLimit,
    SpecificationType,
)

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

# To set up the server configuration to point to your instance of SystemLink Enterprise, uncomment
# the following lines and provide your server URI and API key.
# server_configuration = HttpConfiguration(
#     server_uri="https://yourserver.yourcompany.com",
#     api_key="",
# )

client = SpecClient(configuration=server_configuration)

# Create the spec requests
product = "Amplifier"
spec_requests = [
    CreateSpecificationsRequestObject(
        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",
    ),
    CreateSpecificationsRequestObject(
        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",
                ),
            ),
        ],
    ),
    CreateSpecificationsRequestObject(
        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 typing import cast

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

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

# To set up the server configuration to point to your instance of SystemLink Enterprise, uncomment
# the following lines and provide your server URI and API key.
# server_configuration = HttpConfiguration(
#     server_uri="https://yourserver.yourcompany.com",
#     api_key="",
# )

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 = UpdateSpecificationsRequestObject(
        id=cast(str, original_spec1.id),
        product_id=cast(str, original_spec1.product_id),
        spec_id=cast(str, original_spec1.spec_id),
        type=SpecificationType.FUNCTIONAL,
        keywords=["work", "reviewed"],
        block="modifiedBlock",
        version=cast(int, original_spec1.version),
        workspace=cast(str, 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, query and search for files, download and delete files.

• Start upload sessions, upload file chunks, and finish sessions for large file uploads.

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.core import HttpConfiguration
from nisystemlink.clients.file import FileClient

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

# To set up the server configuration to point to your instance of SystemLink Enterprise, uncomment
# the following lines and provide your server URI and API key.
# server_configuration = HttpConfiguration(
#     server_uri="https://yourserver.yourcompany.com",
#     api_key="",
# )

client = FileClient(configuration=server_configuration)

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.core import HttpConfiguration
from nisystemlink.clients.file import FileClient

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

# To set up the server configuration to point to your instance of SystemLink Enterprise, uncomment
# the following lines and provide your server URI and API key.
# server_configuration = HttpConfiguration(
#     server_uri="https://yourserver.yourcompany.com",
#     api_key="",
# )

client = FileClient(configuration=server_configuration)

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}")

Search for files with filtering and pagination

"""Example demonstrating how to search for files using the File API.

Note:
    This example requires Elasticsearch to be configured in the SystemLink cluster.
    If Elasticsearch is not configured, this example will fail with an ApiException.
    For deployments without Elasticsearch, use query_files_linq() instead.
"""

import io
import time

from nisystemlink.clients.core import HttpConfiguration
from nisystemlink.clients.file import FileClient, models
from nisystemlink.clients.file.models import SearchFilesOrderBy, UpdateMetadataRequest

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

# To set up the server configuration to point to your instance of SystemLink Enterprise, uncomment
# the following lines and provide your server URI and API key.
# server_configuration = HttpConfiguration(
#     server_uri="https://yourserver.yourcompany.com",
#     api_key="",
# )

client = FileClient(configuration=server_configuration)

# Upload a test file first
print("Uploading test file...")
test_file_content = b"This is a test file for search demonstration."
test_file = io.BytesIO(test_file_content)
test_file.name = "search-example-test-file.txt"

file_id = client.upload_file(file=test_file)
print(f"Uploaded test file with ID: {file_id}")

# Wait for the file to be indexed for search
# Note: Files may take a few seconds to appear in search results after upload
time.sleep(5)
print()

# Example 1: Basic file search with filter - search for the uploaded file
print("Example 1: Search for the uploaded test file")
search_request = models.SearchFilesRequest(
    filter='name:("search-example-test-file.txt")',
    skip=0,
    take=10,
)

response = client.search_files(search_request)
print(
    f"Found {response.total_count.value if response.total_count else 0} file(s) matching the filter"
)
if response.available_files:
    for file in response.available_files:
        if file.properties:
            print(
                f"- {file.properties.get('Name')} (ID: {file.id}, Size: {file.size} bytes)"
            )

# Example 2: Search with wildcard pattern
print("\nExample 2: Search with wildcard pattern")
search_request = models.SearchFilesRequest(
    filter='name:("search-example*")',
    skip=0,
    take=20,
    order_by=SearchFilesOrderBy.CREATED,
    order_by_descending=True,
)

response = client.search_files(search_request)
print(
    f"Found {response.total_count.value if response.total_count else 0} file(s) starting with 'search-example'"
)
if response.available_files:
    for file in response.available_files:
        if file.properties:
            print(
                f"- {file.properties.get('Name')} created at {file.created} (Size: {file.size} bytes)"
            )

# Example 3: Search by size range
print("\nExample 3: Search by size range")
search_request = models.SearchFilesRequest(
    filter="size:([1 TO 1000])",
    skip=0,
    take=10,
)

response = client.search_files(search_request)
print(
    f"Found {response.total_count.value if response.total_count else 0} file(s) between 1 and 1000 bytes"
)
if response.available_files:
    for file in response.available_files:
        if file.properties:
            print(f"- {file.properties.get('Name')} (Size: {file.size} bytes)")

# Example 4: Search by multiple custom properties
print("\nExample 4: Search by multiple custom properties")
print("Adding custom properties to existing file...")

# Update the existing file with custom properties
custom_metadata = UpdateMetadataRequest(
    replace_existing=False,
    properties={
        "TestProperty1": "TestValue1",
        "TestProperty2": "TestValue2",
    },
)
client.update_metadata(metadata=custom_metadata, id=file_id)

# Wait for indexing
time.sleep(5)

# Search by multiple custom properties using AND operator
search_request = models.SearchFilesRequest(
    filter='(properties.TestProperty1:"TestValue1") AND (properties.TestProperty2:"TestValue2")',
    skip=0,
    take=10,
)

response = client.search_files(search_request)
print(
    f"Found {response.total_count.value if response.total_count else 0} file(s) with "
    "TestProperty1=TestValue1 AND TestProperty2=TestValue2"
)
if response.available_files:
    for file in response.available_files:
        if file.properties:
            print(f"- {file.properties.get('Name')}")
            print(f"  TestProperty1: {file.properties.get('TestProperty1')}")
            print(f"  TestProperty2: {file.properties.get('TestProperty2')}")

# Clean up: delete the test file
print("\nCleaning up...")
client.delete_file(id=file_id)
print(f"Deleted test file with ID: {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 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 configuration is not required when used with SystemLink Client or run through Jupyter on SystemLink
server_configuration: HttpConfiguration | None = None

# To set up the server configuration to point to your instance of SystemLink Enterprise, uncomment
# the following lines and provide your server URI and API key.
# server_configuration = HttpConfiguration(
#     server_uri="https://yourserver.yourcompany.com",
#     api_key="",
# )

client = FeedsClient(configuration=server_configuration)

# 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 import FeedsClient
from nisystemlink.clients.feeds.models import Platform
from nisystemlink.clients.feeds.utilities 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 configuration is not required when used with SystemLink Client or run through Jupyter on SystemLink
server_configuration: HttpConfiguration | None = None

# To set up the server configuration to point to your instance of SystemLink Enterprise, uncomment
# the following lines and provide your server URI and API key.
# server_configuration = HttpConfiguration(
#     server_uri="https://yourserver.yourcompany.com",
#     api_key="",
# )

client = FeedsClient(configuration=server_configuration)

# 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 import FeedsClient
from nisystemlink.clients.feeds.models import Platform
from nisystemlink.clients.feeds.utilities 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 configuration is not required when used with SystemLink Client or run through Jupyter on SystemLink
server_configuration: HttpConfiguration | None = None

# To set up the server configuration to point to your instance of SystemLink Enterprise, uncomment
# the following lines and provide your server URI and API key.
# server_configuration = HttpConfiguration(
#     server_uri="https://yourserver.yourcompany.com",
#     api_key="",
# )

client = FeedsClient(configuration=server_configuration)

# 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.core import HttpConfiguration
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: HttpConfiguration | None = None

# To set up the server configuration to point to your instance of SystemLink Enterprise, uncomment
# the following lines and provide your server URI and API key.
# server_configuration = HttpConfiguration(
#     server_uri="https://yourserver.yourcompany.com",
#     api_key="",
# )

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 typing import cast

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


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 through Jupyter on SystemLink
server_configuration: HttpConfiguration | None = None

# To set up the server configuration to point to your instance of SystemLink Enterprise, uncomment
# the following lines and provide your server URI and API key.
# server_configuration = HttpConfiguration(
#     server_uri="https://yourserver.yourcompany.com",
#     api_key="",
# )

client = TestMonitorClient(configuration=server_configuration)

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

# Create the step requests
result_id = cast(str, create_response.results[0].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=cast(str, 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",
                    )
                ],
            ),
        )
        for step in created_steps
    ]
)

# delete all steps at once
delete_response = client.delete_steps(
    steps=[
        StepIdResultIdPair(
            step_id=cast(str, step.step_id), result_id=cast(str, 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

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

# To set up the server configuration to point to your instance of SystemLink Enterprise, uncomment
# the following lines and provide your server URI and API key.
# server_configuration = HttpConfiguration(
#     server_uri="https://yourserver.yourcompany.com",
#     api_key="",
# )

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

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

# To set up the server configuration to point to your instance of SystemLink Enterprise, uncomment
# the following lines and provide your server URI and API key.
# server_configuration = HttpConfiguration(
#     server_uri="https://yourserver.yourcompany.com",
#     api_key="",
# )

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, query assets and link files to assets.

• Track asset utilization with start, heartbeat, end, and query history operations.

Examples

Create, delete, and query assets and link files to assets.

from datetime import datetime, timezone

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 import HttpConfiguration

# workspace where the asset will be created
workspace_id = "yourWorkspaceId"

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

# To set up the server configuration to point to your instance of SystemLink Enterprise, uncomment
# the following lines and provide your server URI and API key.
# server_configuration = HttpConfiguration(
#     server_uri="https://yourserver.yourcompany.com",
#     api_key="",
# )

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=datetime(2022, 6, 7, 18, 58, 5, tzinfo=timezone.utc),
        ),
        is_NI_asset=True,
        workspace=workspace_id,
        location=AssetLocationForCreate(
            state=AssetPresence(asset_presence=AssetPresenceStatus.PRESENT)
        ),
        external_calibration=ExternalCalibration(
            temperature_sensors=[TemperatureSensor(name="Sensor0", reading=25.8)],
            date=datetime(2022, 6, 7, 18, 58, 5, tzinfo=timezone.utc),
            recommended_interval=10,
            next_recommended_date=datetime(
                2023, 11, 14, 20, 42, 11, 583000, tzinfo=timezone.utc
            ),
            next_custom_due_date=datetime(
                2024, 11, 14, 20, 42, 11, 583000, tzinfo=timezone.utc
            ),
            resolved_due_date=datetime(2022, 6, 7, 18, 58, 5, tzinfo=timezone.utc),
        ),
        properties={"Key1": "Value1"},
        keywords=["Keyword1"],
        discovery_type=AssetDiscoveryType.MANUAL,
        file_ids=["608a5684800e325b48837c2a"],
        supports_self_test=True,
        supports_reset=True,
        part_number="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(
    filter=f'AssetIdentifier = "{created_asset_id}"',
    skip=0,
    take=1,
    descending=False,
    return_count=False,
)
client.query_assets(query=query_asset_request)

# Link files to the created asset.
file_ids = ["sample-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])

Track asset utilization.

import threading
import time
from datetime import datetime
from uuid import uuid4

from nisystemlink.clients.assetmanagement import AssetManagementClient
from nisystemlink.clients.assetmanagement.models import (
    AssetBusType,
    AssetIdentification,
    AssetLocationForCreate,
    AssetPresence,
    AssetPresenceStatus,
    CreateAssetRequest,
    StartUtilizationRequest,
)
from nisystemlink.clients.core import HttpConfiguration
from nisystemlink.clients.core.helpers import read_minion_id

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

# To set up the server configuration to point to your instance of SystemLink Enterprise, uncomment
# the following lines and provide your server URI and API key.
# server_configuration = HttpConfiguration(
#     server_uri="https://yourserver.yourcompany.com",
#     api_key="",
# )

client = AssetManagementClient(configuration=server_configuration)

# Generate a unique identifier for this utilization session
utilization_id = str(uuid4())

# Create the assets first
# Define the assets to be created and used in the test
create_assets_request = [
    CreateAssetRequest(
        model_name="NI PXIe-6368",
        model_number=4000,
        serial_number="01BB877A",
        vendor_name="NI",
        vendor_number=4244,
        bus_type=AssetBusType.ACCESSORY,
        name="DAQ Device - 01BB877A",
        location=AssetLocationForCreate(
            state=AssetPresence(asset_presence=AssetPresenceStatus.PRESENT)
        ),
    ),
    CreateAssetRequest(
        model_name="NI PXIe-5163",
        model_number=5000,
        serial_number="02CC988B",
        vendor_name="NI",
        vendor_number=4244,
        bus_type=AssetBusType.ACCESSORY,
        name="Oscilloscope - 02CC988B",
        location=AssetLocationForCreate(
            state=AssetPresence(asset_presence=AssetPresenceStatus.PRESENT)
        ),
    ),
]

# Create the assets in SystemLink
create_assets_response = client.create_assets(assets=create_assets_request)

if create_assets_response.assets:
    print(f"Created {len(create_assets_response.assets)} asset(s)")
    created_asset_ids = [
        asset.id for asset in create_assets_response.assets if asset.id
    ]
else:
    print("Failed to create assets")
    exit(1)

# Define the asset identifications for utilization tracking
test_assets = [
    AssetIdentification(
        model_name="NI PXIe-6368",
        model_number=4000,
        serial_number="01BB877A",
        vendor_name="NI",
        vendor_number=4244,
        bus_type=AssetBusType.ACCESSORY,
    ),
    AssetIdentification(
        model_name="NI PXIe-5163",
        model_number=5000,
        serial_number="02CC988B",
        vendor_name="NI",
        vendor_number=4244,
        bus_type=AssetBusType.ACCESSORY,
    ),
]

# Start asset utilization tracking
# This marks the assets as "in use" in the SystemLink UI
# Read the minion ID from the Salt configuration
minion_id = read_minion_id() or "test-station-01"  # Fallback minion ID if not found

start_utilization_request = StartUtilizationRequest(
    utilization_identifier=utilization_id,
    minion_id=minion_id,
    asset_identifications=test_assets,
    utilization_category="Automated Testing",
    task_name="DUT Validation Suite",
    user_name="automation_user",
    utilization_timestamp=datetime.now(),
)

start_utilization_response = client.start_utilization(request=start_utilization_request)

print(start_utilization_response)

# Verify utilization started successfully
if start_utilization_response.assets_with_started_utilization:
    print(
        f"Utilization started for {len(start_utilization_response.assets_with_started_utilization)} asset(s)"
    )
else:
    print("Failed to start utilization")


# Heartbeat mechanism using a background thread
# IMPORTANT: Heartbeats are for UI purposes only, to keep assets visually
# marked as "in use" in the SystemLink UI. This applies only to utilizations
# that have not been ended. While the standard heartbeat interval is 5 minutes,
# the UI requires heartbeats at least every 10 minutes to continue showing
# assets as actively utilized. If heartbeats stop, the assets will no longer
# appear as "in use" in the UI.
heartbeat_interval = 300  # 5 minutes in seconds
stop_event = threading.Event()


def heartbeat_loop():
    """Background thread that sends periodic heartbeats.

    This keeps the asset visually marked as "in use" in the SystemLink UI
    (for UI purposes only). Applies only to utilizations that have not been ended.
    The UI requires heartbeats at least every 10 minutes to continue displaying
    the asset as actively utilized.
    """
    while not stop_event.wait(heartbeat_interval):
        heartbeat_response = client.utilization_heartbeat(
            ids=[utilization_id],
            timestamp=datetime.now(),
        )

        if heartbeat_response.updated_utilization_ids:
            print(
                f"Heartbeat sent at {datetime.now().strftime('%H:%M:%S')} - asset remains 'in use'"
            )


# Start the heartbeat thread
heartbeat_thread = threading.Thread(target=heartbeat_loop, daemon=True)
heartbeat_thread.start()

# Simulate a long-running operation where assets are in use
# In a real scenario, this would be your actual test or operation
print("\nAssets are now in use. Heartbeats will be sent every 5 minutes...")
print("Waiting for 10 minutes to demonstrate heartbeats...\n")

time.sleep(600)  # Wait 10 minutes

# Stop the heartbeat thread
stop_event.set()
heartbeat_thread.join()

# End asset utilization tracking
end_utilization_response = client.end_utilization(
    ids=[utilization_id],
    timestamp=datetime.now(),
)

if end_utilization_response.updated_utilization_ids:
    print("\nUtilization ended - asset(s) released")

# Clean up: delete the created assets
if created_asset_ids:
    client.delete_assets(ids=created_asset_ids)
    print(f"Deleted {len(created_asset_ids)} asset(s)")

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 import HttpConfiguration
from nisystemlink.clients.systems import SystemsClient
from nisystemlink.clients.systems.models import (
    CreateVirtualSystemRequest,
    QuerySystemsRequest,
)

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

# To set up the server configuration to point to your instance of SystemLink Enterprise, uncomment
# the following lines and provide your server URI and API key.
# server_configuration = HttpConfiguration(
#     server_uri="https://yourserver.yourcompany.com",
#     api_key="",
# )

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)

WorkItem API

Overview

The WorkItemClient class is the primary entry point of the WorkItem API.

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

With a WorkItemClient object, you can:

• Create, query, get, update, schedule and delete work items

• Create, query, update and delete work item templates

Examples

Create, query, get, update, schedule and delete work items

from datetime import datetime

from nisystemlink.clients.core import HttpConfiguration
from nisystemlink.clients.work_item import WorkItemClient
from nisystemlink.clients.work_item.models import (
    CreateWorkItemRequest,
    Dashboard,
    Job,
    JobExecution,
    ManualExecution,
    QueryWorkItemsRequest,
    ResourceDefinition,
    ResourcesDefinition,
    ResourceSelectionDefinition,
    ScheduleDefinition,
    ScheduleResourcesDefinition,
    ScheduleSystemResourceDefinition,
    ScheduleWorkItemRequest,
    ScheduleWorkItemsRequest,
    SystemResourceDefinition,
    SystemResourceSelectionDefinition,
    TimelineDefinition,
    UpdateWorkItemRequest,
    UpdateWorkItemsRequest,
)

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

# To set up the server configuration to point to your instance of SystemLink Enterprise, uncomment
# the following lines and provide your server URI and API key.
# server_configuration = HttpConfiguration(
#     server_uri="https://yourserver.yourcompany.com",
#     api_key="",
# )

client = WorkItemClient(configuration=server_configuration)

create_work_items_request = [
    CreateWorkItemRequest(
        name="Python integration work item",
        type="testplan",
        state="NEW",
        description="Work item for verifying integration flow",
        assigned_to="test.user@example.com",
        requested_by="test.manager@example.com",
        properties={"env": "staging", "priority": "high"},
        part_number="px40482",
        test_program="TP-Integration-001",
        workspace="your_workspace_id",
        timeline=TimelineDefinition(
            earliest_start_date_time=datetime.strptime(
                "2024-01-15T08:00:00Z", "%Y-%m-%dT%H:%M:%SZ"
            ),
            due_date_time=datetime.strptime(
                "2024-01-20T17:00:00Z", "%Y-%m-%dT%H:%M:%SZ"
            ),
            estimated_duration_in_seconds=86400,
        ),
        resources=ResourcesDefinition(
            systems=SystemResourceDefinition(
                selections=[
                    SystemResourceSelectionDefinition(
                        id="system-001",
                        target_location_id="location-001",
                    ),
                ],
                filter='properties.data["Lab"] = "Battery Pack Lab"',
            ),
            duts=ResourceDefinition(
                selections=[
                    ResourceSelectionDefinition(
                        id="dut-001",
                        target_location_id="location-001",
                        target_system_id="system-001",
                        target_parent_id="parent-001",
                    ),
                ],
                filter='modelName = "cRIO-9045" && serialNumber = "01E82ED0"',
            ),
            assets=ResourceDefinition(
                selections=[
                    ResourceSelectionDefinition(
                        id="asset-001",
                        target_location_id="location-001",
                        target_system_id="system-001",
                        target_parent_id="parent-001",
                    ),
                ],
                filter='modelName = "cRIO-9045" && serialNumber = "01E82ED0"',
            ),
            fixtures=ResourceDefinition(
                selections=[
                    ResourceSelectionDefinition(
                        id="fixture-001",
                        target_location_id="location-001",
                        target_system_id="system-001",
                        target_parent_id="parent-001",
                    ),
                ],
                filter='modelName = "cRIO-9045" && serialNumber = "01E82ED0"',
            ),
        ),
        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 work item
create_work_items_response = client.create_work_items(
    work_items=create_work_items_request
)

if create_work_items_response.created_work_items:
    created_work_item_id = create_work_items_response.created_work_items[0].id
    print(f"Created work item: {created_work_item_id}")

# Query work items using id.
query_work_items_request = QueryWorkItemsRequest(
    filter=f'id = "{created_work_item_id}"',
    take=1,
    descending=False,
    return_count=False,
)
query_work_items_response = client.query_work_items(
    query_work_items=query_work_items_request
)
if query_work_items_response.work_items:
    print(f"Found work item: {query_work_items_response.work_items[0].name}")

# Get work item
if created_work_item_id is not None:
    get_work_item_response = client.get_work_item(work_item_id=created_work_item_id)
    print(f"Retrieved work item: {get_work_item_response.name}")

# Update work item
if created_work_item_id is not None:
    update_work_items_request = UpdateWorkItemsRequest(
        work_items=[
            UpdateWorkItemRequest(id=created_work_item_id, name="Updated work item")
        ]
    )
    update_work_items_response = client.update_work_items(
        update_work_items=update_work_items_request
    )
    if update_work_items_response.updated_work_items:
        print(
            f"Work item name updated to: {update_work_items_response.updated_work_items[0].name}"
        )

# Schedule work item
if created_work_item_id is not None:
    schedule_work_items_request = ScheduleWorkItemsRequest(
        work_items=[
            ScheduleWorkItemRequest(
                id=created_work_item_id,
                schedule=ScheduleDefinition(
                    planned_start_date_time=datetime.strptime(
                        "2025-05-20T15:07:42.527Z", "%Y-%m-%dT%H:%M:%S.%fZ"
                    ),
                    planned_end_date_time=datetime.strptime(
                        "2025-05-22T15:07:42.527Z", "%Y-%m-%dT%H:%M:%S.%fZ"
                    ),
                    planned_duration_in_seconds=172800,
                ),
                resources=ScheduleResourcesDefinition(
                    systems=ScheduleSystemResourceDefinition(
                        selections=[
                            SystemResourceSelectionDefinition(
                                id="system-123",
                                target_location_id="location-456",
                            )
                        ]
                    )
                ),
            )
        ],
        replace=True,
    )
    schedule_work_items_response = client.schedule_work_items(
        schedule_work_items=schedule_work_items_request
    )
    if schedule_work_items_response.scheduled_work_items:
        print(
            f"Scheduled work item with ID: {schedule_work_items_response.scheduled_work_items[0].id}"
        )

# Delete work item
if created_work_item_id is not None:
    client.delete_work_items(ids=[created_work_item_id])
    print("Work item deleted successfully.")

Create, query, update and delete work item templates.

from nisystemlink.clients.core import HttpConfiguration
from nisystemlink.clients.work_item import WorkItemClient
from nisystemlink.clients.work_item.models import (
    CreateWorkItemTemplateRequest,
    Dashboard,
    Job,
    JobExecution,
    ManualExecution,
    QueryWorkItemTemplatesRequest,
    TemplateResourceDefinition,
    TemplateResourcesDefinition,
    TemplateTimelineDefinition,
    UpdateWorkItemTemplateRequest,
    UpdateWorkItemTemplatesRequest,
)

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

# To set up the server configuration to point to your instance of SystemLink Enterprise, uncomment
# the following lines and provide your server URI and API key.
# server_configuration = HttpConfiguration(
#     server_uri="https://yourserver.yourcompany.com",
#     api_key="",
# )

client = WorkItemClient(configuration=server_configuration)

create_work_item_template_request = [
    CreateWorkItemTemplateRequest(
        name="Python integration work item template",
        template_group="sample template group",
        type="testplan",
        product_families=["FamilyA", "FamilyB"],
        part_numbers=["PN-1001", "PN-1002"],
        summary="Template for running integration work items",
        description="This template defines execution steps for integration workflows.",
        test_program="TP-INT-002",
        timeline=TemplateTimelineDefinition(estimated_duration_in_seconds=86400),
        resources=TemplateResourcesDefinition(
            systems=TemplateResourceDefinition(
                filter='properties.data["Lab"] = "Battery Pack Lab" && state = "Available"'
            ),
            duts=TemplateResourceDefinition(
                filter='modelName = "cRIO-9045" && serialNumber = "01E82ED0"'
            ),
            assets=TemplateResourceDefinition(
                filter='modelName = "cRIO-9045" && serialNumber = "01E82ED0"'
            ),
            fixtures=TemplateResourceDefinition(
                filter='modelName = "cRIO-9045" && serialNumber = "01E82ED0"'
            ),
        ),
        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 work item template
create_work_item_templates_response = client.create_work_item_templates(
    work_item_templates=create_work_item_template_request
)

if create_work_item_templates_response.created_work_item_templates:
    create_work_item_template_id = (
        create_work_item_templates_response.created_work_item_templates[0].id
    )
    print(f"Created work item template: {create_work_item_template_id}")

# Query work item templates using id
query_work_item_template_request = QueryWorkItemTemplatesRequest(
    filter=f'id="{create_work_item_template_id}"', take=1
)
query_work_item_templates_response = client.query_work_item_templates(
    query_work_item_templates=query_work_item_template_request
)
if query_work_item_templates_response.work_item_templates:
    print(
        f"Found work item template: {query_work_item_templates_response.work_item_templates[0].name}"
    )

# Update work item template
if create_work_item_template_id is not None:
    update_work_item_template_request = UpdateWorkItemTemplatesRequest(
        work_item_templates=[
            UpdateWorkItemTemplateRequest(
                id=create_work_item_template_id, name="Updated work item template"
            )
        ]
    )
    update_work_item_templates_response = client.update_work_item_templates(
        update_work_item_templates=update_work_item_template_request
    )
    if update_work_item_templates_response.updated_work_item_templates:
        print(
            "Work item template name updated to: "
            f"{update_work_item_templates_response.updated_work_item_templates[0].name}"
        )

# Delete work item template
if create_work_item_template_id is not None:
    client.delete_work_item_templates(ids=[create_work_item_template_id])
    print("Work item template deleted successfully.")

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

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

# To set up the server configuration to point to your instance of SystemLink Enterprise, uncomment
# the following lines and provide your server URI and API key.
# server_configuration = HttpConfiguration(
#     server_uri="https://yourserver.yourcompany.com",
#     api_key="",
# )

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",
        dut_serial_number="serial_number_123",
        test_program="TP-Integration-001",
        system_filter="os:linux AND arch:x64",
        dut_filter="modelName = 'cRIO-9045' AND serialNumber = '01E82ED0'",
        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(
    take=1, descending=False, return_count=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 import HttpConfiguration
from nisystemlink.clients.test_plan import TestPlanClient
from nisystemlink.clients.test_plan.models import (
    CreateTestPlanTemplateRequest,
    Dashboard,
    Job,
    JobExecution,
    ManualExecution,
    QueryTestPlanTemplatesRequest,
)

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

# To set up the server configuration to point to your instance of SystemLink Enterprise, uncomment
# the following lines and provide your server URI and API key.
# server_configuration = HttpConfiguration(
#     server_uri="https://yourserver.yourcompany.com",
#     api_key="",
# )

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",
        dut_filter="modelName = 'cRIO-9045' AND serialNumber = '01E82ED0'",
        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])

Notification API

Overview

The NotificationClient class is the primary entry point of the Notification API.

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

With a NotificationClient object, you can:

• Apply dynamic notification strategy using apply_dynamic_notification_strategy()

Examples

Apply a notification strategy

import uuid
from datetime import datetime

from nisystemlink.clients.alarm import AlarmClient
from nisystemlink.clients.alarm.models._alarm import Alarm, AlarmSeverityLevel
from nisystemlink.clients.alarm.models._create_or_update_alarm_request import (
    CreateOrUpdateAlarmRequest,
    SetAlarmTransition,
)
from nisystemlink.clients.core import HttpConfiguration
from nisystemlink.clients.notification import NotificationClient
from nisystemlink.clients.notification.models import (
    DynamicNotificationConfiguration,
    DynamicNotificationStrategy,
    DynamicStrategyRequest,
    SmtpAddressFields,
    SmtpAddressGroup,
    SmtpMessageTemplate,
    SmtpMessageTemplateFields,
)

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

# To set up the server configuration to point to your instance of SystemLink Enterprise, uncomment
# the following lines and provide your server URI and API key.
# server_configuration = HttpConfiguration(
#     server_uri="https://yourserver.yourcompany.com",
#     api_key="",
# )


# Create request for applying strategy
def create_notification_request_for_alarm(
    alarm: Alarm,
    address_group: SmtpAddressGroup,
    message_template: SmtpMessageTemplate,
) -> DynamicStrategyRequest:
    """Creates and returns a dynamic strategy request."""
    occurred_at = alarm.most_recent_transition_occurred_at

    return DynamicStrategyRequest(
        message_template_substitution_fields={
            "alarm_id": alarm.alarm_id,
            "alarm_condition": alarm.condition,
            "alarm_description": alarm.description,
            "alarm_severity": str(alarm.current_severity_level),
            "alarm_occurred_at": occurred_at.isoformat() if occurred_at else "",
        },
        notification_strategy=DynamicNotificationStrategy(
            notification_configurations=[
                DynamicNotificationConfiguration(
                    address_group=address_group,
                    message_template=message_template,
                )
            ]
        ),
    )


# Create clients for Notification and Alarm services
notification_client = NotificationClient(configuration=server_configuration)
alarm_client = AlarmClient(configuration=server_configuration)

# Create a unique alarm ID for this example
alarm_id = f"example_alarm_{uuid.uuid1().hex}"

# Create an alarm with a SET transition
create_alarm_request = CreateOrUpdateAlarmRequest(
    alarm_id=alarm_id,
    transition=SetAlarmTransition(
        occurred_at=datetime.now(),
        severity_level=AlarmSeverityLevel.HIGH,
        value="85",
        condition="Greater than 80",
        short_text="Temperature is high",
        detail_text="Temperature sensor reading is 85°C (higher than the configured threshold of 80°C)",
    ),
    description="Example alarm for notification",
)
id = alarm_client.create_or_update_alarm(create_alarm_request)
print("Alarm created successfully")

# Get the alarm by its instance ID (the unique occurrence identifier)
retrieved_alarm = alarm_client.get_alarm(instance_id=id)

# Define recipients to notify
recipients = SmtpAddressFields(to_addresses=["sample1@example.com"])

# Create address group
address_group = SmtpAddressGroup(
    display_name="Alarm Notification Recipients",
    properties={"address group": "Alarm"},
    fields=recipients,
)

# Create mail template for alarm creation notification
alarm_creation_template = SmtpMessageTemplate(
    display_name="Alarm Creation Template",
    fields=SmtpMessageTemplateFields(
        subject_template="Alarm Created: <alarm_id>",
        body_template="An alarm with ID <alarm_id> has been created.\n"
        "Condition: <alarm_condition>\n"
        "Description: <alarm_description>\n"
        "Current severity: <alarm_severity>\n"
        "Occurred At: <alarm_occurred_at>",
    ),
)

# Send notification for alarm creation
notification_for_alarm_creation = create_notification_request_for_alarm(
    alarm=retrieved_alarm,
    address_group=address_group,
    message_template=alarm_creation_template,
)
notification_client.apply_dynamic_notification_strategy(
    request=notification_for_alarm_creation
)
print("Notification sent for alarm creation")