NI digital write task

Learn how to send digital commands to NI devices with Synnax.

This task follows the standard task lifecycle.

Task configuration reference

Table

Parameter	Type	Required	Default	Description
`device`	string	Yes	-	Device key to write to
`state_rate`	number	No	`10`	Rate (Hz) at which state channels are updated
`channels`	array	Yes	-	List of digital output channel configurations

Channel configuration reference

DO Channel NI-DAQmx C API Reference

Table

Parameter	Type	Required	Default	Description
`cmd_channel`	number	Yes	-	Synnax channel key for command values
`state_channel`	number	Yes	-	Synnax channel key for state feedback
`port`	number	Yes	-	Physical port
`line`	number	Yes	-	Line number on the port

Important rules

One task per module -> Only one running task can claim a module at a time.

How-to

When you create channels with a client library, give each command channel its own index channel. A writer that holds several commands on one index must write all of them in every frame.

Console

Python

TypeScript

Configure and run task

import synnax as sy
from synnax import ni

client = sy.Synnax()

# Retrieve device
digital_dev = client.devices.retrieve(name="Mod1_Digital")

# Create command channels, each with its own index
do_0_0_cmd_time = client.channels.create(
    name="do_0_0_cmd_time",
    is_index=True,
    data_type=sy.DataType.TIMESTAMP,
    retrieve_if_name_exists=True,
)
do_0_0_cmd = client.channels.create(
    name="do_0_0_cmd",
    index=do_0_0_cmd_time.key,
    data_type=sy.DataType.UINT8,
    retrieve_if_name_exists=True,
)
do_0_1_cmd_time = client.channels.create(
    name="do_0_1_cmd_time",
    is_index=True,
    data_type=sy.DataType.TIMESTAMP,
    retrieve_if_name_exists=True,
)
do_0_1_cmd = client.channels.create(
    name="do_0_1_cmd",
    index=do_0_1_cmd_time.key,
    data_type=sy.DataType.UINT8,
    retrieve_if_name_exists=True,
)
do_1_0_cmd_time = client.channels.create(
    name="do_1_0_cmd_time",
    is_index=True,
    data_type=sy.DataType.TIMESTAMP,
    retrieve_if_name_exists=True,
)
do_1_0_cmd = client.channels.create(
    name="do_1_0_cmd",
    index=do_1_0_cmd_time.key,
    data_type=sy.DataType.UINT8,
    retrieve_if_name_exists=True,
)

# Create state time index
do_state_time = client.channels.create(
    name="do_state_time",
    is_index=True,
    data_type=sy.DataType.TIMESTAMP,
    retrieve_if_name_exists=True,
)

# Create state channels
do_0_0_state = client.channels.create(
    name="do_0_0_state",
    index=do_state_time.key,
    data_type=sy.DataType.UINT8,
    retrieve_if_name_exists=True,
)
do_0_1_state = client.channels.create(
    name="do_0_1_state",
    index=do_state_time.key,
    data_type=sy.DataType.UINT8,
    retrieve_if_name_exists=True,
)
do_1_0_state = client.channels.create(
    name="do_1_0_state",
    index=do_state_time.key,
    data_type=sy.DataType.UINT8,
    retrieve_if_name_exists=True,
)

# Create and configure task
task = ni.DigitalWriteTask(
    name="NI Digital Write Task",
    device=digital_dev.key,
    state_rate=sy.Rate.HZ * 20,
    channels=[
        ni.DOChannel(
            cmd_channel=do_0_0_cmd.key,
            state_channel=do_0_0_state.key,
            port=0,
            line=0,
        ),
        ni.DOChannel(
            cmd_channel=do_0_1_cmd.key,
            state_channel=do_0_1_state.key,
            port=0,
            line=1,
        ),
        ni.DOChannel(
            cmd_channel=do_1_0_cmd.key,
            state_channel=do_1_0_state.key,
            port=1,
            line=0,
        ),
    ],
)

client.tasks.configure(task)

# Start task and write commands
with task.run():
    with client.open_writer(
        start=sy.TimeStamp.now(),
        channels=["do_0_0_cmd", "do_0_1_cmd", "do_1_0_cmd"],
    ) as writer:
        # Set outputs
        writer.write({
            "do_0_0_cmd": 1,  # Port 0 Line 0 high
            "do_0_1_cmd": 0,  # Port 0 Line 1 low
            "do_1_0_cmd": 1,  # Port 1 Line 0 high
        })

        # Read back states
        with client.open_streamer(["do_0_0_state", "do_0_1_state", "do_1_0_state"]) as streamer:
            frame = streamer.read()
            print(f"Port 0 Line 0: {frame['do_0_0_state']}")
            print(f"Port 0 Line 1: {frame['do_0_1_state']}")
            print(f"Port 1 Line 0: {frame['do_1_0_state']}")

Edit task configuration

# Retrieve existing task
task = client.tasks.retrieve(name="NI Digital Write Task")
task = ni.DigitalWriteTask(internal=task)

# Update task-level configuration
task.config.auto_start = True
task.config.state_rate = int(sy.Rate.HZ * 50)

