Using SiLA2 Devices with MADSci

Status: Experimental. SilaNodeClient is an early preview of native SiLA2 support in MADSci. The client surface, optional-dependency name, and binary/error handling may change as the broader migration lands. The umbrella project and design proposal are tracked in openspec/changes/sila2-native-node-design/ (issue #293).

This notebook demonstrates how to use the SilaNodeClient to communicate directly with SiLA2-based laboratory instruments from MADSci.

This notebook also doubles as the SiLA validation harness. It is executed end-to-end via papermill by just validate_nb_sila, and uses assert statements throughout to verify behavior against the live sila_example_server. If you’re reading it as a tutorial, mentally treat the asserts as “this is what the result should look like.”

What You’ll Learn¶

Connecting to a SiLA2 server using sila:// URLs
Discovering features and commands via get_info()
Checking server status via get_status()
Reading properties via get_state()
Executing unobservable (synchronous) commands
Executing observable (long-running) commands with await and polling
Handling binary data responses as ActionFiles
Error handling for unreachable servers

Prerequisites¶

This notebook requires:

The madsci.client[sila] extra installed (pip install "madsci.client[sila]"), which pulls in the sila2 SDK at a compatible version
The SiLA example server running on localhost:50052 (provided by the example lab compose stack)

%pip install "madsci.client[sila]"

Connecting to a SiLA2 Server¶

The SilaNodeClient uses sila://host:port URLs, just like RestNodeClient uses http:// URLs. The workcell manager’s find_node_client() automatically routes sila:// URLs to SilaNodeClient.

from madsci.client.node.sila_node_client import SilaNodeClient
from madsci.common.types.action_types import ActionRequest, ActionStatus
from rich import print

# Connect to the SiLA example server
client = SilaNodeClient(url="sila://localhost:50052")
print("Connected to SiLA server at sila://localhost:50052")

Server Introspection: `get_info()`¶

The get_info() method discovers the server’s features and commands, returning a NodeInfo with ActionDefinition entries for each SiLA command. Action names use dot notation: FeatureName.CommandName.

info = client.get_info()

print(f"Server name: {info.node_name}")
print(f"Module: {info.module_name}")
print(f"\nAvailable actions ({len(info.actions)}):")
for name, action_def in info.actions.items():
    print(f"  - {name}: {action_def.description}")

# Verify we can see the expected commands
action_names = set(info.actions.keys())
assert any("Greet" in name for name in action_names), "Expected Greet command"
assert any("CountDown" in name for name in action_names), "Expected CountDown command"
print("\nAll expected commands found.")

Server Status: `get_status()`¶

The get_status() method probes the server’s connectivity and reports whether it’s busy, errored, or disconnected.

status = client.get_status()

print(f"Ready: {status.ready}")
print(f"Busy: {status.busy}")
print(f"Errored: {status.errored}")
print(f"Disconnected: {status.disconnected}")
print(f"Running actions: {status.running_actions}")

assert status.ready, "Server should be ready"
assert not status.errored, "Server should not be errored"
assert not status.disconnected, "Server should not be disconnected"
print("\nServer is healthy and ready.")

Reading Properties: `get_state()`¶

The get_state() method reads all SiLA property values from the server, returning a dict keyed by FeatureName.PropertyName.

state = client.get_state()

print("Server state (all properties):")
for key, value in state.items():
    print(f"  {key}: {value}")

# Find the ServerUptime property
uptime_keys = [k for k in state if "ServerUptime" in k or "Uptime" in k]
if uptime_keys:
    uptime = state[uptime_keys[0]]
    print(f"\nServer uptime: {uptime:.1f} seconds")
    assert isinstance(uptime, (int, float)), "Uptime should be numeric"
    assert uptime > 0, "Uptime should be positive"
    print("Property reading works correctly.")
else:
    print("Note: ServerUptime property not found in state (may be filtered by callable check)")

Unobservable Commands: Synchronous Execution¶

Unobservable SiLA commands execute immediately and return the result. The action_name uses dot notation (Feature.Command) or short form if unambiguous.

# Find the correct qualified name for Greet
greet_name = [name for name in info.actions if "Greet" in name][0]
print(f"Using action name: {greet_name}")

request = ActionRequest(
    action_name=greet_name,
    args={"Name": "MADSci"},
)
result = client.send_action(request)

print(f"\nStatus: {result.status}")
print(f"Result: {result.json_result}")

assert result.status == ActionStatus.SUCCEEDED, f"Expected SUCCEEDED, got {result.status}"
assert "Greeting" in result.json_result, "Expected Greeting in result"
assert "MADSci" in result.json_result["Greeting"], "Expected name in greeting"
print(f"\nGreeting received: {result.json_result['Greeting']}")

Observable Commands: Blocking with `await_result=True`¶

Observable SiLA commands are long-running. With await_result=True (the default), send_action() blocks until the command completes.

import time

# Find the correct qualified name for CountDown
countdown_name = [name for name in info.actions if "CountDown" in name][0]
print(f"Using action name: {countdown_name}")

request = ActionRequest(
    action_name=countdown_name,
    args={"Count": 3},
)

start = time.time()
result = client.send_action(request, await_result=True, timeout=30)
elapsed = time.time() - start

print(f"\nStatus: {result.status}")
print(f"Result: {result.json_result}")
print(f"Elapsed: {elapsed:.1f}s (expected ~3s)")

assert result.status == ActionStatus.SUCCEEDED, f"Expected SUCCEEDED, got {result.status}"
assert "Message" in result.json_result, "Expected Message in result"
assert elapsed >= 2.5, "CountDown should take at least ~3 seconds"
print(f"\nObservable command completed: {result.json_result['Message']}")

Observable Commands: Polling with `await_result=False`¶

With await_result=False, send_action() returns immediately with RUNNING status. You can then poll with get_action_status() and retrieve results with get_action_result().

request = ActionRequest(
    action_name=countdown_name,
    args={"Count": 3},
)

# Fire and forget
result = client.send_action(request, await_result=False)
action_id = result.action_id

print(f"Initial status: {result.status}")
print(f"Action ID: {action_id}")
assert result.status == ActionStatus.RUNNING, f"Expected RUNNING, got {result.status}"

# Poll until complete
print("\nPolling for completion...")
poll_count = 0
while True:
    status = client.get_action_status(action_id)
    poll_count += 1
    print(f"  Poll {poll_count}: {status}")
    if status.is_terminal:
        break
    time.sleep(1)

# Get the final result
final_result = client.get_action_result(action_id)
print(f"\nFinal status: {final_result.status}")
print(f"Final result: {final_result.json_result}")

assert final_result.status == ActionStatus.SUCCEEDED, f"Expected SUCCEEDED, got {final_result.status}"
print("\nPolling workflow completed successfully.")

Binary Data and ActionFiles¶

SiLA commands can return binary data (bytes). The SilaNodeClient automatically:

Writes bytes to disk as .bin files in .madsci/sila_files/{action_id}/
Populates ActionResult.files as an ActionFiles instance with file paths
Base64-encodes the bytes in json_result for lightweight inspection

The example server’s GenerateData command returns a deterministic byte sequence, making it easy to verify the round-trip.

import base64
from pathlib import Path
from madsci.common.types.action_types import ActionFiles

# Find the GenerateData command
generate_name = [name for name in info.actions if "GenerateData" in name][0]
print(f"Using action name: {generate_name}")

# Generate 16 bytes of deterministic data
request = ActionRequest(
    action_name=generate_name,
    args={"NumBytes": 16},
)
result = client.send_action(request)

print(f"\nStatus: {result.status}")
assert result.status == ActionStatus.SUCCEEDED, f"Expected SUCCEEDED, got {result.status}"

# 1. Check that ActionResult.files is populated
print(f"\nActionResult.files: {result.files}")
assert result.files is not None, "Expected files to be populated for bytes response"
assert isinstance(result.files, ActionFiles), f"Expected ActionFiles, got {type(result.files)}"

# 2. Read the file back from disk and verify contents
file_path = result.files.Data
print(f"File path: {file_path}")
assert isinstance(file_path, Path), f"Expected Path, got {type(file_path)}"
assert file_path.exists(), f"File should exist at {file_path}"

file_contents = file_path.read_bytes()
expected = bytes(i % 256 for i in range(16))
assert file_contents == expected, f"File contents mismatch: {file_contents!r} != {expected!r}"
print(f"File contents ({len(file_contents)} bytes): {file_contents.hex()}")
print("File contents match expected bytes.")

# 3. Verify base64 encoding in json_result
b64_value = result.json_result["Data"]
print(f"\njson_result['Data'] (base64): {b64_value}")
decoded = base64.b64decode(b64_value)
assert decoded == expected, "Base64-decoded json_result should match original bytes"
print("Base64 round-trip verified.")

print("\nBinary data handling works correctly!")

Error Handling: Unreachable Server¶

When a SiLA server is unreachable, get_status() returns a NodeStatus with errored=True and disconnected=True rather than raising an exception.

# Connect to a non-existent server
bad_client = SilaNodeClient(url="sila://localhost:59999")
bad_status = bad_client.get_status()

print(f"Errored: {bad_status.errored}")
print(f"Disconnected: {bad_status.disconnected}")
if bad_status.errors:
    print(f"Error: {bad_status.errors[0].message}")

assert bad_status.errored, "Should be errored for unreachable server"
assert bad_status.disconnected, "Should be disconnected for unreachable server"
print("\nError handling works correctly.")

bad_client.close()

Cleanup¶

client.close()
print("Client closed. Notebook complete!")