Control Sequence Basics
The Synnax Python Client’s control
module makes
it easy to write automated sequences to control your hardware. The Synnax team focuses
on making this module as simple as possible - this page will guide you through the
process of writing your first control sequence.
Prerequisites
Before you can start writing control sequences, you’ll need to have:
- A Synnax cluster running
- The Synnax Python Client installed
Opening a Controller
The Controller
class is the main entry point for writing control sequences. At the start
of your script, you’ll need to open a controller using the control.acquire
method on
the Synnax client. This is best done using a context manager, as shown below:
import synnax as sy
# Open the client
client = sy.Synnax()
# Open the controller
with client.control.acquire(
read=["temperature", "daq_do_1_state"],
write=["daq_do_1_cmd"]
write_authorities=[sy.Authority.ABSOLUTE]
) as controller:
# Your control sequence here
The acquire
method takes a few arguments:
Argument | Description |
---|---|
read | A list of channels that your control sequence will need to read from. Any channels not present in this list will not be available to your control sequence. |
write | A list of channels that your sequence needs control over. Any channels not in this list cannot be commanded |
write_authorities | An optional authority or list of authorities for the channels you’re writing to. These are numbers between 0 and 255 that define the precedence of your sequence over other sequences that may be running. If you don’t provide this argument, the default authority takes absolute control (255). More on authorities later. |
Commanding a Channel
The simplest way to command a channel is to access it as a dictionary key on the controller and assign a value to it.
with client.control.acquire(
read=["pressure_1", "daq_do_1_state"],
write=["daq_do_1_cmd"]
write_authorities=[sy.Authority.ABSOLUTE]
) as controller:
# Set a 1 on the digital output channel
controller["daq_do_1_cmd"] = 1
Synnax will also automatically handle the conversion of values to the correct type for the
channel, so a True
value will be converted to 1
for a digital output channel:
controller["daq_do_1_cmd"] = True
When commanding multiple channels, you can use multiple assignments:
controller["daq_do_1_cmd"] = 1
controller["daq_do_2_cmd"] = 0
The controller will send each of these commands to the hardware in sequence. If you’d
like to send multiple commands at once, you can use the set
method:
controller.set({
"daq_do_1_cmd": True,
"daq_do_2_cmd": False,
})
Waiting for a Condition
The wait_until
method allows you to wait for a condition to be met before continuing.
This method accepts a function that takes in the current sensor values and returns
a boolean.
Lambda Expressions
We recommend relying on lambda
expressions to keep your code simple. A
lambda expression is a simple way to define a function in one line. Here’s an example
of a function that returns True
if the pressure sensor value is above 25:
# This function returns True if the pressure sensor value is above 25
def temperature_above_25(sensor_values):
return sensor_values["pressure_1"] > 25
# This is the same function as a lambda expression
temperature_above_25 = lambda sensor_values: sensor_values["pressure_1"] > 25
In future examples, we’ll use sv
as a shorthand for sensor_values
to keep the code
concise. Here’s how you can use a similar lambda
expression with the wait_until
method:
with client.control.acquire(
read=["pressure_1", "daq_do_1_state"],
write=["daq_do_1_cmd"]
write_authorities=[sy.Authority.ABSOLUTE]
) as controller:
# Wait until the pressure sensor value is above 25
controller.wait_until(lambda sv: sv["pressure_1"] > 25)
Adding a Timeout
The wait_until
method will continuously check the condition on every new value until
it returns True
. If you’d like to add a timeout to the wait, you can pass in a timeout
argument:
controller.wait_until(lambda sv: sv["pressure_1"] > 25, timeout=10)
You can check whether a condition timed out by checking the return value of the wait_until
method:
timed_out = controller.wait_until(lambda sv: sv["pressure_1"] > 25, timeout=10)
if timed_out:
print("The condition timed out")
Blocking While a Condition is True
The wait_while
method allows you to block the control sequence until a condition is
no longer true. This method is useful for waiting for a sensor value to reach a certain
point before continuing. Here’s an example of how you can use the wait_while
method:
with client.control.acquire(
read=["pressure_1", "daq_do_1_state"],
write=["daq_do_1_cmd"]
write_authorities=[sy.Authority.ABSOLUTE]
) as controller:
# Wait until the pressure sensor value is above 25
controller.wait_while(lambda sv: sv["pressure_1"] < 25)
In many ways, you can think of the wait_while
method as the opposite of the wait_until
method.
Asserting a Condition Remains True
The remains_true_for
method allows you to assert that a condition remains true for a
certain duration. This method is useful for ensuring that a sensor value remains within
a certain range for a period of time. Here’s an example of how you can use the
remains_true_for
method:
with client.control.acquire(
read=["pressure_1", "daq_do_1_state"],
write=["daq_do_1_cmd"]
write_authorities=[sy.Authority.ABSOLUTE]
) as controller:
# Assert that the pressure sensor value remains above 25 for 10 seconds
controller.remains_true_for(lambda sv: sv["pressure_1"] > 25, duration=10)
The duration
argument should be treated as a minimum duration, as the block is not
guaranteed to sleep for exactly the provided duration. The actual duration may be longer
due to the operating system’s scheduler. This block will not sleep for less than the
provided duration unless the condition is no longer true.
Remains True for a Certain Percentage of Samples
The remains_true_for
method also accepts a percentage
argument, which is a decimal
value that allows you to specify the percentage of samples that must meet the condition.
This is useful for ensuring that a sensor value remains within a certain range for at
least a certain percentage of the duration. Here’s an example of how you can use the
percentage
argument:
with client.control.acquire(
read=["pressure_1", "daq_do_1_state"],
write=["daq_do_1_cmd"]
write_authorities=[sy.Authority.ABSOLUTE]
) as controller:
# Assert that the pressure sensor value remains above 25 for 10 seconds
# for at least 90% of the samples
controller.remains_true_for(
lambda sv: sv["pressure_1"] > 25,
duration=10,
percentage=0.9
)