XyloAudio 3 Accelerated-Time Mode

XyloAudio 3 Accelerated-Time Mode#

Accelerated-Time mode allows users to observe how a model evolves over time in response to a given input, with minimal configuration. In this mode, an ordered list of spike events is processed on the XyloAudio 3 as quickly as possible, while simultaneously monitoring the network state. This mode is ideal for benchmarking and validating models.

To use Accelerated-Time mode, you should generate a list of input (spike) events ordered by timestep. Samna will automatically trigger the processing of these timesteps, ensuring that all events are delivered at their specified timestep. After each timestep, the network state is read out, and the details included in this samna.xyloAudio3.event.Readout event can be configured as needed. The events are processed at maximum speed, and it’s important to avoid interacting with the device while processing is ongoing. Once all input events have been processed, which can be confirmed by counting the Readout events, full access to the device is restored.

It’s important to note that Samna cannot automatically determine if there are more spikes after the last timestep. Therefore, to process the events in the final timestep, you must manually send a samna.xyloAudio3.event.TriggerProcessing event.

Below is an example that shows typical usage of accelerated mode. It uses the following packages:

- samna                 0.39.6

import samna

# Open the device and connect the source and sink nodes so we can communicate with Xylo.
board = samna.device.open_device("XyloAudio3TestBoard")
xylo = board.get_model()
source = samna.graph.source_to(xylo.get_sink_node())

# We are only interested in Readout events, so we make a filter graph to filter only these events.
# Important note: `graph` needs to be kept alive for the filter to work.
graph = samna.graph.EventFilterGraph()
_, etf, readout_sink = graph.sequential([xylo.get_source_node(), "XyloAudio3OutputEventTypeFilter", samna.graph.JitSink()])
etf.set_desired_type('xyloAudio3::event::Readout')


# Create a basic network with a simple one-to-one mapping of input to output neurons.
xylo_config = samna.xyloAudio3.configuration.XyloConfiguration()
input_count = 3
hidden_count = input_count
output_count = input_count
xylo_config.input.weights = [[127, 0, 0], [0, 127, 0], [0, 0, 127]] # shape(input_count, hidden_count)
xylo_config.hidden.weights = [[0, 0, 0], [0, 0, 0], [0, 0, 0]] # shape(hidden_count, hidden_count)
xylo_config.hidden.neurons = [samna.xyloAudio3.configuration.HiddenNeuron(threshold=127, v_mem_decay=1, i_syn_decay=1)] * hidden_count
xylo_config.readout.weights = [[127, 0, 0], [0, 127, 0], [0, 0, 127]] # shape(hidden_count, output_count)
xylo_config.readout.neurons = [samna.xyloAudio3.configuration.OutputNeuron(threshold=127, v_mem_decay=1, i_syn_decay=1)] * output_count

# Configure Xylo in Accelerated-Time mode and to work with input spike events.
xylo_config.operation_mode = samna.xyloAudio3.OperationMode.AcceleratedTime
xylo_config.input_source = samna.xyloAudio3.InputSource.SpikeEvents

# In Accelerated-Time mode we can use automatic state monitoring, by setting the neuron ids we want to monitor.
# Output Vmem and spikes are always activated, so we only monitor the hidden neurons.
xylo_config.debug.monitor_neuron_v_mem = [i for i in range(hidden_count)]
xylo_config.debug.monitor_neuron_spike = [i for i in range(hidden_count)]
# Output Isyn is not available by default, so we add both hidden and output neurons.
xylo_config.debug.monitor_neuron_i_syn = [i for i in range(hidden_count + output_count)]

# Send the configuration to Xylo
xylo.apply_configuration(xylo_config)


def get_current_timestep():
    """Utility function to obtain the current timestep.

    The `Readout` event always contains the current timestep so we could simply save it.
    But in case you forgot what timestep was last processed, you can obtain it as follows.
    """
    source.write([samna.xyloAudio3.event.TriggerReadout()])
    evts = readout_sink.get_n_events(1, timeout=3000)
    assert(len(evts) == 1)
    return evts[0].timestep


def evolve(input):
    """Continue to evolve the model with the given input.

    Args:
        input (list[list[int]]): Per timestep a list of integers to specify the number of spikes to send to that `neuron_id`.
                                Max number of spikes per timestep for a neuron is 15. All lists must have the same length.

    Returns:
        readouts (list[Readout]): The `Readout` events for the given timesteps.
    """
    timestep_count = len(input)
    if not timestep_count:
        return []

    start_timestep = get_current_timestep() + 1
    final_timestep = start_timestep + timestep_count - 1
    print("** To be processed timesteps: ", [timestep for timestep in range(start_timestep, final_timestep + 1)])

    input_events_list = []
    for i, spike_counts in enumerate(input):
        timestep = start_timestep + i
        for neuron_id, count in enumerate(spike_counts):
            spikes = [samna.xyloAudio3.event.Spike(neuron_id=neuron_id, timestep=timestep)] * count
            input_events_list.extend(spikes)

    input_events_list.append(samna.xyloAudio3.event.TriggerProcessing(target_timestep = final_timestep + 1))
    source.write(input_events_list)

    events = readout_sink.get_n_events(timestep_count, timeout=3000)
    assert(len(events) == timestep_count)

    for idx, ev in enumerate(events):
            print(f"{idx}: {ev}")

    return events


# Now we are ready to send spikes to our network and see how the SNN core reacts.
# Start the filter graph, otherwise the events aren't propagated.
graph.start()

# Send spikes to trigger output neuron 0
readouts = evolve([
    [2, 0, 0],
    [1, 0, 0],
    [0, 0, 0],
    ])
assert(readouts[-1].output_spikes == [1, 0, 0])

# Send spikes to trigger output neuron 1
readouts = evolve([
    [0, 2, 0],
    [0, 1, 0],
    [0, 0, 0],
    ])
assert(readouts[-1].output_spikes == [0, 1, 0])

# Send spikes to trigger output neuron 2
readouts = evolve([
    [0, 0, 2],
    [0, 0, 1],
    [0, 0, 0],
    ])
assert(readouts[-1].output_spikes == [0, 0, 1])

assert(readouts[-1].timestep == 8) # Last processed timestep is 8

# We have finished processing, so we can stop the graph.
graph.stop()