NI Digital Write Task
Learn how to send digital commands to NI devices with Synnax.
For the task lifecycle, see Task Basics.
Task Configuration Reference
Channel Configuration Reference
DO Channel NI-DAQmx C API Reference
Important Rules
- One task per module -> Only one running task can claim a module at a time.
How-To
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 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="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 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.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,
});