ni.datastore.data

Public API for accessing the NI Data Store.

Classes

DataStoreClient	Data store client for publishing and reading data.
ErrorInformation	Represents error or exception information in case of measurement failure.
Outcome	Represents the outcome of a measurement or test operation.
PublishedCondition	Represents a condition that has been published to the data store.
PublishedMeasurement	Represents a measurement that has been published to the data store.
Step	Information about a step into which measurements and conditions are published.
TestResult	Information about a test result.

Package Contents

class ni.datastore.data.DataStoreClient(discovery_client=None, grpc_channel=None, grpc_channel_pool=None)

Data store client for publishing and reading data.

Initialize the DataStoreClient.

Parameters:

• discovery_client (ni.measurementlink.discovery.v1.client.DiscoveryClient | None) – An optional discovery client (recommended).

• grpc_channel (grpc.Channel | None) – An optional data store gRPC channel. Providing this channel will bypass discovery service resolution of the data store.

• grpc_channel_pool (ni_grpc_extensions.channelpool.GrpcChannelPool | None) – An optional gRPC channel pool (recommended).

__slots__ = ('_closed', '_discovery_client', '_grpc_channel', '_grpc_channel_pool', '_data_store_client',...

__enter__()

Enter the runtime context of the data store client.

Return type:

Self

__exit__(exc_type, exc_val, traceback)

Exit the runtime context of the data store client.

Parameters:

• exc_type (type[BaseException] | None)

• exc_val (BaseException | None)

• traceback (types.TracebackType | None)

Return type:

None

close()

Close the data store client and clean up resources that it owns.

Return type:

None

publish_condition(name, condition_type, value, step_id)

Publish a condition value to the data store.

Parameters:

• name (str) – An identifier describing the condition value. For example, “Voltage” or “Temperature”.

• condition_type (str) – The type of this condition. For example, “Upper Limit”, “Environment”, or “Setup”.

• value (object) – The single value for this condition to publish on the test step. This should be a scalar value that can be converted to the appropriate protobuf scalar type.

• step_id (str) – The ID of the step associated with this condition. This value is expected to be a parsable GUID.

Returns:

The condition id - the unique ID of the condition for

referencing in queries

Return type:

str

publish_condition_batch(name, condition_type, values, step_id)

Publish a batch of N values for a condition to the data store.

Parameters:

• name (str) – An identifier describing the condition values. For example, “Voltage” or “Temperature”.

• condition_type (str) – The type of this condition. For example, “Upper Limit”, “Environment”, or “Setup”.

• values (object) – The values for this condition across all publishes on the test step. This should be a Vector of N values.

• step_id (str) – The ID of the step associated with this batch of condition values. This value is expected to be a parsable GUID.

Returns:

The condition id - the unique ID of the condition for

referencing in queries

Return type:

str

publish_measurement(name, value, step_id, timestamp=None, outcome=Outcome.UNSPECIFIED, error_information=None, hardware_item_ids=tuple(), test_adapter_ids=tuple(), software_item_ids=tuple(), notes='')

Publish a single measurement value associated with a test step.

Parameters:

• name (str) – The name used for associating/grouping conceptually alike measurements across multiple publish iterations. For example, “Temperature” can be used for associating temperature readings across multiple iterations.

• value (object) –

  The value of the measurement being published. Supported types:

  • Scalar: Single float, int, str or boolean

  • Vector: Array of float, int, str or boolean values

  • DoubleAnalogWaveform: Analog waveform with double precision

  • DoubleXYData: XY coordinate data with double precision

  • I16AnalogWaveform: Analog waveform with 16-bit integer precision

  • DoubleComplexWaveform: Complex waveform with double precision

  • I16ComplexWaveform: Complex waveform with 16-bit integer precision

  • DoubleSpectrum: Frequency spectrum data with double precision

  • DigitalWaveform: Digital waveform data

• step_id (str) – The ID of the step associated with this measurement. This value is expected to be a parsable GUID.

• timestamp (hightime.datetime | None) – The timestamp of the measurement. If None, the current time will be used.

• outcome (ni.datastore.data._types._outcome.Outcome) – The outcome of the measurement (PASSED, FAILED, INDETERMINATE, or UNSPECIFIED).

• error_information (ni.datastore.data._types._error_information.ErrorInformation | None) – Error or exception information in case of measurement failure.

• hardware_item_ids (collections.abc.Iterable[str]) – The IDs of the hardware items associated with this measurement. These values are expected to be parsable GUIDs or aliases.

• test_adapter_ids (collections.abc.Iterable[str]) – The IDs of the test adapters associated with this measurement. These values are expected to be parsable GUIDs or aliases.

• software_item_ids (collections.abc.Iterable[str]) – The IDs of the software items associated with this measurement. These values are expected to be parsable GUIDs or aliases.

• notes (str) – Any notes to be associated with the captured measurement.

Returns:

The published measurement id.

Return type:

str

publish_measurement_batch(name, values, step_id, timestamps=tuple(), outcomes=tuple(), error_information=tuple(), hardware_item_ids=tuple(), test_adapter_ids=tuple(), software_item_ids=tuple(), notes='')

Publish multiple measurements at once for parametric sweeps.

Parameters:

• name (str) – The name used for associating/grouping conceptually alike measurements across multiple publish iterations. For example, “Temperature” can be used for associating temperature readings across multiple iterations.

• values (object) – The values of the measurement being published across N iterations.

• step_id (str) – The ID of the step associated with this measurement. This value is expected to be a parsable GUID.

• timestamps (collections.abc.Iterable[hightime.datetime]) – The timestamps corresponding to the N iterations of batched measurement being published. Can be empty (no timestamp info), single value (applied to all), or N values (one per measurement).

• outcomes (collections.abc.Iterable[ni.datastore.data._types._outcome.Outcome]) – The outcomes corresponding to the N iterations of batched measurement being published. Can be empty (no outcome info), single value (applied to all), or N values (one per measurement).

• error_information (collections.abc.Iterable[ni.datastore.data._types._error_information.ErrorInformation]) – The error information corresponding to the N iterations of batched measurement being published. Can be empty (no error info), single value (applied to all), or N values (one per measurement).

• hardware_item_ids (collections.abc.Iterable[str]) – The IDs of the hardware items associated with this measurement. These values are expected to be parsable GUIDs or aliases.

• test_adapter_ids (collections.abc.Iterable[str]) – The IDs of the test adapters associated with this measurement. These values are expected to be parsable GUIDs or aliases.

• software_item_ids (collections.abc.Iterable[str]) – The IDs of the software items associated with this measurement. These values are expected to be parsable GUIDs or aliases.

• notes (str) – Any notes to be associated with the published measurements.

Returns:

The IDs of the corresponding PublishedMeasurements. A single ID will be returned when publishing scalar measurement values. N IDs will be returned when publishing (N) non-scalar measurement values.

Return type:

Sequence[str]

read_condition_value(read_source: ni.datastore.data._types._published_condition.PublishedCondition, expected_type: Type[TRead]) → TRead

read_condition_value(read_source: ni.datastore.data._types._published_condition.PublishedCondition) → object

Read data published to the data store.

Parameters:

• read_source – The source from which to read data (PublishedCondition).

• expected_type – Optional type to validate the returned data against. If provided, a TypeError will be raised if the actual data type doesn’t match.

Returns:

The data retrieved from the data store. The return type depends on what was originally published: - Scalar conditions return as Vectors - Other types are returned as originally published If expected_type is specified, the return value is guaranteed to be of that type.

Raises:

TypeError – If expected_type is provided and the actual data type doesn’t match.

read_measurement_value(read_source: ni.datastore.data._types._published_measurement.PublishedMeasurement, expected_type: Type[TRead]) → TRead

read_measurement_value(read_source: ni.datastore.data._types._published_measurement.PublishedMeasurement) → object

Read data published to the data store.

Parameters:

• read_source – The source from which to read data (PublishedMeasurement).

• expected_type – Optional type to validate the returned data against. If provided, a TypeError will be raised if the actual data type doesn’t match.

Returns:

The data retrieved from the data store. The return type depends on what was originally published: - Scalar measurements return as Vectors - Other types are returned as originally published If expected_type is specified, the return value is guaranteed to be of that type.

Raises:

TypeError – If expected_type is provided and the actual data type doesn’t match.

create_step(step)

Create a new step in the data store.

A step is owned by a test result and is a logical grouping of published measurements and conditions. All measurements and conditions must be associated with a step.

Parameters:

step (ni.datastore.data._types._step.Step) – The metadata of the step to be created.

Returns:

The identifier of the created step.

Return type:

str

get_step(step_id)

Get the step associated with the given identifier.

Parameters:

step_id (str) – The identifier of the desired step.

Returns:

The metadata of the requested step.

Return type:

Step

get_measurement(measurement_id)

Get the measurement associated with the given identifier.

Parameters:

measurement_id (str) – The identifier of the desired measurement.

Returns:

The metadata of the requested measurement.

Return type:

PublishedMeasurement

get_condition(condition_id)

Get the condition associated with the given identifier.

Parameters:

condition_id (str) – The identifier of the desired condition.

Returns:

The metadata of the requested condition.

Return type:

PublishedCondition

create_test_result(test_result)

Create a test result object for publishing measurements.

Once a test result is created, you can publish an arbitrary number of measurements and conditions to a step which is owned by the test result.

Parameters:

test_result (ni.datastore.data._types._test_result.TestResult) – The metadata of the test result to be created.

Returns:

The test result ID. Generated if not specified in the request.

Return type:

str

get_test_result(test_result_id)

Get the test result associated with the given identifier.

Parameters:

test_result_id (str) – The ID of the desired test result. This value is expected to be a parsable GUID.

Returns:

The TestResult object that corresponds to the

requested ID.

Return type:

TestResult

query_conditions(odata_query='')

Query conditions using OData query syntax.

Parameters:

odata_query (str) – An OData query string. Example: “$filter=name eq ‘Value’”. An empty string will return all conditions. $expand, $count, and $select are not supported. For more information, see https://learn.microsoft.com/en-us/odata/concepts/ queryoptions-overview.

Returns:

The list of matching conditions. Each

item contains an id for retrieving the condition measurements, as well as the metadata associated with the condition.

Return type:

Sequence[PublishedCondition]

query_measurements(odata_query='')

Query measurements using OData query syntax.

Parameters:

odata_query (str) – An OData query string. Example: “$filter=name eq ‘Value’”. An empty string will return all measurements. $expand, $count, and $select are not supported. For more information, see https://learn.microsoft.com/en-us/odata/ concepts/queryoptions-overview.

Returns:

The list of matching measurements.

Each item contains an id for retrieving the measurement, as well as the metadata associated with the measurement.

Return type:

Sequence[PublishedMeasurement]

query_test_results(odata_query='')

Query test results using OData query syntax.

Parameters:

odata_query (str) – An OData query string. Example: “$filter=name eq ‘Value’”. An empty string will return all test results. $expand, $count, and $select are not supported. For more information, see https://learn.microsoft.com/en-us/odata/ concepts/queryoptions-overview.

Returns:

The list of matching test results. Each

item contains the metadata associated with the test result, including test result ID, name, timestamps, and other properties.

Return type:

Sequence[TestResult]

query_steps(odata_query='')

Query for steps matching the given OData query.

Parameters:

odata_query (str) – An OData query string. Example: “$filter=name eq ‘Value’”. An empty string will return all steps. $expand, $count, and $select are not supported. For more information, see https://learn.microsoft.com/en-us/odata/concepts/ queryoptions-overview.

Returns:

The list of steps that match the query.

Return type:

Sequence[Step]

class ni.datastore.data.ErrorInformation(*, error_code=0, message='', source='')

Represents error or exception information in case of measurement failure.

An ErrorInformation contains an error code, descriptive message, and source information to help identify and diagnose issues during measurement execution.

Initialize an ErrorInformation instance.

Parameters:

• error_code (int) – The numeric error code associated with the error.

• message (str) – A descriptive message explaining the error.

• source (str) – The source or location where the error occurred.

__slots__ = ('error_code', 'message', 'source')

error_code = 0

message = ''

source = ''

static from_protobuf(error_information_proto)

Create an ErrorInformation instance from a protobuf ErrorInformation message.

Parameters:

error_information_proto (ni.measurements.data.v1.data_store_pb2.ErrorInformation)

Return type:

ErrorInformation

to_protobuf()

Convert this ErrorInformation instance to a protobuf ErrorInformation message.

Return type:

ni.measurements.data.v1.data_store_pb2.ErrorInformation

__eq__(other)

Determine equality.

Parameters:

other (object)

Return type:

bool

__str__()

Return a string representation of the ErrorInformation.

Return type:

str

class ni.datastore.data.Outcome

Bases: enum.IntEnum

Represents the outcome of a measurement or test operation.

The Outcome enum indicates whether a measurement or test passed, failed, or had an indeterminate result.

Initialize self. See help(type(self)) for accurate signature.

UNSPECIFIED

The outcome is not specified or unknown.

PASSED

The measurement or test passed successfully.

FAILED

The measurement or test failed.

INDETERMINATE

The measurement or test result is indeterminate or inconclusive.

classmethod from_protobuf(outcome_proto)

Create an Outcome instance from a protobuf Outcome value.

Parameters:

outcome_proto (ni.measurements.data.v1.data_store_pb2.Outcome.ValueType) – The protobuf Outcome value.

Returns:

The corresponding Outcome enum value.

Raises:

ValueError – If the protobuf value doesn’t correspond to a known Outcome.

Return type:

Outcome

to_protobuf()

Convert this Outcome instance to a protobuf Outcome value.

Returns:

The corresponding protobuf Outcome value.

Return type:

ni.measurements.data.v1.data_store_pb2.Outcome.ValueType

class ni.datastore.data.PublishedCondition(*, id='', name='', condition_type='', step_id='', test_result_id='')

Represents a condition that has been published to the data store.

A published condition contains metadata about a condition value that was published, including an id for data retrieval and associated metadata like condition name, type, and associated step/test result IDs.

Initialize a PublishedCondition instance.

Parameters:

• id (str) – The unique identifier of the condition. This can be used to reference and find the condition in the data store.

• name (str) – The name of the condition.

• condition_type (str) – The type of the condition. For example, “Setup” or “Environment”.

• step_id (str) – The ID of the step with which this condition is associated.

• test_result_id (str) – The ID of the test result with which this condition is associated.

__slots__ = ('id', 'name', 'condition_type', 'step_id', 'test_result_id')

id = ''

name = ''

condition_type = ''

step_id = ''

test_result_id = ''

static from_protobuf(published_condition_proto)

Create a PublishedCondition instance from a protobuf PublishedCondition message.

Parameters:

published_condition_proto (ni.measurements.data.v1.data_store_pb2.PublishedCondition)

Return type:

PublishedCondition

to_protobuf()

Convert this PublishedCondition instance to a protobuf PublishedCondition message.

Return type:

ni.measurements.data.v1.data_store_pb2.PublishedCondition

__eq__(other)

Determine equality.

Parameters:

other (object)

Return type:

bool

__str__()

Return a string representation of the PublishedCondition.

Return type:

str

class ni.datastore.data.PublishedMeasurement(*, published_conditions=None, id='', test_result_id='', step_id='', software_item_ids=None, hardware_item_ids=None, test_adapter_ids=None, name='', value_type='', notes='', start_date_time=None, end_date_time=None, outcome=Outcome.UNSPECIFIED, parametric_index=0, error_information=None)

Represents a measurement that has been published to the data store.

A published measurement contains the measurement data and the associated metadata (name, type, timestamps, outcome, hardware, software, and test adapter IDs).

Initialize a PublishedMeasurement instance.

Parameters:

• published_conditions (Iterable[ni.datastore.data._types._published_condition.PublishedCondition] | None) – The published conditions associated with this measurement from the test step.

• id (str) – The unique identifier of the measurement. This can be used to reference and find the measurement in the data store.

• test_result_id (str) – The ID of the test result with which this measurement is associated.

• step_id (str) – The ID of the step with which this measurement is associated.

• software_item_ids (Iterable[str] | None) – The IDs of the software items associated with this measurement.

• hardware_item_ids (Iterable[str] | None) – The IDs of the hardware items associated with this measurement.

• test_adapter_ids (Iterable[str] | None) – The IDs of the test adapters associated with this measurement.

• name (str) – The name of the measurement.

• value_type (str) – The data type of the measurement value.

• notes (str) – Additional notes or comments about the measurement.

• start_date_time (hightime.datetime | None) – The start timestamp of the measurement.

• end_date_time (hightime.datetime | None) – The end timestamp of the measurement.

• outcome (ni.datastore.data._types._outcome.Outcome) – The outcome of the measurement (PASSED, FAILED, INDETERMINATE, or UNSPECIFIED).

• parametric_index (int) – The index of this measurement in a parametric sweep operation.

• error_information (ni.datastore.data._types._error_information.ErrorInformation | None) – Error or exception information in case of measurement failure.

__slots__ = ('_published_conditions', 'id', 'test_result_id', 'step_id', '_software_item_ids',...

property published_conditions: MutableSequence[ni.datastore.data._types._published_condition.PublishedCondition]

The published conditions associated with the published measurement.

Return type:

MutableSequence[ni.datastore.data._types._published_condition.PublishedCondition]

property software_item_ids: MutableSequence[str]

The software item IDs associated with the published measurement.

Return type:

MutableSequence[str]

property hardware_item_ids: MutableSequence[str]

The hardware item IDs associated with the published measurement.

Return type:

MutableSequence[str]

property test_adapter_ids: MutableSequence[str]

The test adapter IDs associated with the published measurement.

Return type:

MutableSequence[str]

id = ''

test_result_id = ''

step_id = ''

name = ''

value_type = ''

notes = ''

start_date_time = None

end_date_time = None

outcome

parametric_index = 0

error_information = None

static from_protobuf(published_measurement_proto)

Create a PublishedMeasurement instance from a protobuf PublishedMeasurement message.

Parameters:

published_measurement_proto (ni.measurements.data.v1.data_store_pb2.PublishedMeasurement)

Return type:

PublishedMeasurement

to_protobuf()

Convert this PublishedMeasurement instance to a protobuf PublishedMeasurement message.

Return type:

ni.measurements.data.v1.data_store_pb2.PublishedMeasurement

__eq__(other)

Determine equality.

Parameters:

other (object)

Return type:

bool

__str__()

Return a string representation of the PublishedMeasurement.

Return type:

str

class ni.datastore.data.Step(name, *, id='', parent_step_id='', test_result_id='', test_id='', step_type='', notes='', start_date_time=None, end_date_time=None, link='', extension=None, schema_id='', error_information=None, outcome=Outcome.UNSPECIFIED)

Information about a step into which measurements and conditions are published.

Represents a hierarchical execution step within a test result that can contain measurements and conditions. Steps are linked to a test result and can be organized into hierarchical structures using parent_step_id. Each step has test execution time, metadata, and optional extensions for custom metadata.

Initialize a Step instance.

Parameters:

• name (str) – Human-readable name of the step.

• id (optional) – Unique identifier for the step.

• parent_step_id (optional) – ID of the parent step if this is a nested step.

• test_result_id (optional) – ID of the test result this step belongs to.

• test_id (optional) – ID of the test associated with this step.

• step_type (optional) – Type or category of the step.

• notes (optional) – Additional notes or comments about the step.

• start_date_time (optional) – The start date and time of the step execution.

• end_date_time (optional) – The end date and time of the step execution.

• link (optional) – Optional link to external resources for this step.

• extension (optional) – Additional extension attributes as key-value pairs.

• schema_id (optional) – ID of the extension schema for validating extensions.

• error_information (optional) – Error or exception information in case of step failure.

• outcome (optional) – The outcome of the step (PASSED, FAILED, INDETERMINATE, or UNSPECIFIED).

__slots__ = ('name', 'id', 'parent_step_id', 'test_result_id', 'test_id', 'step_type', 'notes',...

property extension: MutableMapping[str, str]

The extension of the step.

Return type:

MutableMapping[str, str]

name

id = ''

parent_step_id = ''

test_result_id = ''

test_id = ''

step_type = ''

notes = ''

start_date_time = None

end_date_time = None

link = ''

schema_id = ''

error_information = None

outcome

static from_protobuf(step_proto)

Create a Step instance from a protobuf Step message.

Parameters:

step_proto (ni.measurements.data.v1.data_store_pb2.Step)

Return type:

Step

to_protobuf()

Convert this Step to a protobuf Step message.

Return type:

ni.measurements.data.v1.data_store_pb2.Step

__eq__(other)

Determine equality.

Parameters:

other (object)

Return type:

bool

__str__()

Return a string representation of the Step.

Return type:

str

class ni.datastore.data.TestResult(name, *, id='', uut_instance_id='', operator_id='', test_station_id='', test_description_id='', software_item_ids=None, hardware_item_ids=None, test_adapter_ids=None, start_date_time=None, end_date_time=None, outcome=Outcome.UNSPECIFIED, link='', extension=None, schema_id='', error_information=None)

Information about a test result.

Represents the results of executing a test sequence, including metadata about the execution environment (UUT, operator, test station), test execution time, outcome, and associations with hardware/software components and test adapters. Each step which references measurement data and conditions is associated with a test result.

Initialize a TestResult instance.

Parameters:

• name (str) – Human-readable name for the test result.

• id (optional) – Unique identifier for the test result.

• uut_instance_id (optional) – ID of the UUT instance that was tested.

• operator_id (optional) – ID of the operator who ran the test.

• test_station_id (optional) – ID of the test station used.

• test_description_id (optional) – ID of the test description that was executed.

• software_item_ids (optional) – IDs of software items used in the test.

• hardware_item_ids (optional) – IDs of hardware items used in the test.

• test_adapter_ids (optional) – IDs of test adapters used in the test.

• start_date_time (optional) – The start date and time of the test execution.

• end_date_time (optional) – The end date and time of the test execution.

• outcome (optional) – The outcome of the test execution (PASSED, FAILED, INDETERMINATE, or UNSPECIFIED).

• link (optional) – Optional link to external resources for this test result.

• extension (optional) – Additional extension attributes as key-value pairs.

• schema_id (optional) – ID of the extension schema for validating extensions.

• error_information (optional) – Error or exception information in case of test result failure.

__slots__ = ('name', 'id', 'uut_instance_id', 'operator_id', 'test_station_id', 'test_description_id',...

property software_item_ids: MutableSequence[str]

The software item IDs associated with the test result.

Return type:

MutableSequence[str]

property hardware_item_ids: MutableSequence[str]

The hardware item IDs associated with the test result.

Return type:

MutableSequence[str]

property test_adapter_ids: MutableSequence[str]

The test adapter IDs associated with the test result.

Return type:

MutableSequence[str]

property extension: MutableMapping[str, str]

The extension of the test result.

Return type:

MutableMapping[str, str]

name

id = ''

uut_instance_id = ''

operator_id = ''

test_station_id = ''

test_description_id = ''

start_date_time = None

end_date_time = None

outcome

link = ''

schema_id = ''

error_information = None

static from_protobuf(test_result_proto)

Create a TestResult instance from a protobuf TestResult message.

Parameters:

test_result_proto (ni.measurements.data.v1.data_store_pb2.TestResult)

Return type:

TestResult

to_protobuf()

Convert this TestResult to a protobuf TestResult message.

Return type:

ni.measurements.data.v1.data_store_pb2.TestResult

__eq__(other)

Determine equality.

Parameters:

other (object)

Return type:

bool

__str__()

Return a string representation of the TestResult.

Return type:

str