Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

examples: remove PinMapClient class from _helpers.py #550

Merged
merged 3 commits into from
Dec 13, 2023
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
49 changes: 2 additions & 47 deletions examples/game_of_life/_helpers.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,44 +6,10 @@

import click
import grpc
from ni_measurementlink_service._internal.stubs.ni.measurementlink.pinmap.v1 import (
pin_map_service_pb2,
pin_map_service_pb2_grpc,
)
from ni_measurementlink_service.discovery import DiscoveryClient
from ni_measurementlink_service.measurement.service import GrpcChannelPool


class PinMapClient(object):
"""Class that communicates with the pin map service."""

def __init__(self, *, grpc_channel: grpc.Channel):
"""Initialize pin map client."""
self._client: pin_map_service_pb2_grpc.PinMapServiceStub = (
pin_map_service_pb2_grpc.PinMapServiceStub(grpc_channel)
)

def update_pin_map(self, pin_map_path: str) -> str:
"""Update registered pin map contents.

Create and register a pin map if a pin map resource for the specified pin map id is not
found.

Args:
pin_map_path: The file path of the pin map to register as a pin map resource.

Returns:
The resource id of the pin map that is registered to the pin map service.
"""
pin_map_path_obj = pathlib.Path(pin_map_path)
# By convention, the pin map id is the .pinmap file path.
request = pin_map_service_pb2.UpdatePinMapFromXmlRequest(
pin_map_id=pin_map_path, pin_map_xml=pin_map_path_obj.read_text(encoding="utf-8")
)
response: pin_map_service_pb2.PinMap = self._client.UpdatePinMapFromXml(request)
return response.pin_map_id


class GrpcChannelPoolHelper(GrpcChannelPool):
"""Class that manages gRPC channel lifetimes."""

Expand Down Expand Up @@ -77,26 +43,15 @@ def __init__(self, sequence_context: Any) -> None:
self._sequence_context = sequence_context

def get_active_pin_map_id(self) -> str:
"""Get the active pin map id from the NI.MeasurementLink.PinMapId temporary global variable.
"""Get the active pin map id from the NI.MeasurementLink.PinMapId runtime variable.

Returns:
The resource id of the pin map that is registered to the pin map service.
"""
return self._sequence_context.Engine.TemporaryGlobals.GetValString(
return self._sequence_context.Execution.RunTimeVariables.GetValString(
"NI.MeasurementLink.PinMapId", 0x0
)

def set_active_pin_map_id(self, pin_map_id: str) -> None:
"""Set the NI.MeasurementLink.PinMapId temporary global variable to the specified id.

Args:
pin_map_id:
The resource id of the pin map that is registered to the pin map service.
"""
self._sequence_context.Engine.TemporaryGlobals.SetValString(
"NI.MeasurementLink.PinMapId", 0x1, pin_map_id
)

def resolve_file_path(self, file_path: str) -> str:
"""Resolve the absolute path to a file using the TestStand search directories.

Expand Down
36 changes: 1 addition & 35 deletions examples/nidaqmx_analog_input/_helpers.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,44 +6,10 @@

import click
import grpc
from ni_measurementlink_service._internal.stubs.ni.measurementlink.pinmap.v1 import (
pin_map_service_pb2,
pin_map_service_pb2_grpc,
)
from ni_measurementlink_service.discovery import DiscoveryClient
from ni_measurementlink_service.measurement.service import GrpcChannelPool


class PinMapClient(object):
"""Class that communicates with the pin map service."""

def __init__(self, *, grpc_channel: grpc.Channel):
"""Initialize pin map client."""
self._client: pin_map_service_pb2_grpc.PinMapServiceStub = (
pin_map_service_pb2_grpc.PinMapServiceStub(grpc_channel)
)

