Add gRPC examples

[bkeryan]

Problem to Solve

There are no example programs in this repo demonstrating how to use the nidaqmx package with gRPC.

The only example of nidaqmx with gRPC is in another repo:

• https://github.com/ni/measurement-plugin-python/tree/main/examples/nidaqmx_analog_input

Proposed Solution

Add a small number of gRPC examples (1-3).

AB#3709351

[Nikongen]

In order do provide some information in the meantime

Preliminary examples

Here is information and examples I gathered during installation and testing

NI grpc server on Ubuntu 24.04 or Debian 13:

Installation

Steps

apt-get install git cmake g++ python3-venv ninja-build
git clone https://github.com/ni/grpc-device.git /usr/local/src/grpc-device
cd /usr/local/src/grpc-device
git submodule update --init --recursive
mkdir -p cmake/build
cd cmake/build
cmake -DCMAKE_BUILD_TYPE=Release ../..
make
chmod +x ni_grpc_device_server
cp ni_grpc_device_server /usr/local/bin
mkdir -p /etc/ni_grpc_device_server
cp server_config.json /etc/ni_grpc_device_server

Running

Running ./ni_grpc_device_server --daemonize --use-syslog creates a PID file /var/run/ni_grpc_device_server.pid .

I use a systemd service to run the server:

/etc/systemd/system/ni-grpc-server.service

[Unit]
Description=NI gRPC Device Server
After=network.target
Wants=network.target

[Service]
Type=forking
PIDFile=/run/ni_grpc_device_server.pid
ExecStart=/usr/local/bin/ni_grpc_device_server \
  --daemonize \
  --use-syslog \
  /etc/grpc-device/server_config.json

ExecStop=/bin/kill -TERM $MAINPID

Restart=on-failure
RestartSec=5
TimeoutStartSec=30
TimeoutStopSec=30

[Install]
WantedBy=multi-user.target

Updating

/usr/local/sbin/update-ni-grpc-device.sh

#!/bin/bash
set -euo pipefail

SERVICE=ni-grpc-device
SRC_DIR=/usr/local/src/grpc-device
BUILD_DIR=${SRC_DIR}/cmake/build
BIN_NAME=ni_grpc_device_server
INSTALL_PATH=/usr/local/bin/${BIN_NAME}

echo "Stopping service..."
systemctl stop ${SERVICE}

echo "Updating source..."
cd "${SRC_DIR}"
git pull
git submodule update --init --recursive

echo "Building..."
mkdir -p "${BUILD_DIR}"
cd "${BUILD_DIR}"
cmake -DCMAKE_BUILD_TYPE=Release ../..
make -j$(nproc)

echo "Installing binary..."
cp "${BIN_NAME}" "${INSTALL_PATH}"
chmod 755 "${INSTALL_PATH}"

echo "Starting service..."
systemctl start ${SERVICE}

echo "Update completed successfully."

/etc/systemd/system/ni-grpc-device-update.service

[Unit]
Description=Update and rebuild NI gRPC Device Server
Requires=ni-grpc-device.service
After=network.target
Conflicts=ni-grpc-device.service

[Service]
Type=oneshot
ExecStart=/usr/local/sbin/update-ni-grpc-device.sh
RemainAfterExit=no

[Install]
WantedBy=multi-user.target

grpc client

Python

One can use the standard ni packages on PyPi. To install grpc clients one has to specify e.g.

• local device: pip install nidaqmx
• grpc device: pip install nidaqmx[grpc]

Python example for nidaqmx

"""Test file for nidaqmx"""
import logging
# https://nidaqmx-python.readthedocs.io/en/stable/
# Examples: https://github.com/ni/nidaqmx-python
import nidaqmx
from nidaqmx.task import Task as ni_task
from nidaqmx import (
    constants as ni_const,
    Scale as ni_scale
)
from nidaqmx.errors import Error as ni_error
import grpc

logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger("NI-Test")

remote = True

if remote:
# https://github.com/ni/grpc-device/blob/main/examples/nidaqmx/analog-output.py
    grpc_options = nidaqmx.GrpcSessionOptions(
        grpc_channel=grpc.insecure_channel("10.10.10.15:31763"),
        session_name="grpc_test",
        initialization_behavior=nidaqmx.SessionInitializationBehavior.AUTO
    )
    system = nidaqmx.system.system.System.remote(grpc_options=grpc_options)

else:
    # Use local device
    grpc_options = None
    system = nidaqmx.system.System.local()


# nidaqmx driver version
logging.info("nidaqmx driver version: %s",system.driver_version)
# List of devices.
# For more see examples https://github.com/juliancabaleiro/nidaqmx-python-examples
for device in system.devices:
    logging.info("Found device: %s", device)

try: 
    with ni_task("grpc_test", grpc_options=grpc_options ) as do_task:
        # Add digital channel
        do_task.do_channels.add_do_chan("Dev1/port0")
        # Set output value
        do_task.write(ni_const.LogicFamily.FIVE_V.value)
        # Wait until tasks are done
        do_task.wait_until_done()
        logger.info("Test successful.")
except ni_error as e:
        logger.error("NiError during grpc test: %s", e)