Examples
Healthcheck
Simple connectivity test — verify that AllegoClient can reach the server.
#!/usr/bin/env python3
"""Example usage of Healthcheck method."""
from radiens_core import AllegoClient
from radiens_core.exceptions import RadiensError
def main() -> None:
"""Demonstrate Healthcheck usage."""
with AllegoClient() as client:
try:
# Perform lightweight health check
client.healthcheck()
print("✅ System health check passed")
# For comprehensive system information, use get_status()
status = client.get_status()
print(f"System connected: {status.is_connected}")
print(f"Stream mode: {status.streaming.stream_mode}")
print(f"Record mode: {status.recording.record_mode}")
except RadiensError as e:
print(f"❌ Health check failed: {e}")
if __name__ == "__main__":
main()
Stream State
Monitor and control the streaming state with AllegoClient.
#!/usr/bin/env python3
"""Example usage of SetStreamState method."""
from time import sleep
from radiens_core import AllegoClient
from radiens_core.models.status import RecordMode, RecordStateRequest, StreamMode, StreamStateRequest
def main() -> None:
"""Demonstrate SetStreamState usage."""
with AllegoClient() as client:
# Turn streaming on
start_request = StreamStateRequest(mode=StreamMode.ON)
client.set_stream_state(start_request)
sleep(5) # Simulate some streaming time
start_request = RecordStateRequest(mode=RecordMode.ON)
client.set_record_state(start_request)
# Check status
status = client.get_status()
print(f"Current stream mode: {status.streaming.stream_mode}")
sleep(5) # Simulate some streaming time
stop_request = RecordStateRequest(mode=RecordMode.OFF)
client.set_record_state(stop_request)
sleep(5)
# Turn streaming off
stop_request = StreamStateRequest(mode=StreamMode.OFF)
client.set_stream_state(stop_request)
if __name__ == "__main__":
main()
Record State
Manage recordings — start, stop, and query recording state.
#!/usr/bin/env python3
"""Example usage of SetRecordState method."""
from time import sleep
from radiens_core import AllegoClient
from radiens_core.models.status import RecordMode, RecordStateRequest
def main() -> None:
"""Demonstrate SetRecordState usage."""
with AllegoClient() as client:
# System-level recording control (all ports)
print("=== System-Level Recording ===")
# Start recording (system level)
start_request = RecordStateRequest(mode=RecordMode.ON)
client.set_record_state(start_request)
print("✅ System recording started")
sleep(2) # Simulate some recording time
# Check status
status = client.get_status()
print(f"Current record mode: {status.recording.record_mode}")
print(f"Active filename: {status.recording.active_filename}")
sleep(2) # Simulate more recording time
# Stop recording (system level)
stop_request = RecordStateRequest(mode=RecordMode.OFF)
client.set_record_state(stop_request)
print("✅ System recording stopped")
if __name__ == "__main__":
main()
Stimulation Steps
Run a stimulation sequence (restart example).
#!/usr/bin/env python3
"""Example usage of Restart method."""
from asyncio import sleep
from radiens_core import AllegoClient
from radiens_core.models.status import BackboneMode, RestartRequest
def main() -> None:
"""Demonstrate Restart usage."""
with AllegoClient() as client:
print("=== System Restart Examples ===")
# Check current system status before restart
print("\\n1. Checking current system status...")
try:
status = client.get_status()
print(f"System connected: {status.is_connected}")
print(f"Current stream mode: {status.streaming.stream_mode}")
sleep(2) # Simulate some time before restart
except Exception as e:
print(f"Status check failed: {e}")
print("\\n2. Restarting with SmartBox Pro mode...")
try:
# Restart with SmartBox Pro hardware mode
restart_req = RestartRequest(mode=BackboneMode.SMARTBOX_PRO)
client.restart(restart_req)
sleep(2) # Simulate some time before restart
print("✅ System restarted with SmartBox Pro mode")
except Exception as e:
print(f"❌ Restart failed: {e}")
print("\\n3. Restarting with simulation mode...")
try:
# Restart with simulation mode for testing
sim_restart = RestartRequest(mode=BackboneMode.SMARTBOX_SIM_GEN_SINE)
client.restart(sim_restart)
sleep(2) # Simulate some time before restart
print("✅ System restarted with simulation sine wave generation")
except Exception as e:
print(f"❌ Restart failed: {e}")
print("\\n4. Restarting with XDAQ recording mode...")
try:
# Restart with XDAQ recording hardware
xdaq_restart = RestartRequest(mode=BackboneMode.XDAQ_ONE_REC)
client.restart(xdaq_restart)
sleep(2) # Simulate some time before restart
print("✅ System restarted with XDAQ ONE recording mode")
except Exception as e:
print(f"❌ Restart failed: {e}")
print("\\n=== Available Hardware Modes ===")
print("SmartBox modes:")
print(f" - SMARTBOX_PRO: {BackboneMode.SMARTBOX_PRO}")
print(f" - SMARTBOX_CLASSIC: {BackboneMode.SMARTBOX_CLASSIC}")
print("\\nSimulation modes:")
print(f" - SINE_GEN: {BackboneMode.SMARTBOX_SIM_GEN_SINE}")
print(f" - SPIKES_GEN: {BackboneMode.SMARTBOX_SIM_GEN_SPIKES}")
print("\\nXDAQ modes:")
print(f" - XDAQ_ONE_REC: {BackboneMode.XDAQ_ONE_REC}")
print(f" - XDAQ_CORE_REC: {BackboneMode.XDAQ_CORE_REC}")
print("\\n(See BackboneMode enum for complete list)")
if __name__ == "__main__":
main()
Data Curation
Convert and manage recorded data with CurateClient.
"""
Tutorial: Using CurateClient for Data Transformation
This script demonstrates how to:
1. Link an existing .xdat recording.
2. Perform a transformation (e.g., 4x downsampling) with a real-time progress bar.
3. Verify the metadata of the resulting file.
"""
from pathlib import Path
from radiens_core.curate_client import CurateClient
def main():
# --- CONFIGURATION (Change these to match your local setup) ---
DATA_DIR = Path("path/to/your/data")
INPUT_FILENAME = "recording_data.xdat"
# --------------------------------------------------------------
input_path = DATA_DIR / INPUT_FILENAME
# Generate output name based on input
# e.g., "recording_data.xdat" -> "recording_downsampled.xdat"
base_stem = INPUT_FILENAME.replace("_data.xdat", "").replace(".xdat", "")
output_name = f"{base_stem}_downsampled.xdat"
output_path = DATA_DIR / output_name
if not input_path.exists():
print(f"Error: Input file not found at {input_path}")
print("Please update DATA_DIR and INPUT_FILENAME in the script.")
return
print("--- Radiens CurateClient Tutorial ---")
try:
with CurateClient() as client:
# 1. Link the data file
print(f"Linking: {INPUT_FILENAME}")
meta = client.link_data_file(input_path)
print(f" ID: {meta.id}")
print(f" Sampling Rate: {meta.time_range.fs} Hz")
print(f" Duration: {meta.time_range.dur_sec:.2f} seconds")
# 2. Perform downsampling (4x)
# This will automatically show a tqdm progress bar in your terminal.
# NOTE: This will fail with a RadiensError if the output file already exists.
print("\nStarting Downsampling (4x)...")
new_meta = client.downsample(meta, factor=4, output_path=output_path)
# 3. Verify results
print("\nSuccess!")
print(f"New File: {new_meta.path}/{new_meta.base_name}")
print(f"New Sampling Rate: {new_meta.time_range.fs} Hz")
print(f"New Duration: {new_meta.time_range.dur_sec:.2f} seconds")
except Exception as e:
print(f"\nError during curation: {e}")
if __name__ == "__main__":
main()
Running Examples Locally
Copy any example above into a .py file and run it after installing radiens-core: