nyox-sdk 1.0.5

Skope NYOX SDK. Contains data reader for skope-fx session files and protocol definitions to directly connect to the NYOX controller

NYOX SDK for Python

A Python SDK for reading Skope NYOX session files and communicating with NYOX controllers.

Features

• Data Reader: Read and parse skope-fx session files (.h5 format) to access complete system configuration, field models, and scan data
• Protocol Client: Direct communication interface with NYOX controllers
• Cross-platform Support: Works on Windows and Linux with compiled extension modules for performance

Requirements

• Python 3.10 - 3.13
• h5py (HDF5 support)
• numpy
• cachetools
• jsonschema

Installation

pip install nyox-sdk

Quick Start

Reading Session Data

Please refer to the User Manual for detailed instructions on using the data reader.

from nyox_sdk import reader

# Load a session file
session = reader.Session("path/to/session.h5")

# Access system configuration
print(session.system_config.nucleus)
print(session.system_config.B0)

# Read scans
for scan in session.scans:
    print(f"Scan ID: {scan.id}")

Protocol Client

from nyox_sdk.protocol import NyoxClient

# Connect and send commands to controller
client = NyoxClient()

Examples Usage

Reader

# Example of how to use the reader module to access session and scan data from a session file created by skope-fx.

from nyox_sdk import reader

# load and print session
session_file = "Path/To/SessionFile.h5"
session = reader.Session(session_file)
print(session)

# iterate through scans
for scan in session.scans:
    nr_datapoints = f"{scan.nr_samples * scan.nr_channels * scan.nr_acquisitions:,}"
    print(str(scan.scan_nr) + " - " + nr_datapoints)

# select a specific scan
scan = session.get_scan(3)

# get information from the scan
print(scan)
print(scan.name)
print(scan.nr_acquisitions)
print(scan.dt)

# get trigger information
trigger = scan.trigger

# accessing raw data
raw_data_0 = scan.load_raw_data(0)                  # load raw data for first acquisition only
raw_data_all = scan.raw                             # access all raw data and save it to cache
data = scan.load_raw_data(1)                        # using cached raw data -> faster access

# accessing phase data
phase_data_0 = scan.load_phase(0)                   # calculating the phase for the first acquisition only
phase_data_all = scan.phase                         # get all phase data -> this will load all raw data if not cached before

# access the trajectory data
traj_0 = scan.calculate_trajectory(0)               # calculate the trajectory for the first acquisition only
traj_all = scan.trajectory                          # calculate all trajectories and save it to cache
traj_1 = scan.calculate_trajectory(1)               # using cached trajectories -> no recalculating needed

# change the field model
field_model = scan.get_field_model()
field_model.selected_spatial_basis = [True, True, True, True]
field_model.includes_coco_terms = False
scan.set_field_model(field_model)

# reset the field model to the original
scan.reset_field_model()

Protocol Client

"""
Simple example: Connect to a NYOX system and acquire data

Edit the configuration section below to match your setup.
This script connects, acquires a short scan, and plots the data.
"""

import os
import asyncio
from nyox_sdk.protocol import NyoxClient
from nyox_sdk.protocol import protocol_json as jsp

from nyox_sdk.reader.skope_processing import skope_processing_python as spp # For data processing
spp.version()
import numpy as np 
import matplotlib.pyplot as plt

# ----------------------
# User configuration
# ----------------------
DEVICE_ADDR = "127.0.0.1"     # NYOX system IP address
DATA_FOLDER = "data/"         # Folder to save data (must exist)
BASE_NAME = "example_stream"  # Base name for output files
ACQ_DURATION = 50e-3          # Acquisition duration (seconds)
NR_ACQ = 3                    # Number of acquisitions
NR_CHANNELS = 16              # Number of channels (typically 16 for NYOX)


async def acquire_data(client: NyoxClient):
    """Acquire a scan from the NYOX system."""
    # Get default scan definition
    resp = await client.get_defaultShortScanDef()

    # modify shortScanDef according to your needs
    scan_def = jsp.parse_shortscandef(resp)
    scan_def['nrAcquisitions'] = NR_ACQ
    scan_def['acqDuration'] = ACQ_DURATION

    # Start scan
    await client.start_scan(scan_def)
    await client.wait_for_stream_done()
    print("Acquisition complete.")
    await client.disconnect()

async def process_data(client: NyoxClient, payload: bytes):
    """Example of a custom data processing function that can be passed to NyoxClient."""
    # Get information from scan header
    header = client.scan_header
    # Convert raw bytes to numpy array
    raw_data = np.frombuffer(payload, dtype=np.uint32)
    try:
        raw_data = raw_data.reshape(-1, NR_CHANNELS)
        raw_data = raw_data.transpose()  # Shape: (NR_CHANNELS, samples)
    except Exception as e:
        print("Error reshaping data:", e)
        return
    print("Received data shape:", raw_data.shape)

def do_plot(data: np.ndarray):
    fig, ax = plt.subplots()
    for aq in range (data.shape[0]): 
        for ch in range(data.shape[1]):
            ax.plot(data[aq,ch,:], label=f"channel {ch} - acq {aq}")
    ax.set(xlabel='sample#', ylabel='value (raw)',
            title='Skope data acquisition')
    plt.legend()
    plt.show()

def main():
    # Check if the data folder exists and create it if not
    if not os.path.exists(DATA_FOLDER):
        try:
            os.makedirs(DATA_FOLDER)
            print(f"Created data folder: {DATA_FOLDER}")
        except Exception as e:
            print(f"Error creating data folder: {e}")
            return

    # Connect to NYOX system and acquire data
    print("Connecting to NYOX system at:", DEVICE_ADDR)
    try:
        client = NyoxClient(DEVICE_ADDR, acquire_data, outfile_prefix=os.path.join(DATA_FOLDER, BASE_NAME), data_callback=process_data)
        asyncio.run(client.run())
    except Exception as e:
        print("Error during acquisition:", e)
        return

    # Load and print shape of acquired data
    import numpy as np
    data_path = os.path.join(DATA_FOLDER, f"{BASE_NAME}-0000.bin")
    if not os.path.exists(data_path):
        print(f"Data file not found: {data_path}")
        return
    raw_data = np.fromfile(data_path, dtype=np.uint32)
    try:
        raw_data = raw_data.reshape(NR_ACQ, -1, NR_CHANNELS)
        raw_data = raw_data.transpose(0,2,1)  # Shape: (NR_ACQ, NR_CHANNELS, samples)
    except Exception as e:
        print("Error reshaping data:", e)
        return
    print("Data shape:", raw_data.shape)

    # Use skope_processing_python to convert raw data to magnitude and phase
    mag_data: np.ndarray = spp.pm_raw_to_magnitude(raw_data)
    phase_data: np.ndarray = spp.pm_raw_to_phase(raw_data,-1)

    do_plot(mag_data)
    do_plot(phase_data)

if __name__ == "__main__":
    main()

License

See End User License Agreement for details.

Documentation

Please refer to the User Manual for detailed instructions on using the data reader.

For more information, visit skope.swiss/software