Tecan FluentControl
Full Workflow Example
A complete end-to-end example connecting to FluentControl, running a named method, and sending XML commands at runtime.
This example brings together the steps from the previous guides into a single script. It uses the reusable step functions from the shared.steps.fluent_control module included in the UniteLabs workflow template, and extends the run phase with an XML command sent through the execution channel.
Prerequisites
- A Tecan FluentControl connector connected to the UniteLabs platform
- A FluentControl method available on the instrument
- The
sharedpackage from the UniteLabs workflow template - Familiarity with the XML command format (see Sending XML Commands)
Full Script
import asyncio
from shared.config.fluent_control import CONNECTOR_NAME, DEFAULT_METHOD_NAME
from shared.steps.fluent_control import (
check_execution_channel,
finish_command,
get_fluentcontrol_service,
prepare_method,
run_method,
)
async def run_workflow(
connector_name: str = CONNECTOR_NAME,
method_name: str = DEFAULT_METHOD_NAME,
):
# Phase 01: Connect and reset any open channel from a previous session
service = await get_fluentcontrol_service(connector_name=connector_name)
await check_execution_channel(service=service)
# Phase 02: Prepare and start the method, then wait for the XML channel to open
await prepare_method(service=service, method_name=method_name)
await run_method(service=service)
await asyncio.sleep(3) # wait for FluentControl to open the XML channel
# Phase 03: Send an XML command through the execution channel.
# This example sends a UserPromptStatement — FluentControl blocks until
# the operator presses OK on the touchscreen.
# Build XML command strings following the ScriptGroup envelope format
# described in Sending XML Commands.
user_prompt = (
"<ScriptGroup><Objects>"
'<Object Type="Tecan.Core.Scripting.UserPromptStatement">'
"<UserPromptStatement>"
"<Prompt>Verify the deck layout, then press OK to continue</Prompt>"
"<AutoClose>False</AutoClose><Timeout>1</Timeout>"
"<SoundFile></SoundFile><RepeatSound>False</RepeatSound>"
"<ChangeStatusLightColor>False</ChangeStatusLightColor>"
"<StatusLightColorString>#FFFFFF00</StatusLightColorString>"
"<IsBreakpoint>False</IsBreakpoint>"
"<IsDisabledForExecution>False</IsDisabledForExecution>"
"<LineNumber>1</LineNumber>"
"</UserPromptStatement></Object>"
"</Objects><Name></Name>"
"<IsBreakpoint>False</IsBreakpoint>"
"<IsDisabledForExecution>False</IsDisabledForExecution>"
"<LineNumber>0</LineNumber></ScriptGroup>"
)
await service.execution_channel_controller.execute_command(command=user_prompt)
# Phase 04: Close the XML channel and return FluentControl to EditMode
await finish_command(service=service)
asyncio.run(run_workflow())
Phase Summary
| Phase | Step functions used | Purpose |
|---|---|---|
| 01 Connect & reset | get_fluentcontrol_service, check_execution_channel | Locate the service; close any leftover channel |
| 02 Start method | prepare_method, run_method | Load and start the named FluentControl method |
| 03 Send commands | execute_command() (direct SDK call) | Stream XML commands into the running method |
| 04 Finalise | finish_command | Close the XML channel; return to EditMode |
Simulation mode
The shared.steps.fluent_control module includes FluentControlMock, a drop-in replacement for the live service that no-ops all calls. Use it during development to test the workflow structure without a connected instrument:
from shared.steps.fluent_control import FluentControlMock
service = FluentControlMock()
Pass FluentControlMock() wherever the live service is expected. All execute_command(), prepare_method(), run_method(), and finish_command() calls succeed silently.