def update_pin_map(self, pin_map_path: str) -> str:
"""Update registered pin map contents.

Create and register a pin map if a pin map resource for the specified pin map id is not
found.

Args:
pin_map_path: The file path of the pin map to register as a pin map resource.

Returns:
The resource id of the pin map that is registered to the pin map service.
"""
pin_map_path_obj = pathlib.Path(pin_map_path)
# By convention, the pin map id is the .pinmap file path.
request = pin_map_service_pb2.UpdatePinMapFromXmlRequest(
pin_map_id=pin_map_path, pin_map_xml=pin_map_path_obj.read_text(encoding="utf-8")
)
response: pin_map_service_pb2.PinMap = self._client.UpdatePinMapFromXml(request)
return response.pin_map_id


class GrpcChannelPoolHelper(GrpcChannelPool):
"""Class that manages gRPC channel lifetimes."""

Expand Down Expand Up @@ -77,7 +43,7 @@ def __init__(self, sequence_context: Any) -> None:
self._sequence_context = sequence_context

def get_active_pin_map_id(self) -> str:
"""Get the active pin map id from the NI.MeasurementLink.PinMapId temporary global variable.
"""Get the active pin map id from the NI.MeasurementLink.PinMapId runtime variable.

Returns:
The resource id of the pin map that is registered to the pin map service.
Expand Down
36 changes: 1 addition & 35 deletions examples/nidcpower_source_dc_voltage/_helpers.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,44 +6,10 @@

import click
import grpc
from ni_measurementlink_service._internal.stubs.ni.measurementlink.pinmap.v1 import (
pin_map_service_pb2,
pin_map_service_pb2_grpc,
)
from ni_measurementlink_service.discovery import DiscoveryClient
from ni_measurementlink_service.measurement.service import GrpcChannelPool


class PinMapClient(object):
"""Class that communicates with the pin map service."""

def __init__(self, *, grpc_channel: grpc.Channel):
"""Initialize pin map client."""
self._client: pin_map_service_pb2_grpc.PinMapServiceStub = (
pin_map_service_pb2_grpc.PinMapServiceStub(grpc_channel)
)

def update_pin_map(self, pin_map_path: str) -> str:
"""Update registered pin map contents.

Create and register a pin map if a pin map resource for the specified pin map id is not
found.

Args:
pin_map_path: The file path of the pin map to register as a pin map resource.

Returns:
The resource id of the pin map that is registered to the pin map service.
"""
pin_map_path_obj = pathlib.Path(pin_map_path)
# By convention, the pin map id is the .pinmap file path.
request = pin_map_service_pb2.UpdatePinMapFromXmlRequest(
pin_map_id=pin_map_path, pin_map_xml=pin_map_path_obj.read_text(encoding="utf-8")
)
response: pin_map_service_pb2.PinMap = self._client.UpdatePinMapFromXml(request)
return response.pin_map_id


class GrpcChannelPoolHelper(GrpcChannelPool):
"""Class that manages gRPC channel lifetimes."""

Expand Down Expand Up @@ -77,7 +43,7 @@ def __init__(self, sequence_context: Any) -> None:
self._sequence_context = sequence_context

def get_active_pin_map_id(self) -> str:
"""Get the active pin map id from the NI.MeasurementLink.PinMapId temporary global variable.
"""Get the active pin map id from the NI.MeasurementLink.PinMapId runtime variable.

Returns:
The resource id of the pin map that is registered to the pin map service.
Expand Down
36 changes: 1 addition & 35 deletions examples/nidigital_spi/_helpers.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,44 +6,10 @@

import click
import grpc
from ni_measurementlink_service._internal.stubs.ni.measurementlink.pinmap.v1 import (
pin_map_service_pb2,
pin_map_service_pb2_grpc,
)
from ni_measurementlink_service.discovery import DiscoveryClient
from ni_measurementlink_service.measurement.service import GrpcChannelPool


class PinMapClient(object):
"""Class that communicates with the pin map service."""

