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
"""Demonstrate streaming state control with AllegoClient."""
from time import sleep
from radiens_core import AllegoClient
def main() -> None:
with AllegoClient() as client:
client.start_streaming()
print("✅ Streaming started")
sleep(5) # stream for 5 seconds
status = client.get_status()
print(f"Stream mode: {status.streaming.stream_mode}")
cache = status.streaming.primary_cache_t_range
print(f"Cache: {cache[0]:.3f} - {cache[1]:.3f} s")
client.stop_streaming()
print("✅ Streaming stopped")
if __name__ == "__main__":
main()
Record State
Manage recordings — start, stop, and query recording state.
#!/usr/bin/env python3
"""Demonstrate recording state control with AllegoClient."""
from time import sleep
from radiens_core import AllegoClient
def main() -> None:
with AllegoClient() as client:
print("=== Recording Control ===")
client.start_streaming()
client.start_recording()
print("✅ Recording started")
sleep(2)
status = client.get_status()
print(f"Record mode: {status.recording.record_mode}")
print(f"Active filename: {status.recording.active_filename}")
sleep(2)
client.stop_recording()
print("✅ Recording stopped")
client.stop_streaming()
if __name__ == "__main__":
main()
Closed-Loop Signal Acquisition
Continuously read only the new signal samples that have arrived since the last fetch. Useful for real-time processing, closed-loop stimulation, or any application where you need to consume the live cache incrementally.
#!/usr/bin/env python3
"""Closed-loop signal acquisition that continuously fetches only new data.
Tracks the end time of the last successful fetch and requests only the
samples that have arrived since, polling every POLL_INTERVAL_SEC seconds.
Interrupt with Ctrl-C to stop.
"""
from time import sleep
import numpy as np
from radiens_core import AllegoClient
from radiens_core.exceptions import RadiensError
POLL_INTERVAL_SEC = 0.5
def main() -> None:
with AllegoClient() as client:
client.start_streaming()
# Let the cache fill briefly before the first fetch
sleep(1.0)
# Seed the cursor at the current cache head so we only fetch new data
status = client.get_status()
cursor = status.streaming.primary_cache_t_range[1]
print(f"Starting at cache head: {cursor:.3f} s — polling every {POLL_INTERVAL_SEC} s")
print("Press Ctrl-C to stop.\n")
try:
while True:
sleep(POLL_INTERVAL_SEC)
status = client.get_status()
cache_start, cache_end = status.streaming.primary_cache_t_range
if cache_end <= cursor:
continue # no new data yet
# Clamp in case the rolling cache has advanced past our cursor
fetch_start = max(cursor, cache_start)
try:
sigs = client.get_signals(time_range=[fetch_start, cache_end])
except RadiensError as e:
print(f"Fetch error (skipping window): {e}")
cursor = cache_end
continue
# --- Process the new chunk here ---
if sigs.amp is not None and sigs.amp.ndim == 2 and sigs.amp.shape[1] > 0:
n_ch, n_samp = sigs.amp.shape
rms = float(np.sqrt(np.mean(sigs.amp**2)))
print(f"[{fetch_start:.3f} - {cache_end:.3f} s] {n_ch} ch x {n_samp} samples RMS = {rms:.4f}")
cursor = cache_end
except KeyboardInterrupt:
print("\nStopped.")
finally:
client.stop_streaming()
if __name__ == "__main__":
main()
System Restart
Restart the Radiens backbone with a specific hardware mode.
#!/usr/bin/env python3
"""Demonstrate system restart with different hardware modes."""
from radiens_core import AllegoClient
from radiens_core.models.status import BackboneMode
def main() -> None:
with AllegoClient() as client:
print("=== System Restart Examples ===")
try:
client.restart(mode=BackboneMode.SMARTBOX_PRO)
print("✅ Restarted with SmartBox Pro mode")
except Exception as e:
print(f"❌ Restart failed: {e}")
try:
client.restart(mode=BackboneMode.SMARTBOX_SIM_GEN_SINE)
print("✅ Restarted with sine wave simulation mode")
except Exception as e:
print(f"❌ Restart failed: {e}")
try:
client.restart(mode=BackboneMode.XDAQ_ONE_REC)
print("✅ 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("Simulation modes:")
print(f" SMARTBOX_SIM_GEN_SINE: {BackboneMode.SMARTBOX_SIM_GEN_SINE}")
print(f" SMARTBOX_SIM_GEN_SPIKES: {BackboneMode.SMARTBOX_SIM_GEN_SPIKES}")
print("XDAQ modes:")
print(f" XDAQ_ONE_REC: {BackboneMode.XDAQ_ONE_REC}")
print(f" XDAQ_CORE_REC: {BackboneMode.XDAQ_CORE_REC}")
print("(See BackboneMode enum for the 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: