Getting Started
Tag API
Overview
The TagManager
class is the primary entry point of the Tag API.
When constructing a TagManager
, you can pass an
HttpConfiguration
(like one retrieved from the
HttpConfigurationManager
), or let TagManager
use the
default connection. The default connection depends on your SystemLink Client
settings.
With a TagManager
object, you can:
Query, create, modify, and delete tags from the server
Use
open()
to get a tag’sTagData
from the server when you know the tag’spath
.Use
query()
to get acollection
ofTagData
objects based on the tags’paths
,keywords
, and/orproperties
.Use
refresh()
to update a list ofTagData
objects with fresh metadata from the server.Use
update()
to modify the server metadata for a list of tags, using eitherTagData
objects to overwrite the server’s tag data orTagDataUpdate
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’scollect_aggregates
attribute is enabled on the server).Use
create_writer()
to get aBufferedTagWriter
that will buffer a set of writes, and automatically send the writes when a givenbuffer_size
is reached or whenmax_buffer_time
has elapsed (or whensend_buffered_writes()
is called).
Get a
TagSelection
that can help perform several of the above operations on several tags at onceUse
TagManager.create_selection()
if you already have a list ofTagData
objects that you want to perform a set of operations on.Use
TagManager.open_selection()
if you just have a list ofpaths
– 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
1from datetime import timedelta
2
3from nisystemlink.clients.tag import DataType, TagManager
4
5mgr = TagManager()
6tag = mgr.open("MyTags.Example Tag", DataType.DOUBLE, create=True)
7
8with mgr.create_writer(buffer_size=10, max_buffer_time=timedelta(seconds=3)) as writer:
9 writer.write(tag.path, tag.data_type, 3.5)
10# Note: Exiting the "with" block automatically calls writer.send_buffered_writes()
11
12read_result = mgr.read(tag.path)
13assert read_result is not None
14assert read_result.value == 3.5
Subscribe to tag changes
1from contextlib import ExitStack
2from time import sleep
3
4from nisystemlink.clients.tag import DataType, TagData, TagManager, TagValueReader
5
6SIMULATE_EXTERNAL_TAG_CHANGES = True
7
8
9def on_tag_changed(tag: TagData, reader: TagValueReader) -> None:
10 """Callback for tag_changed events."""
11 path = tag.path
12 data_type = tag.data_type.name
13
14 if reader is not None:
15 read_result = reader.read()
16 # A read_result of None means that the tag has no value, but it *must*
17 # have a value, because we got a tag_changed event!
18 assert read_result is not None
19 value = read_result.value
20 else:
21 value = "???" # tag has unknown data type
22
23 print(f'Tag changed: "{path}" = {value} ({data_type})')
24
25
26mgr = TagManager()
27if SIMULATE_EXTERNAL_TAG_CHANGES:
28 mgr.open("MyTags.Example Tag", DataType.DOUBLE, create=True)
29 writer = mgr.create_writer(buffer_size=1)
30
31with ExitStack() as stack:
32 # Notes:
33 # 1. The tags are assumed to already exist before this example is run, but
34 # setting SIMULATE_EXTERNAL_TAG_CHANGES to True will ensure there is one.
35 # 2. Any tags that get added later will NOT automatically appear in the
36 # selection just because the path matches the wildcard used below; you
37 # must call one of the selection's refresh methods to update the tag list
38 # from the server. But even if you do that:
39 # 3. The subscription will only contain the tags that were in the selection
40 # when the subscription was created. If you want the subscription to add
41 # new tags that were added to the selection, you must recreate it.
42 paths = ["MyTags.*"]
43 selection = stack.enter_context(mgr.open_selection(paths))
44 if not selection.metadata:
45 print(f"Found no tags that match {paths}")
46 else:
47 print("Matching tags:")
48 for path in selection.metadata.keys():
49 print(f" - {path}")
50 print()
51
52 subscription = stack.enter_context(selection.create_subscription())
53 subscription.tag_changed += on_tag_changed
54
55 # Wait forever, until a KeyboardInterrupt (Ctrl+C)
56 print("Watching for tag changes; hit Ctrl+C to stop")
57 try:
58 i = 0
59 while True:
60 sleep(1)
61 if SIMULATE_EXTERNAL_TAG_CHANGES:
62 writer.write("MyTags.Example Tag", DataType.DOUBLE, i)
63 i += 1
64 except KeyboardInterrupt:
65 pass
Product API
Overview
The ProductClient
class is the primary entry point of the Product API.
When constructing a ProductClient
, you can pass an
HttpConfiguration
(like one retrieved from the
HttpConfigurationManager
), or let ProductClient
use the
default connection. The default connection depends on your environment.
With a ProductClient
object, you can:
Create, update, query, and delete Products
Examples
Create, query, update, and delete some products
1from nisystemlink.clients.core import HttpConfiguration
2from nisystemlink.clients.product import ProductClient
3from nisystemlink.clients.product.models import (
4 CreateProductRequest,
5 ProductField,
6 ProductOrderBy,
7 QueryProductsRequest,
8 QueryProductValuesRequest,
9)
10
11name = "Example Name"
12family = "Example Family"
13
14
15def create_some_products():
16 """Create two example products on your server."""
17 new_products = [
18 CreateProductRequest(
19 part_number="Example 123 AA",
20 name=name,
21 family=family,
22 keywords=["original keyword"],
23 properties={"original property key": "yes"},
24 ),
25 CreateProductRequest(
26 part_number="Example 123 AA1",
27 name=name,
28 family=family,
29 keywords=["original keyword"],
30 properties={"original property key": "original"},
31 ),
32 ]
33 create_response = client.create_products(new_products)
34 return create_response
35
36
37# Setup the server configuration to point to your instance of SystemLink Enterprise
38server_configuration = HttpConfiguration(
39 server_uri="https://yourserver.yourcompany.com",
40 api_key="YourAPIKeyGeneratedFromSystemLink",
41)
42client = ProductClient(configuration=server_configuration)
43
44# Get all the products using the continuation token in batches of 100 at a time.
45response = client.get_products_paged(take=100, return_count=True)
46all_products = response.products
47while response.continuation_token:
48 response = client.get_products_paged(
49 take=100, continuation_token=response.continuation_token, return_count=True
50 )
51 all_products.extend(response.products)
52
53create_response = create_some_products()
54
55# use get for first product created
56created_product = client.get_product(create_response.products[0].id)
57
58# Query products without continuation
59query_request = QueryProductsRequest(
60 filter=f'family="{family}" && name="{name}"',
61 return_count=True,
62 order_by=ProductOrderBy.FAMILY,
63)
64query_response = client.query_products_paged(query_request)
65
66# Update the first product that you just created and replace the keywords
67updated_product = create_response.products[0]
68updated_product.keywords = ["new keyword"]
69updated_product.properties = {"new property key": "new value"}
70update_response = client.update_products([create_response.products[0]], replace=True)
71
72# Query for just the ids of products that match the family
73values_query = QueryProductValuesRequest(
74 filter=f'family="{family}"', field=ProductField.ID
75)
76values_response = client.query_product_values(query=values_query)
77
78# delete each created product individually by id
79for product in create_response.products:
80 client.delete_product(product.id)
81
82# Create some more and delete them with a single call to delete.
83create_response = create_some_products()
84client.delete_products([product.id for product in create_response.products])
DataFrame API
Overview
The DataFrameClient
class is the primary entry point of the DataFrame API.
When constructing a DataFrameClient
, you can pass an
HttpConfiguration
(like one retrieved from the
HttpConfigurationManager
), or let DataFrameClient
use the
default connection. The default connection depends on your environment.
With a DataFrameClient
object, you can:
Create and delete data tables.
Modify table metadata and query for tables by their metadata.
Append rows of data to a table, query for rows of data from a table, and decimate table data.
Export table data in a comma-separated values (CSV) format.
Examples
Create and write data to a table
1import random
2from datetime import datetime
3
4from nisystemlink.clients.dataframe import DataFrameClient
5from nisystemlink.clients.dataframe.models import (
6 AppendTableDataRequest,
7 Column,
8 ColumnType,
9 CreateTableRequest,
10 DataFrame,
11 DataType,
12)
13
14client = DataFrameClient()
15
16# Create table
17table_id = client.create_table(
18 CreateTableRequest(
19 name="Example Table",
20 columns=[
21 Column(name="ix", data_type=DataType.Int32, column_type=ColumnType.Index),
22 Column(name="Float_Column", data_type=DataType.Float32),
23 Column(name="Timestamp_Column", data_type=DataType.Timestamp),
24 ],
25 )
26)
27
28# Generate example data
29frame = DataFrame(
30 data=[[i, random.random(), datetime.now().isoformat()] for i in range(100)]
31)
32
33# Write example data to table
34client.append_table_data(
35 table_id, data=AppendTableDataRequest(frame=frame, endOfData=True)
36)
Query and read data from a table
1from nisystemlink.clients.dataframe import DataFrameClient
2from nisystemlink.clients.dataframe.models import (
3 DecimationMethod,
4 DecimationOptions,
5 QueryDecimatedDataRequest,
6)
7
8client = DataFrameClient()
9
10# List a table
11response = client.list_tables(take=1)
12table = response.tables[0]
13
14# Get table metadata by table id
15client.get_table_metadata(table.id)
16
17# Query decimated table data
18request = QueryDecimatedDataRequest(
19 decimation=DecimationOptions(
20 x_column="index",
21 y_columns=["col1"],
22 intervals=1,
23 method=DecimationMethod.MaxMin,
24 )
25)
26client.query_decimated_data(table.id, request)
Export data from a table
1from shutil import copyfileobj
2
3from nisystemlink.clients.dataframe import DataFrameClient
4from nisystemlink.clients.dataframe.models import (
5 ColumnFilter,
6 ColumnOrderBy,
7 ExportFormat,
8 ExportTableDataRequest,
9 FilterOperation,
10)
11
12client = DataFrameClient()
13
14# List a table
15response = client.list_tables(take=1)
16table = response.tables[0]
17
18# Export table data with query options
19request = ExportTableDataRequest(
20 columns=["col1"],
21 order_by=[ColumnOrderBy(column="col2", descending=True)],
22 filters=[
23 ColumnFilter(column="col1", operation=FilterOperation.NotEquals, value="0")
24 ],
25 response_format=ExportFormat.CSV,
26)
27
28data = client.export_table_data(id=table.id, query=request)
29
30# Write the export data to a file
31with open(f"{table.name}.csv", "wb") as f:
32 copyfileobj(data, f)
33
34# Alternatively, load the export data into a pandas dataframe
35# import pandas as pd
36# df = pd.read_csv(data)
Spec API
Overview
The SpecClient
class is the primary entry point of the Specification Compliance API.
When constructing a SpecClient
, you can pass an
HttpConfiguration
(like one retrieved from the
HttpConfigurationManager
), or let SpecClient
use the
default connection. The default connection depends on your environment.
With a SpecClient
object, you can:
Create and delete specifications under a product.
Modify any fields of an existing specification
Query for specifications on any fields using DynamicLinq syntax.
Get a specification using an Id.
Examples
Create, Get and Query Specifications
1from nisystemlink.clients.core import HttpConfiguration
2from nisystemlink.clients.spec import SpecClient
3from nisystemlink.clients.spec.models import (
4 Condition,
5 ConditionRange,
6 ConditionType,
7 CreateSpecificationsRequest,
8 NumericConditionValue,
9 QuerySpecificationsRequest,
10 SpecificationDefinition,
11 SpecificationLimit,
12 SpecificationType,
13)
14
15# Setup the server configuration to point to your instance of SystemLink Enterprise
16server_configuration = HttpConfiguration(
17 server_uri="https://yourserver.yourcompany.com",
18 api_key="YourAPIKeyGeneratedFromSystemLink",
19)
20client = SpecClient(configuration=server_configuration)
21
22# Create the spec requests
23product = "Amplifier"
24spec_requests = [
25 SpecificationDefinition(
26 product_id=product,
27 spec_id="spec1",
28 type=SpecificationType.PARAMETRIC,
29 category="Parametric Specs",
30 name="output voltage",
31 limit=SpecificationLimit(min=1.2, max=1.5),
32 unit="mV",
33 ),
34 SpecificationDefinition(
35 product_id=product,
36 spec_id="spec2",
37 type=SpecificationType.PARAMETRIC,
38 category="Parametric Specs",
39 name="input voltage",
40 limit=SpecificationLimit(min=0.02, max=0.15),
41 unit="mV",
42 conditions=[
43 Condition(
44 name="Temperature",
45 value=NumericConditionValue(
46 condition_type=ConditionType.NUMERIC,
47 range=[ConditionRange(min=-25, step=20, max=85)],
48 unit="C",
49 ),
50 ),
51 Condition(
52 name="Supply Voltage",
53 value=NumericConditionValue(
54 condition_type=ConditionType.NUMERIC,
55 discrete=[1.3, 1.5, 1.7],
56 unit="mV",
57 ),
58 ),
59 ],
60 ),
61 SpecificationDefinition(
62 product_id=product,
63 spec_id="spec3",
64 type=SpecificationType.FUNCTIONAL,
65 category="Noise Thresholds",
66 name="noise",
67 ),
68]
69
70# Create the specs on the server
71created_response = client.create_specs(CreateSpecificationsRequest(specs=spec_requests))
72
73# use get for first spec created
74if created_response.created_specs and len(created_response.created_specs) > 0:
75 created_spec = client.get_spec(created_response.created_specs[0].id)
76
77# You can query specs based on any field using DynamicLinq syntax.
78# These are just some representative examples.
79
80response = client.query_specs(QuerySpecificationsRequest(product_ids=[product]))
81all_product_specs = response.specs
82
83# Query based on spec id
84response = client.query_specs(
85 QuerySpecificationsRequest(product_ids=[product], filter='specId == "spec2"')
86)
87if response.specs:
88 spec2 = response.specs[0]
89
90# Query based on name
91response = client.query_specs(
92 QuerySpecificationsRequest(product_ids=[product], filter='name.Contains("voltage")')
93)
94voltage_specs = response.specs
95
96# Query based on Category
97response = client.query_specs(
98 QuerySpecificationsRequest(
99 product_ids=[product], filter='category == "Noise Thresholds"'
100 )
101)
102noise_category = response.specs
103print(noise_category)
Update and Delete Specifications
1from nisystemlink.clients.core import HttpConfiguration
2from nisystemlink.clients.spec import SpecClient
3from nisystemlink.clients.spec.models import (
4 QuerySpecificationsRequest,
5 SpecificationDefinition,
6 SpecificationType,
7 UpdateSpecificationsRequest,
8)
9
10# Setup the server configuration to point to your instance of SystemLink Enterprise
11server_configuration = HttpConfiguration(
12 server_uri="https://yourserver.yourcompany.com",
13 api_key="YourAPIKeyGeneratedFromSystemLink",
14)
15client = SpecClient(configuration=server_configuration)
16
17# The query and delete examples assume you have created the specs from the query_specs example
18product = "Amplifier"
19
20# update spec1 to change the block to "modifiedBlock"
21# query the original spec
22response = client.query_specs(
23 QuerySpecificationsRequest(product_ids=[product], filter='specId == "spec1"')
24)
25if response.specs:
26 original_spec1 = response.specs[0]
27 print(f"Original spec1 block: {original_spec1.block}")
28 print(f"Original spec1 version: {original_spec1.version}")
29
30# make the modifications
31modified_spec = SpecificationDefinition(
32 id=original_spec1.id,
33 product_id=original_spec1.product_id,
34 spec_id=original_spec1.spec_id,
35 type=SpecificationType.FUNCTIONAL,
36 keywords=["work", "reviewed"],
37 block="modifiedBlock",
38 version=original_spec1.version,
39 workspace=original_spec1.workspace,
40)
41update_response = client.update_specs(
42 specs=UpdateSpecificationsRequest(specs=[modified_spec])
43)
44if update_response and update_response.updated_specs:
45 print(f"New spec1 version: {update_response.updated_specs[0].version}")
46
47# query again to see new version
48response = client.query_specs(
49 QuerySpecificationsRequest(product_ids=[product], filter='specId == "spec1"')
50)
51if response.specs:
52 original_spec1 = response.specs[0]
53 print(f"Modified spec1 block: {original_spec1.block}")
54
55# delete all the specs for the product
56# query all specs
57response = client.query_specs(QuerySpecificationsRequest(product_ids=[product]))
58if response.specs:
59 client.delete_specs(ids=[spec.id for spec in response.specs if spec.id])
File API
Overview
The FileClient
class is the primary entry point of the File API.
When constructing a FileClient
, you can pass an
HttpConfiguration
(like one retrieved from the
HttpConfigurationManager
), or let FileClient
use the
default connection. The default connection depends on your environment.
With a FileClient
object, you can:
Get the list of files, download and delete files
Examples
Get the metadata of a File using its Id and download it.
1"""Example to download a file from SystemLink."""
2
3from shutil import copyfileobj
4
5from nisystemlink.clients.file import FileClient
6
7client = FileClient()
8
9file_id = "a55adc7f-5068-4202-9d70-70ca6a06bee9"
10
11# Fetch the file metadata to get the name
12files = client.get_files(ids=[file_id])
13
14if not files.available_files:
15 raise Exception(f"File ID {file_id} not found.")
16
17
18file_name = "Untitled"
19
20file_properties = files.available_files[0].properties
21
22if file_properties:
23 file_name = file_properties["Name"]
24
25# Download the file using FileId with content inline
26content = client.download_file(id=file_id)
27
28# Write the content to a file
29with open(file_name, "wb") as f:
30 copyfileobj(content, f)
Upload a File from disk or memory to SystemLink
"""Example to upload a file to SystemLink."""
import io
from nisystemlink.clients.file import FileClient
client = FileClient()
workspace_id = None # Upload to default workspace of the auth key
# Upload file from disk
file_path = "path/to/your/file"
with open(file_path, "rb") as fp:
file_id = client.upload_file(file=fp, workspace=workspace_id)
print(f"Uploaded file from {file_path} to SystemLink with FileID - {file_id}")
# Upload file-like object from memory
test_file = io.BytesIO(b"This is an example file content.")
test_file.name = "File_From_Memory.txt" # assign a name to the file object
file_id = client.upload_file(file=test_file)
print(f"Uploaded file from memory to SystemLink with FileID - {file_id}")
Feeds API
Overview
The FeedsClient
class is the primary entry point of the Feeds API.
When constructing a FeedsClient
, you can pass an
HttpConfiguration
(like one retrieved from the
HttpConfigurationManager
), or let FeedsClient
use the
default connection. The default connection depends on your environment.
With a FeedsClient
object, you can:
Get the list of feeds, create feed, upload package to feed and delete feed.
Examples
Create a new feed.
1"""Functionality of creating feeds APIs."""
2
3from nisystemlink.clients.core import ApiException, HttpConfiguration
4from nisystemlink.clients.feeds._feeds_client import FeedsClient
5from nisystemlink.clients.feeds.models import (
6 CreateFeedRequest,
7 Platform,
8)
9
10# Update the constants.
11FEED_NAME = ""
12FEED_DESCRIPTION = ""
13PLATFORM = Platform.WINDOWS
14WORKSPACE_ID = (
15 None # None uses Default workspace. Replace with Systemlink workspace id.
16)
17
18server_url = "" # SystemLink API URL
19server_api_key = "" # SystemLink API key
20
21# Provide the valid API key and API URL for client initialization.
22client = FeedsClient(HttpConfiguration(api_key=server_api_key, server_uri=server_url))
23
24# Creating Feeds.
25try:
26 feed_request = CreateFeedRequest(
27 name=FEED_NAME,
28 description=FEED_DESCRIPTION,
29 platform=PLATFORM,
30 workspace=WORKSPACE_ID,
31 )
32 feed_details = client.create_feed(feed=feed_request)
33
34 print("Feed created Successfully.")
35 print(f"Created feed details: {feed_details}")
36
37except ApiException as exp:
38 print(exp)
39except Exception as exp:
40 print(exp)
Query feeds and upload a package to feed.
1"""Functionality of uploading & querying feeds APIs."""
2
3from nisystemlink.clients.core import ApiException, HttpConfiguration
4from nisystemlink.clients.feeds._feeds_client import FeedsClient
5from nisystemlink.clients.feeds.models import Platform
6from nisystemlink.clients.feeds.utilities._get_feed_details import get_feed_by_name
7
8# Update the constants.
9FEED_NAME = ""
10PLATFORM = None
11FEED_DESCRIPTION = ""
12PLATFORM = Platform.WINDOWS
13WORKSPACE_ID = (
14 None # None uses Default workspace. Replace with Systemlink workspace id.
15)
16PACKAGE_PATH = ""
17
18server_url = "" # SystemLink API URL
19server_api_key = "" # SystemLink API key
20
21# Provide the valid API key and API URL for client initialization.
22client = FeedsClient(HttpConfiguration(api_key=server_api_key, server_uri=server_url))
23
24# To upload a package to feed.
25try:
26 # Get ID of the Feed to upload by name
27 feeds = client.query_feeds(platform=PLATFORM, workspace=WORKSPACE_ID)
28 feed = get_feed_by_name(feeds=feeds, name=FEED_NAME)
29 feed_id = feed.id if feed else None
30
31 # Upload the package to Feed by ID
32 if feed_id:
33 client.upload_package(
34 feed_id=feed_id,
35 overwrite=True,
36 package_file_path=PACKAGE_PATH,
37 )
38 print("Package uploaded sucessfully.")
39
40except ApiException as exp:
41 print(exp)
42
43except Exception as exp:
44 print(exp)
Delete a feed.
1"""Functionality of deleting feed API."""
2
3from nisystemlink.clients.core import ApiException, HttpConfiguration
4from nisystemlink.clients.feeds._feeds_client import FeedsClient
5from nisystemlink.clients.feeds.models import Platform
6from nisystemlink.clients.feeds.utilities._get_feed_details import get_feed_by_name
7
8# Update the constants.
9FEED_NAME = ""
10PLATFORM = Platform.WINDOWS
11WORKSPACE_ID = (
12 None # None uses Default workspace. Replace with Systemlink workspace id.
13)
14
15server_url = "" # SystemLink API URL
16server_api_key = "" # SystemLink API key
17
18# Provide the valid API key and API URL for client initialization.
19client = FeedsClient(HttpConfiguration(api_key=server_api_key, server_uri=server_url))
20
21# Deleting Feed.
22try:
23 # Get ID of the Feed to delete by name
24 feeds = client.query_feeds(platform=PLATFORM, workspace=WORKSPACE_ID)
25 feed = get_feed_by_name(feeds=feeds, name=FEED_NAME)
26 feed_id = feed.id if feed else None
27
28 # Delete the Feed by ID
29 if feed_id:
30 client.delete_feed(id=feed_id)
31 print("Feed deleted successfully.")
32
33except ApiException as exp:
34 print(exp)
35except Exception as exp:
36 print(exp)
TestMonitor API (Results and Steps)
Overview
The TestMonitorClient
class is the primary entry point of the Test Monitor API
used to interact with test results (Results) and test steps (Steps).
When constructing a TestMonitorClient
, you can pass an
HttpConfiguration
(like one retrieved from the
HttpConfigurationManager
), or let TestMonitorClient
use the
default connection. The default connection depends on your environment.
With a TestMonitorClient
object, you can:
Create, update, query, and delete results
Create, update, query, and delete steps
Examples
Create, query, update, and delete some results
1from nisystemlink.clients.testmonitor import TestMonitorClient
2from nisystemlink.clients.testmonitor.models import (
3 CreateResultRequest,
4 QueryResultsRequest,
5 QueryResultValuesRequest,
6 ResultField,
7 Status,
8 StatusType,
9 UpdateResultRequest,
10)
11
12program_name = "Example Name"
13host_name = "Example Host"
14status_type = StatusType.PASSED
15
16
17def create_some_results():
18 """Create two example results on your server."""
19 new_results = [
20 CreateResultRequest(
21 part_number="Example 123 AA",
22 program_name=program_name,
23 host_name=host_name,
24 status=Status.PASSED(),
25 keywords=["original keyword"],
26 properties={"original property key": "yes"},
27 ),
28 CreateResultRequest(
29 part_number="Example 123 AA1",
30 program_name=program_name,
31 host_name=host_name,
32 status=Status(status_type=StatusType.CUSTOM, status_name="Custom"),
33 keywords=["original keyword"],
34 properties={"original property key": "original"},
35 ),
36 ]
37 create_response = client.create_results(new_results)
38 return create_response
39
40
41# Server configuration is not required when used with SystemLink Client or run through Jupyter on SystemLink
42server_configuration = None
43
44# # Example of setting up the server configuration to point to your instance of SystemLink Enterprise
45# server_configuration = HttpConfiguration(
46# server_uri="https://yourserver.yourcompany.com",
47# api_key="YourAPIKeyGeneratedFromSystemLink",
48# )
49
50client = TestMonitorClient(configuration=server_configuration)
51
52create_response = create_some_results()
53
54# Get all the results using the continuation token in batches of 100 at a time.
55response = client.get_results(take=100, return_count=True)
56all_results = response.results
57while response.continuation_token:
58 response = client.get_results(
59 take=100, continuation_token=response.continuation_token, return_count=True
60 )
61 all_results.extend(response.results)
62
63# use get for first result created
64created_result = client.get_result(create_response.results[0].id)
65
66# Query results without continuation
67query_request = QueryResultsRequest(
68 filter=f'status.statusType="{status_type.value}"', return_count=True
69)
70response = client.query_results(query_request)
71
72# Update the first result that you just created and replace the keywords
73updated_result = create_response.results[0]
74updated_result = UpdateResultRequest(
75 id=create_response.results[0].id,
76 keywords=["new keyword"],
77 properties={"new property key": "new value"},
78)
79update_response = client.update_results([updated_result], replace=True)
80
81# Query for just the ids of results that match the family
82values_query = QueryResultValuesRequest(
83 filter=f'programName="{program_name}"', field=ResultField.ID
84)
85values_response = client.query_result_values(query=values_query)
86
87# delete each created result individually by id
88for result in create_response.results:
89 client.delete_result(result.id)
90
91# Create some more and delete them with a single call to delete.
92create_response = create_some_results()
93client.delete_results([result.id for result in create_response.results])
Create, update, query, and delete steps
1from nisystemlink.clients.testmonitor import TestMonitorClient
2from nisystemlink.clients.testmonitor.models import (
3 CreateResultRequest,
4 CreateStepRequest,
5 NamedValue,
6 QueryStepsRequest,
7 QueryStepValuesRequest,
8 StepField,
9 StepIdResultIdPair,
10 UpdateStepRequest,
11)
12from nisystemlink.clients.testmonitor.models._status import Status
13from nisystemlink.clients.testmonitor.models._step_data import Measurement, StepData
14
15
16def create_test_result():
17 """Create example result on your server."""
18 new_results = [
19 CreateResultRequest(
20 part_number="Example 123 AA",
21 program_name="Example Name",
22 host_name="Example Host",
23 status=Status.PASSED(),
24 keywords=["original keyword"],
25 properties={"original property key": "yes"},
26 )
27 ]
28 create_response = client.create_results(new_results)
29 return create_response
30
31
32# Server configuration is not required when used with Systemlink Client or run throught Jupyter on SLE
33server_configuration = None
34
35# # Example of setting up the server configuration to point to your instance of SystemLink Enterprise
36# server_configuration = HttpConfiguration(
37# server_uri="https://yourserver.yourcompany.com",
38# api_key="YourAPIKeyGeneratedFromSystemLink",
39# )
40
41client = TestMonitorClient(configuration=server_configuration)
42
43# create a result to attach the steps to
44create_response = create_test_result()
45
46# Create the step requests
47result_id = create_response.results[0].result_id
48step_requests = [
49 CreateStepRequest(
50 step_id="step1",
51 name="step1",
52 result_id=result_id,
53 inputs=[
54 NamedValue(name="Temperature", value="35"),
55 NamedValue(name="Voltage", value="5"),
56 ],
57 ),
58 CreateStepRequest(
59 step_id="step2",
60 name="step2",
61 result_id=result_id,
62 ),
63]
64
65# Create the steps
66create_response = client.create_steps(steps=step_requests)
67created_steps = create_response.steps
68print(create_response)
69
70# You can query steps based on any field using DynamicLinq syntax.
71# These are just some representative examples.
72# Query based on result id
73query_response = client.query_steps(
74 QueryStepsRequest(filter=f'resultId == "{result_id}"')
75)
76queried_steps = query_response.steps
77
78# query step name using query step values
79query_values_response = client.query_step_values(
80 QueryStepValuesRequest(
81 filter=f'resultId == "{result_id}"',
82 field=StepField.NAME,
83 )
84)
85
86# update the data of the step
87# extra properties of the measurements will be converted to string if not already a string
88update_response = client.update_steps(
89 steps=[
90 UpdateStepRequest(
91 step_id=step.step_id,
92 result_id=step.result_id,
93 data=StepData(
94 text="My output string",
95 parameters=[
96 Measurement(
97 name="Temperature",
98 status="Passed",
99 measurement="35",
100 lowLimit="30",
101 highLimit="40",
102 units="C",
103 comparisonType="Numeric",
104 spec_id="spec1",
105 spec_info={
106 "specKey": 10
107 }, # will be converted to string as '{"specKey": 10}'
108 )
109 ],
110 ),
111 )
112 for step in created_steps
113 ]
114)
115
116# delete all steps at once
117delete_response = client.delete_steps(
118 steps=[
119 StepIdResultIdPair(step_id=step.step_id, result_id=step.result_id)
120 for step in queried_steps
121 ]
122)
123
124create_response = client.create_steps(steps=step_requests)
125created_steps = create_response.steps
126
127# delete steps one by one
128for step in created_steps:
129 if step.step_id and step.result_id:
130 client.delete_step(result_id=step.result_id, step_id=step.step_id)
Notebook API
Overview
The NotebookClient
class is the primary entry point of the Notebook API.
When constructing a NotebookClient
, you can pass an
HttpConfiguration
(like one retrieved from the
HttpConfigurationManager
), or let NotebookClient
use the
default connection. The default connection depends on your environment.
With a NotebookClient
object, you can:
Create, update, query, and delete Notebooks
Create, get and query Notebook Executions
Examples
Create, query, update, and delete some notebooks.
1from nisystemlink.clients.core import HttpConfiguration
2from nisystemlink.clients.notebook import NotebookClient
3from nisystemlink.clients.notebook.models import NotebookMetadata, QueryNotebookRequest
4
5# Setup the server configuration to point to your instance of SystemLink Enterprise
6server_configuration = HttpConfiguration(
7 server_uri="https://yourserver.yourcompany.com",
8 api_key="YourAPIKeyGeneratedFromSystemLink",
9)
10client = NotebookClient(configuration=server_configuration)
11
12# Create a notebook with metadata and content
13metadata = NotebookMetadata(
14 name="Example Notebook",
15 parameters={"param1": "value1"},
16 properties={"property1": "value1"},
17)
18
19with open("example.ipynb", "rb") as file:
20 notebook_response = client.create_notebook(metadata=metadata, content=file)
21
22# Get the notebook by ID
23notebook = client.get_notebook("your_notebook_id")
24
25# Update the notebook with new metadata and content
26metadata = NotebookMetadata(
27 name="Updated Example Notebook",
28 parameters={"param1": "value2"},
29 properties={"property1": "value2"},
30)
31
32with open("example_updated.ipynb", "rb") as file:
33 notebook_response = client.update_notebook(
34 id="your_notebook_id",
35 metadata=metadata,
36 content=file,
37 )
38
39# Get notebook content by ID
40notebook_content = client.get_notebook_content("your_notebook_id")
41
42# Query notebook by name
43query_request = QueryNotebookRequest(
44 filter='name="Example Notebook"',
45)
46
47query_response = client.query_notebooks(query_request)
48
49# Query notebooks by take
50query_request = QueryNotebookRequest(take=2)
51query_response = client.query_notebooks(query_request)
52
53query_request = QueryNotebookRequest(
54 continuation_token=query_response.continuation_token,
55 take=1,
56)
57query_response = client.query_notebooks(query_request)
58
59# Delete the notebook by ID
60client.delete_notebook("your_notebook_id")
Create, query, retry, and cancel notebook executions.
1from nisystemlink.clients.core import HttpConfiguration
2from nisystemlink.clients.notebook import NotebookClient
3from nisystemlink.clients.notebook.models import (
4 CreateExecutionRequest,
5 ExecutionField,
6 ExecutionPriority,
7 ExecutionResourceProfile,
8 ExecutionSortField,
9 ExecutionStatus,
10 QueryExecutionsRequest,
11 ReportSettings,
12 ReportType,
13)
14
15# Setup the server configuration to point to your instance of SystemLink Enterprise
16server_configuration = HttpConfiguration(
17 server_uri="https://yourserver.yourcompany.com",
18 api_key="YourAPIKeyGeneratedFromSystemLink",
19)
20client = NotebookClient(configuration=server_configuration)
21
22
23# Create a notebook execution
24execution_request = CreateExecutionRequest(
25 notebook_id="your_notebook_id",
26 parameters={"param1": "value1"},
27 workspace_id="your_workspace_id",
28 timeout=300,
29 result_cache_period=3600,
30 report_settings=ReportSettings(
31 format=ReportType.HTML,
32 exclude_code=False,
33 ),
34 client_requests_id="your_client_request_id",
35 priority=ExecutionPriority.HIGH,
36 resource_profile=ExecutionResourceProfile.DEFAULT,
37)
38
39# Pass the list of execution requests to the create_executions method
40create_execution_response = client.create_executions([execution_request])
41
42# Get the execution by ID
43execution = client.get_execution_by_id("your_execution_id")
44
45# Query executions
46query_request = QueryExecutionsRequest(
47 filter=f"(status = {ExecutionStatus.FAILED.value}))",
48 order_by=ExecutionSortField.COMPLETED_AT,
49 descending=True,
50 projection=[
51 ExecutionField.ID,
52 ExecutionField.NOTEBOOK_ID,
53 ExecutionField.STATUS,
54 ],
55)
56
57query_executions_response = client.query_executions(query_request)
58
59# Cancel execution
60cancel_execution_response = client.cancel_executions(["your_execution_id"])
61
62# Retry execution
63retry_execution_response = client.retry_executions(["your_execution_id"])
64
65# Create execution from existing one
66run_new = client.create_executions_from_existing(["your_execution_id"])
Asset Management API
Overview
The AssetManagementClient
class is the primary entry point of the Asset Management API.
When constructing a AssetManagementClient
, you can pass an
HttpConfiguration
(like one retrieved from the
HttpConfigurationManager
), or let AssetManagementClient
use the
default connection. The default connection depends on your environment.
With a AssetManagementClient
object, you can:
Create, delete, get the list of assets and link files to assets.
Examples
create, delete, query asset and link files to assets.
1from nisystemlink.clients.assetmanagement import AssetManagementClient
2from nisystemlink.clients.assetmanagement.models import (
3 AssetBusType,
4 AssetDiscoveryType,
5 AssetLocationForCreate,
6 AssetPresence,
7 AssetPresenceStatus,
8 AssetType,
9 CreateAssetRequest,
10 ExternalCalibration,
11 QueryAssetsRequest,
12 SelfCalibration,
13 TemperatureSensor,
14)
15from nisystemlink.clients.core._http_configuration import HttpConfiguration
16
17
18server_configuration = HttpConfiguration(
19 server_uri="https://yourserver.yourcompany.com",
20 api_key="YourAPIKeyGeneratedFromSystemLink",
21)
22client = AssetManagementClient(configuration=server_configuration)
23
24create_assets_request = [
25 CreateAssetRequest(
26 model_number=4000,
27 model_name="NI PXIe-6368",
28 serial_number="01BB877A",
29 vendor_name="NI",
30 vendor_number="4244",
31 bus_type=AssetBusType.ACCESSORY,
32 name="PCISlot2",
33 asset_type=AssetType.DEVICE_UNDER_TEST,
34 firmware_version="A1",
35 hardware_version="12A",
36 visa_resource_name="vs-3144",
37 temperature_sensors=[TemperatureSensor(name="Sensor0", reading=25.8)],
38 supports_self_calibration=True,
39 supports_external_calibration=True,
40 custom_calibration_interval=24,
41 self_calibration=SelfCalibration(
42 temperature_sensors=[TemperatureSensor(name="Sensor0", reading=25.8)],
43 is_limited=False,
44 date="2022-06-07T18:58:05.000Z",
45 ),
46 is_NI_asset=True,
47 workspace="your-workspace-id",
48 location=AssetLocationForCreate(
49 state=AssetPresence(asset_presence=AssetPresenceStatus.PRESENT)
50 ),
51 external_calibration=ExternalCalibration(
52 temperature_sensors=[TemperatureSensor(name="Sensor0", reading=25.8)],
53 date="2022-06-07T18:58:05.000Z",
54 recommended_interval=10,
55 next_recommended_date="2023-11-14T20:42:11.583Z",
56 next_custom_due_date="2024-11-14T20:42:11.583Z",
57 resolved_due_date="2022-06-07T18:58:05.000Z",
58 ),
59 properties={"Key1": "Value1"},
60 keywords=["Keyword1"],
61 discovery_type=AssetDiscoveryType.MANUAL,
62 file_ids=["608a5684800e325b48837c2a"],
63 supports_self_test=True,
64 supports_reset=True,
65 partNumber="A1234 B5",
66 )
67]
68
69# Create an asset.
70create_assets_response = client.create_assets(assets=create_assets_request)
71
72created_asset_id = None
73if create_assets_response.assets and len(create_assets_response.assets) > 0:
74 created_asset_id = str(create_assets_response.assets[0].id)
75
76# Query assets using id.
77query_asset_request = QueryAssetsRequest(
78 ids=[created_asset_id],
79 skip=0,
80 take=1,
81 descending=False,
82 calibratable_only=False,
83 returnCount=False,
84)
85client.query_assets(query=query_asset_request)
86
87# Link files to the created asset.
88file_ids = ["sameple-file-id"]
89if created_asset_id:
90 link_files_response = client.link_files(
91 asset_id=created_asset_id, file_ids=file_ids
92 )
93
94# Delete the created asset.
95if created_asset_id is not None:
96 client.delete_assets(ids=[created_asset_id])
Systems API
Overview
The SystemsClient
class is the primary entry point of the Systems API.
When constructing a SystemsClient
, you can pass an
HttpConfiguration
(like one retrieved from the
HttpConfigurationManager
), or let SystemsClient
use the
default connection. The default connection depends on your environment.
With a SystemsClient
object, you can:
Create, query, and remove systems.
Examples
Create, query, and remove some systems.
1from nisystemlink.clients.core._http_configuration import HttpConfiguration
2from nisystemlink.clients.systems._systems_client import SystemsClient
3from nisystemlink.clients.systems.models._create_virtual_systems_request import (
4 CreateVirtualSystemRequest,
5)
6from nisystemlink.clients.systems.models._query_systems_request import (
7 QuerySystemsRequest,
8)
9
10# Setup the server configuration to point to your instance of
11# SystemLink Enterprise.
12server_configuration = HttpConfiguration(
13 server_uri="https://yourserver.yourcompany.com",
14 api_key="YourAPIKeyGeneratedFromSystemLink",
15)
16client = SystemsClient(configuration=server_configuration)
17
18# Systems request metadata.
19create_virtual_system_request: CreateVirtualSystemRequest = CreateVirtualSystemRequest(
20 alias="Python integration virtual system",
21 workspace="your-workspace-id",
22)
23
24# Create a virtual system.
25create_virtual_system_response = client.create_virtual_system(
26 create_virtual_system_request=create_virtual_system_request
27)
28
29minion_id = None
30
31if create_virtual_system_response and create_virtual_system_response.minionId:
32 minion_id = create_virtual_system_response.minionId
33
34# Query systems using id.
35query_systems_request = QuerySystemsRequest(
36 filter=f'id="{minion_id}"', projection="new(id, alias)"
37)
38
39client.query_systems(query=query_systems_request)
40
41# Delete the created systems.
42if minion_id is not None:
43 remove_systems = [minion_id]
44 client.remove_systems(tgt=remove_systems)
TestPlan API
Overview
The TestPlanClient
class is the primary entry point of the TestPlan API.
When constructing a TestPlanClient
, you can pass an
HttpConfiguration
(like one retrieved from the
HttpConfigurationManager
), or let TestPlanClient
use the
default connection. The default connection depends on your environment.
With a TestPlanClient
object, you can:
Create, query, get, update, schedule and delete TestPlans
Create, query and delete test plan templates
Examples
Create, query, get, update, schedule and delete TestPlans
1from datetime import datetime
2
3from nisystemlink.clients.core._http_configuration import HttpConfiguration
4from nisystemlink.clients.test_plan import TestPlanClient
5from nisystemlink.clients.test_plan.models import (
6 CreateTestPlanRequest,
7 Dashboard,
8 Job,
9 JobExecution,
10 ManualExecution,
11 QueryTestPlansRequest,
12 ScheduleTestPlanRequest,
13 ScheduleTestPlansRequest,
14 UpdateTestPlanRequest,
15 UpdateTestPlansRequest,
16)
17
18# Setup the server configuration to point to your instance of SystemLink Enterprise
19server_configuration = HttpConfiguration(
20 server_uri="https://yourserver.yourcompany.com",
21 api_key="YourAPIKeyGeneratedFromSystemLink",
22)
23client = TestPlanClient(configuration=server_configuration)
24
25create_test_plans_request = [
26 CreateTestPlanRequest(
27 name="Python integration test plan",
28 state="NEW",
29 description="Test plan for verifying integration flow",
30 assigned_to="test.user@example.com",
31 estimated_duration_in_seconds=86400,
32 properties={"env": "staging", "priority": "high"},
33 part_number="px40482",
34 dut_id="Sample-Dut_Id",
35 test_program="TP-Integration-001",
36 system_filter="os:linux AND arch:x64",
37 workspace="your_workspace_id",
38 file_ids_from_template=["file1", "file2"],
39 dashboard=Dashboard(
40 id="DashBoardId", variables={"product": "PXIe-4080", "location": "Lab1"}
41 ),
42 execution_actions=[
43 ManualExecution(action="boot", type="MANUAL"),
44 JobExecution(
45 action="run",
46 type="JOB",
47 jobs=[
48 Job(
49 functions=["run_test_suite"],
50 arguments=[["test_suite.py"]],
51 metadata={"env": "staging"},
52 )
53 ],
54 systemId="system-001",
55 ),
56 ],
57 )
58]
59
60# create a test plan
61created_test_plans_response = client.create_test_plans(
62 test_plans=create_test_plans_request
63)
64
65if created_test_plans_response.created_test_plans:
66 created_test_plan_id = created_test_plans_response.created_test_plans[0].id
67
68# Query test plan using id.
69query_test_plans_request = QueryTestPlansRequest(
70 skip=0, take=1, descending=False, returnCount=False
71)
72client.query_test_plans(query_request=query_test_plans_request)
73
74# Get test plan
75get_test_plan = client.get_test_plan(test_plan_id=created_test_plan_id)
76
77# Update test plan
78update_test_plans_request = UpdateTestPlansRequest(
79 test_plans=[
80 UpdateTestPlanRequest(
81 id=created_test_plan_id,
82 name="Updated Test Plan",
83 )
84 ]
85)
86updated_test_plan = client.update_test_plans(update_request=update_test_plans_request)
87
88# Schedule the test plan
89schedule_test_plans_request = ScheduleTestPlansRequest(
90 test_plans=[
91 ScheduleTestPlanRequest(
92 id=created_test_plan_id,
93 planned_start_date_time=datetime.strptime(
94 "2025-05-20T15:07:42.527Z", "%Y-%m-%dT%H:%M:%S.%fZ"
95 ),
96 estimated_end_date_time=datetime.strptime(
97 "2025-05-22T15:07:42.527Z", "%Y-%m-%dT%H:%M:%S.%fZ"
98 ),
99 system_id="fake-system",
100 )
101 ]
102)
103schedule_test_plan_response = client.schedule_test_plans(
104 schedule_request=schedule_test_plans_request
105)
106
107# Delete test plan
108client.delete_test_plans(ids=[created_test_plan_id])
Create, query and delete test plan templates.
1from nisystemlink.clients.core._http_configuration import HttpConfiguration
2from nisystemlink.clients.test_plan import TestPlanClient
3from nisystemlink.clients.test_plan.models import (
4 CreateTestPlanTemplateRequest,
5 Dashboard,
6 Job,
7 JobExecution,
8 ManualExecution,
9 QueryTestPlanTemplatesRequest,
10)
11
12
13# Setup the server configuration to point to your instance of SystemLink Enterprise
14server_configuration = HttpConfiguration(
15 server_uri="https://yourserver.yourcompany.com",
16 api_key="YourAPIKeyGeneratedFromSystemLink",
17)
18client = TestPlanClient(configuration=server_configuration)
19
20# Test plan template request metadata
21create_test_plan_template_request = [
22 CreateTestPlanTemplateRequest(
23 name="Python integration test plan template",
24 template_group="sample template group",
25 product_families=["FamilyA", "FamilyB"],
26 part_numbers=["PN-1001", "PN-1002"],
27 summary="Template for running integration test plans",
28 description="This template defines execution steps for integration workflows.",
29 test_program="TP-INT-002",
30 estimated_duration_in_seconds=86400,
31 system_filter="os:linux AND arch:x64",
32 execution_actions=[
33 ManualExecution(action="boot", type="MANUAL"),
34 JobExecution(
35 action="run",
36 type="JOB",
37 jobs=[
38 Job(
39 functions=["run_test_suite"],
40 arguments=[["test_suite.py"]],
41 metadata={"env": "staging"},
42 )
43 ],
44 systemId="system-001",
45 ),
46 ],
47 file_ids=["file1", "file2"],
48 workspace="your_workspace_id",
49 properties={"env": "staging", "priority": "high"},
50 dashboard=Dashboard(
51 id="DashBoardId", variables={"product": "PXIe-4080", "location": "Lab1"}
52 ),
53 )
54]
55
56# Create a test plan template
57create_test_plan_template_response = client.create_test_plan_templates(
58 test_plan_templates=create_test_plan_template_request
59)
60
61create_test_plan_template_id = None
62
63if (
64 create_test_plan_template_response.created_test_plan_templates
65 and create_test_plan_template_response.created_test_plan_templates[0].id
66):
67 create_test_plan_template_id = str(
68 create_test_plan_template_response.created_test_plan_templates[0].id
69 )
70
71# Query test plan templates using id
72query_test_plan_template_request = QueryTestPlanTemplatesRequest(
73 filter=f'id="{create_test_plan_template_id}"', take=1
74)
75
76client.query_test_plan_templates(
77 query_test_plan_templates=query_test_plan_template_request
78)
79
80# Delete the created test plan template.
81if create_test_plan_template_id is not None:
82 client.delete_test_plan_templates(ids=[create_test_plan_template_id])