def __init__(self, *, grpc_channel: grpc.Channel):
"""Initialize pin map client."""
self._client: pin_map_service_pb2_grpc.PinMapServiceStub = (
pin_map_service_pb2_grpc.PinMapServiceStub(grpc_channel)
)

def update_pin_map(self, pin_map_path: str) -> str:
"""Update registered pin map contents.

Create and register a pin map if a pin map resource for the specified pin map id is not
found.

Args:
pin_map_path: The file path of the pin map to register as a pin map resource.

Returns:
The resource id of the pin map that is registered to the pin map service.
"""
pin_map_path_obj = pathlib.Path(pin_map_path)
# By convention, the pin map id is the .pinmap file path.
request = pin_map_service_pb2.UpdatePinMapFromXmlRequest(
pin_map_id=pin_map_path, pin_map_xml=pin_map_path_obj.read_text(encoding="utf-8")
)
response: pin_map_service_pb2.PinMap = self._client.UpdatePinMapFromXml(request)
return response.pin_map_id


class GrpcChannelPoolHelper(GrpcChannelPool):
"""Class that manages gRPC channel lifetimes."""

Expand Down Expand Up @@ -77,7 +43,7 @@ def __init__(self, sequence_context: Any) -> None:
self._sequence_context = sequence_context

def get_active_pin_map_id(self) -> str:
"""Get the active pin map id from the NI.MeasurementLink.PinMapId temporary global variable.
"""Get the active pin map id from the NI.MeasurementLink.PinMapId runtime variable.

Returns:
The resource id of the pin map that is registered to the pin map service.
Expand Down
36 changes: 1 addition & 35 deletions examples/nidmm_measurement/_helpers.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,44 +6,10 @@

import click
import grpc
from ni_measurementlink_service._internal.stubs.ni.measurementlink.pinmap.v1 import (
pin_map_service_pb2,
pin_map_service_pb2_grpc,
)
from ni_measurementlink_service.discovery import DiscoveryClient
from ni_measurementlink_service.measurement.service import GrpcChannelPool


class PinMapClient(object):
"""Class that communicates with the pin map service."""

def __init__(self, *, grpc_channel: grpc.Channel):
"""Initialize pin map client."""
self._client: pin_map_service_pb2_grpc.PinMapServiceStub = (
pin_map_service_pb2_grpc.PinMapServiceStub(grpc_channel)
)

def update_pin_map(self, pin_map_path: str) -> str:
"""Update registered pin map contents.

Create and register a pin map if a pin map resource for the specified pin map id is not
found.

Args:
pin_map_path: The file path of the pin map to register as a pin map resource.

Returns:
The resource id of the pin map that is registered to the pin map service.
"""
pin_map_path_obj = pathlib.Path(pin_map_path)
# By convention, the pin map id is the .pinmap file path.
request = pin_map_service_pb2.UpdatePinMapFromXmlRequest(
pin_map_id=pin_map_path, pin_map_xml=pin_map_path_obj.read_text(encoding="utf-8")
)
response: pin_map_service_pb2.PinMap = self._client.UpdatePinMapFromXml(request)
return response.pin_map_id


class GrpcChannelPoolHelper(GrpcChannelPool):
"""Class that manages gRPC channel lifetimes."""

Expand Down Expand Up @@ -77,7 +43,7 @@ def __init__(self, sequence_context: Any) -> None:
self._sequence_context = sequence_context

def get_active_pin_map_id(self) -> str:
"""Get the active pin map id from the NI.MeasurementLink.PinMapId temporary global variable.
"""Get the active pin map id from the NI.MeasurementLink.PinMapId runtime variable.

Returns:
The resource id of the pin map that is registered to the pin map service.
Expand Down
36 changes: 1 addition & 35 deletions examples/nifgen_standard_function/_helpers.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,44 +6,10 @@

import click
import grpc
from ni_measurementlink_service._internal.stubs.ni.measurementlink.pinmap.v1 import (
pin_map_service_pb2,
pin_map_service_pb2_grpc,
)
from ni_measurementlink_service.discovery import DiscoveryClient
from ni_measurementlink_service.measurement.service import GrpcChannelPool