# Update first channel configuration
task.config.channels[0].port = 0
task.config.channels[0].line = 2

# Update second channel configuration
task.config.channels[1].port = 0
task.config.channels[1].line = 3

# Update third channel configuration
task.config.channels[2].port = 1
task.config.channels[2].line = 1

# Apply changes
client.tasks.configure(task)

Configure and run task

import { Synnax, TimeStamp } from "@synnaxlabs/client";

const client = new Synnax();

// Retrieve device
const [digitalDev] = await client.devices.retrieve({ names: ["Mod1_Digital"] });

// Create command channels, each with its own index
const do00CmdTime = await client.channels.create(
  {
    name: "do_0_0_cmd_time",
    isIndex: true,
    dataType: "timestamp",
  },
  { retrieveIfNameExists: true },
);
const do00Cmd = await client.channels.create(
  {
    name: "do_0_0_cmd",
    index: do00CmdTime.key,
    dataType: "uint8",
  },
  { retrieveIfNameExists: true },
);

const do01CmdTime = await client.channels.create(
  {
    name: "do_0_1_cmd_time",
    isIndex: true,
    dataType: "timestamp",
  },
  { retrieveIfNameExists: true },
);
const do01Cmd = await client.channels.create(
  {
    name: "do_0_1_cmd",
    index: do01CmdTime.key,
    dataType: "uint8",
  },
  { retrieveIfNameExists: true },
);

const do10CmdTime = await client.channels.create(
  {
    name: "do_1_0_cmd_time",
    isIndex: true,
    dataType: "timestamp",
  },
  { retrieveIfNameExists: true },
);
const do10Cmd = await client.channels.create(
  {
    name: "do_1_0_cmd",
    index: do10CmdTime.key,
    dataType: "uint8",
  },
  { retrieveIfNameExists: true },
);

// Create state time index
const doStateTime = await client.channels.create(
  {
    name: "do_state_time",
    isIndex: true,
    dataType: "timestamp",
  },
  { retrieveIfNameExists: true },
);

// Create state channels
const do00State = await client.channels.create(
  {
    name: "do_0_0_state",
    index: doStateTime.key,
    dataType: "uint8",
  },
  { retrieveIfNameExists: true },
);

const do01State = await client.channels.create(
  {
    name: "do_0_1_state",
    index: doStateTime.key,
    dataType: "uint8",
  },
  { retrieveIfNameExists: true },
);

const do10State = await client.channels.create(
  {
    name: "do_1_0_state",
    index: doStateTime.key,
    dataType: "uint8",
  },
  { retrieveIfNameExists: true },
);

// Create and configure task
const task = await client.tasks.create({
  name: "NI Digital Write Task",
  rack: digitalDev.rack,
  type: "ni_digital_write",
  config: {
    device: digitalDev.key,
    stateRate: 20,
    channels: [
      {
        cmdChannel: do00Cmd.key,
        stateChannel: do00State.key,
        port: 0,
        line: 0,
      },
      {
        cmdChannel: do01Cmd.key,
        stateChannel: do01State.key,
        port: 0,
        line: 1,
      },
      {
        cmdChannel: do10Cmd.key,
        stateChannel: do10State.key,
        port: 1,
        line: 0,
      },
    ],
  },
});

// Start task
await task.start();

// Write commands
const writer = await client.openWriter({
  start: TimeStamp.now(),
  channels: ["do_0_0_cmd", "do_0_1_cmd", "do_1_0_cmd"],
});

await writer.write({
  do_0_0_cmd: 1, // Port 0 Line 0 high
  do_0_1_cmd: 0, // Port 0 Line 1 low
  do_1_0_cmd: 1, // Port 1 Line 0 high
});

// Read states
const streamer = await client.openStreamer([
  "do_0_0_state",
  "do_0_1_state",
  "do_1_0_state",
]);
const frame = await streamer.read();
console.log(`Port 0 Line 0: ${frame.get("do_0_0_state").at(-1)}`);
console.log(`Port 0 Line 1: ${frame.get("do_0_1_state").at(-1)}`);
console.log(`Port 1 Line 0: ${frame.get("do_1_0_state").at(-1)}`);

// Stop task
await task.stop();
await writer.close();
await streamer.close();

Edit task configuration

import { ni } from "@synnaxlabs/client";

// Retrieve existing task
const task = await client.tasks.retrieve({ name: "NI Digital Write Task" });

// Parse and update configuration
const config = ni.digitalWriteConfigZ.parse(task.config);

// Update task-level configuration
config.autoStart = true;
config.stateRate = 50;

// Update first channel configuration
config.channels[0].port = 0;
config.channels[0].line = 2;

// Update second channel configuration
config.channels[1].port = 0;
config.channels[1].line = 3;

// Update third channel configuration
config.channels[2].port = 1;
config.channels[2].line = 1;

// Apply changes
await client.tasks.create({
  key: task.key,
  rack: task.rack,
  name: task.name,
  type: task.type,
  config,
});

Digital read task

OPC UA