NI Digital Write Task
Learn how to send digital commands to NI devices with Synnax.
For task lifecycle management, see the Task Basics page.
How Commands Work
Write tasks use command and state channels:
- Command channels (
_cmd): Write values here to set digital outputs (1 = high, 0 = low) - State channels (
_state): Reflect the current output state - Command time channels (
_cmd_time): Index channels storing command timestamps
When you write to a command channel, the task processes it and updates the state channel, providing acknowledgment that the command was executed.
Task Configuration Reference
| Parameter | Type | Required | Default | Description |
|---|---|---|---|---|
name | string | Yes | - | Human-readable task name |
state_rate | number | No | 20 | Rate (Hz) at which state channels are updated |
data_saving | boolean | No | false | Enable permanent storage in Synnax |
auto_start | boolean | No | false | Automatically start task after configuration |
channels | array | Yes | - | List of digital output channel configurations |
Channel Configuration
DO Channel NI-DAQmx C API Reference
| Parameter | Type | Required | Default | Description |
|---|---|---|---|---|
device | string | Yes | - | Device identifier |
port | number | Yes | - | Physical port |
line | number | Yes | - | Line number on the port |
Important Rules
- Command/State pairs: Each digital output requires both a command and state channel.
- One task per module: Only one running task can claim a module at a time.
- State rate: Higher state rates provide faster feedback but consume more resources.
How-To
Configure and run task
import synnax as sy
from synnax.hardware import ni
client = sy.Synnax()
# Retrieve device
digital_dev = client.hardware.devices.retrieve(name="Mod1_Digital")
# Create command time index
do_cmd_time = client.channels.create(
name="do_cmd_time",
is_index=True,
data_type=sy.DataType.TIMESTAMP,
retrieve_if_name_exists=True,
)
# Create command channels
do_0_0_cmd = client.channels.create(
name="do_0_0_cmd",
index=do_cmd_time.key,
data_type=sy.DataType.UINT8,
retrieve_if_name_exists=True,
)
do_0_1_cmd = client.channels.create(
name="do_0_1_cmd",
index=do_cmd_time.key,
data_type=sy.DataType.UINT8,
retrieve_if_name_exists=True,
)
do_1_0_cmd = client.channels.create(
name="do_1_0_cmd",
index=do_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="Digital Write Task",
state_rate=sy.Rate.HZ * 20,
data_saving=True,
channels=[
ni.DOLineChan(
cmd_channel=do_0_0_cmd.key,
state_channel=do_0_0_state.key,
device=digital_dev.key,
port=0,
line=0,
),
ni.DOLineChan(
cmd_channel=do_0_1_cmd.key,
state_channel=do_0_1_state.key,
device=digital_dev.key,
port=0,
line=1,
),
ni.DOLineChan(
cmd_channel=do_1_0_cmd.key,
state_channel=do_1_0_state.key,
device=digital_dev.key,
port=1,
line=0,
),
],
)
client.hardware.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.hardware.tasks.retrieve(name="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.hardware.tasks.configure(task) Configure and run task
import { Synnax, TimeStamp } from "@synnaxlabs/client";
const client = new Synnax();
// Retrieve device
const digitalDev = await client.hardware.devices.retrieve({ name: "Mod1_Digital" });
// Create command time index
const doCmdTime = await client.channels.create({
name: "do_cmd_time",
isIndex: true,
dataType: "timestamp",
retrieveIfNameExists: true,
});
// Create command channels
const do00Cmd = await client.channels.create({
name: "do_0_0_cmd",
index: doCmdTime.key,
dataType: "uint8",
retrieveIfNameExists: true,
});
const do01Cmd = await client.channels.create({
name: "do_0_1_cmd",
index: doCmdTime.key,
dataType: "uint8",
retrieveIfNameExists: true,
});
const do10Cmd = await client.channels.create({
name: "do_1_0_cmd",
index: doCmdTime.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.hardware.tasks.create({
name: "Digital Write Task",
type: "ni_digital_write",
config: JSON.stringify({
state_rate: 20,
data_saving: true,
channels: [
{
type: "do_line",
cmd_channel: do00Cmd.key,
state_channel: do00State.key,
device: digitalDev.key,
port: 0,
line: 0,
},
{
type: "do_line",
cmd_channel: do01Cmd.key,
state_channel: do01State.key,
device: digitalDev.key,
port: 0,
line: 1,
},
{
type: "do_line",
cmd_channel: do10Cmd.key,
state_channel: do10State.key,
device: digitalDev.key,
port: 1,
line: 0,
},
],
}),
});
// Start task
await task.executeCommandSync("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["do_0_0_state"]}`);
console.log(`Port 0 Line 1: ${frame["do_0_1_state"]}`);
console.log(`Port 1 Line 0: ${frame["do_1_0_state"]}`);
// Stop task
await task.executeCommandSync("stop");
await writer.close();
await streamer.close(); Edit task configuration
// Retrieve existing task
const task = await client.hardware.tasks.retrieve({ name: "Digital Write Task" });
// Parse and update configuration
const config = JSON.parse(task.config);
// Update task-level configuration
config.auto_start = true;
config.state_rate = 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.hardware.tasks.create({
key: task.key,
name: task.name,
type: task.type,
config: JSON.stringify(config),
});