class PinMapClient(object):
"""Class that communicates with the pin map service."""

def __init__(self, *, grpc_channel: grpc.Channel):
"""Initialize pin map client."""
self._client: pin_map_service_pb2_grpc.PinMapServiceStub = (
pin_map_service_pb2_grpc.PinMapServiceStub(grpc_channel)
)

def update_pin_map(self, pin_map_path: str) -> str:
"""Update registered pin map contents.

Create and register a pin map if a pin map resource for the specified pin map id is not
found.

Args:
pin_map_path: The file path of the pin map to register as a pin map resource.

Returns:
The resource id of the pin map that is registered to the pin map service.
"""
pin_map_path_obj = pathlib.Path(pin_map_path)
# By convention, the pin map id is the .pinmap file path.
request = pin_map_service_pb2.UpdatePinMapFromXmlRequest(
pin_map_id=pin_map_path, pin_map_xml=pin_map_path_obj.read_text(encoding="utf-8")
)
response: pin_map_service_pb2.PinMap = self._client.UpdatePinMapFromXml(request)
return response.pin_map_id


class GrpcChannelPoolHelper(GrpcChannelPool):
"""Class that manages gRPC channel lifetimes."""

Expand Down Expand Up @@ -77,7 +43,7 @@ def __init__(self, sequence_context: Any) -> None:
self._sequence_context = sequence_context

def get_active_pin_map_id(self) -> str:
"""Get the active pin map id from the NI.MeasurementLink.PinMapId temporary global variable.
"""Get the active pin map id from the NI.MeasurementLink.PinMapId runtime variable.

Returns:
The resource id of the pin map that is registered to the pin map service.
Expand Down
36 changes: 1 addition & 35 deletions examples/niscope_acquire_waveform/_helpers.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,44 +6,10 @@

import click
import grpc
from ni_measurementlink_service._internal.stubs.ni.measurementlink.pinmap.v1 import (
pin_map_service_pb2,
pin_map_service_pb2_grpc,
)
from ni_measurementlink_service.discovery import DiscoveryClient
from ni_measurementlink_service.measurement.service import GrpcChannelPool


class PinMapClient(object):
"""Class that communicates with the pin map service."""

def __init__(self, *, grpc_channel: grpc.Channel):
"""Initialize pin map client."""
self._client: pin_map_service_pb2_grpc.PinMapServiceStub = (
pin_map_service_pb2_grpc.PinMapServiceStub(grpc_channel)
)

def update_pin_map(self, pin_map_path: str) -> str:
"""Update registered pin map contents.

Create and register a pin map if a pin map resource for the specified pin map id is not
found.

Args:
pin_map_path: The file path of the pin map to register as a pin map resource.

Returns:
The resource id of the pin map that is registered to the pin map service.
"""
pin_map_path_obj = pathlib.Path(pin_map_path)
# By convention, the pin map id is the .pinmap file path.
request = pin_map_service_pb2.UpdatePinMapFromXmlRequest(
pin_map_id=pin_map_path, pin_map_xml=pin_map_path_obj.read_text(encoding="utf-8")
)
response: pin_map_service_pb2.PinMap = self._client.UpdatePinMapFromXml(request)
return response.pin_map_id


class GrpcChannelPoolHelper(GrpcChannelPool):
"""Class that manages gRPC channel lifetimes."""

Expand Down Expand Up @@ -77,7 +43,7 @@ def __init__(self, sequence_context: Any) -> None:
self._sequence_context = sequence_context

def get_active_pin_map_id(self) -> str:
"""Get the active pin map id from the NI.MeasurementLink.PinMapId temporary global variable.
"""Get the active pin map id from the NI.MeasurementLink.PinMapId runtime variable.

Returns:
The resource id of the pin map that is registered to the pin map service.
Expand Down
Loading