lava.proc.io

lava.proc.io.dataloader

class lava.proc.io.dataloader.AbstractDataloader(gt_shape, dataset, interval=1, offset=0)

Bases: AbstractProcess

Abstract dataloader object.

Parameters

dataset (Iterable) – The actual dataset object. Dataset is expected to return (input, label/ground_truth) when indexed.
interval (int, optional) – Interval between each data load, by default 1
offset (int, optional) – Offset (phase) for each data load, by default 0

class lava.proc.io.dataloader.AbstractPyDataloaderModel(proc_params)

Bases: PyLoihiProcessModel

ground_truth: PyOutPort = LavaPyType(cls=<class 'lava.magma.core.model.py.ports.PyOutPortVectorDense'>, d_type=<class 'float'>, precision=None)

ground_truth_array()

Return type: ndarray

interval: ndarray = LavaPyType(cls=<class 'numpy.ndarray'>, d_type=<class 'int'>, precision=None)

offset: ndarray = LavaPyType(cls=<class 'numpy.ndarray'>, d_type=<class 'int'>, precision=None)

required_resources: ty.List[ty.Type[AbstractResource]] = [<class 'lava.magma.core.resources.HostCPU'>]

class lava.proc.io.dataloader.AbstractPySpikeDataloaderModel(proc_params)

Bases: AbstractPyDataloaderModel

data: Optional[ndarray] = None

post_guard()

Guard function that determines if post lrn mgmt phase will get executed or not for the current timestep.

Return type: None

run_post_mgmt()

Function that runs in Post Lrn Mgmt Phase

Return type: None

run_spk()

Function that runs in Spiking Phase

Return type: None

s_out: Optional[PyOutPort] = None

class lava.proc.io.dataloader.AbstractPyStateDataloaderModel(proc_params)

Bases: AbstractPyDataloaderModel

post_guard()

Guard function that determines if post lrn mgmt phase will get executed or not for the current timestep.

Return type: None

run_post_mgmt()

Function that runs in Post Lrn Mgmt Phase

Return type: None

run_spk()

Function that runs in Spiking Phase

Return type: None

state: Optional[PyRefPort] = None

class lava.proc.io.dataloader.PySpikeModelFixed(proc_params)

Bases: AbstractPySpikeDataloaderModel

data: ndarray = LavaPyType(cls=<class 'numpy.ndarray'>, d_type=<class 'numpy.int32'>, precision=None)

implements_process: alias of SpikeDataloader

implements_protocol: alias of LoihiProtocol

s_out: PyOutPort = LavaPyType(cls=<class 'lava.magma.core.model.py.ports.PyOutPortVectorDense'>, d_type=<class 'numpy.int32'>, precision=None)

tags: ty.List[str] = ['fixed_pt']

class lava.proc.io.dataloader.PySpikeModelFloat(proc_params)

Bases: AbstractPySpikeDataloaderModel

data: ndarray = LavaPyType(cls=<class 'numpy.ndarray'>, d_type=<class 'float'>, precision=None)

implements_process: alias of SpikeDataloader

implements_protocol: alias of LoihiProtocol

s_out: PyOutPort = LavaPyType(cls=<class 'lava.magma.core.model.py.ports.PyOutPortVectorDense'>, d_type=<class 'float'>, precision=None)

tags: ty.List[str] = ['floating_pt']

class lava.proc.io.dataloader.PyStateModelFixed(proc_params)

Bases: AbstractPyStateDataloaderModel

implements_process: alias of StateDataloader

implements_protocol: alias of LoihiProtocol

state: PyRefPort = LavaPyType(cls=<class 'lava.magma.core.model.py.ports.PyRefPortVectorDense'>, d_type=<class 'numpy.int32'>, precision=None)

tags: ty.List[str] = ['fixed_pt']

class lava.proc.io.dataloader.PyStateModelFloat(proc_params)

Bases: AbstractPyStateDataloaderModel

implements_process: alias of StateDataloader

implements_protocol: alias of LoihiProtocol

state: PyRefPort = LavaPyType(cls=<class 'lava.magma.core.model.py.ports.PyRefPortVectorDense'>, d_type=<class 'float'>, precision=None)

tags: ty.List[str] = ['floating_pt']

class lava.proc.io.dataloader.SpikeDataloader(*, dataset, interval=1, offset=0)

Bases: AbstractDataloader

Dataloader object that sends spike for a input sample at a set interval and offset (phase).

Parameters

dataset (Iterable) – The actual dataset object. Dataset is expected to return (spike, label/ground_truth) when indexed.
interval (int, optional) – Interval between each data load, by default 1
offset (int, optional) – Offset (phase) for each data load, by default 0

class lava.proc.io.dataloader.StateDataloader(*, dataset, interval=1, offset=0)

Bases: AbstractDataloader

Dataloader object that loads new data sample to internal state at a set interval and offset (phase).

Parameters

dataset (Iterable) – The actual dataset object. Dataset is expected to return (input, label/ground_truth) when indexed.
interval (int, optional) – Interval between each data load, by default 1
offset (int, optional) – Offset (phase) for each data load, by default 0

connect_var(var)

Connects the internal state (ref-port) to the variable. The variable can be internal state of some other process which needs to reflect the dataset input.

Parameters: var (Var) – Variable that is connected to this object’s state (ref-port).
Return type: None

lava.proc.io.encoder

class lava.proc.io.encoder.AbstractPyDeltaEncoderModel(proc_params=None)

Bases: PyLoihiProcessModel

Implementation of Delta encoder.

a_in = None

act: ndarray = LavaPyType(cls=<class 'numpy.ndarray'>, d_type=<class 'numpy.int32'>, precision=24)

encode_delta(act_new)

required_resources: ty.List[ty.Type[AbstractResource]] = [<class 'lava.magma.core.resources.HostCPU'>]

residue: ndarray = LavaPyType(cls=<class 'numpy.ndarray'>, d_type=<class 'numpy.int32'>, precision=24)

s_out = None

spike_exp: ndarray = LavaPyType(cls=<class 'numpy.ndarray'>, d_type=<class 'numpy.int32'>, precision=3)

vth: ndarray = LavaPyType(cls=<class 'numpy.ndarray'>, d_type=<class 'numpy.int32'>, precision=24)

class lava.proc.io.encoder.Compression(value)

Bases: Enum

Enumeration of message compression mode.

DENSE: No compression. Raw 32 bit data is communicated as it is.

SPARSE: Sparse 32 bit data and index is communicated.

PACKED_4: Four 8 bit data packed into 32 bit message. NOTE: only works for 8 bit data.

DELTA_SPARSE_8: 8 bit data and 8 bit delta encoded index. NOTE: only works for 8 bit data.

DELTA_SPARSE_8 = 3

DENSE = 0

PACKED_4 = 2

SPARSE = 1

class lava.proc.io.encoder.DeltaEncoder(*, shape, vth, spike_exp=0, compression=Compression.DENSE)

Bases: AbstractProcess

Delta encoding with threshold.

Delta encoding looks at the difference of new input and sends only the difference (delta) when it is more than a positive threshold.

Delta dynamics: delta = act_new - act + residue # delta encoding s_out = delta if abs(delta) > vth else 0 # spike mechanism residue = delta - s_out # residue accumulation act = act_new

Parameters

shape (Tuple) – Shape of the sigma process.
vth (int or float) – Threshold of the delta encoder.
spike_exp (int) – Scaling exponent with base 2 for the spike message. Note: This should only be used for fixed point models. Default is 0.
compression (Compression) – Data compression mode, by default DENSE compression.

property shape: Tuple[int, ...]

Return type: Tuple[int, ...]

class lava.proc.io.encoder.PyDeltaEncoderModelDense(proc_params=None)

Bases: AbstractPyDeltaEncoderModel

Dense (No) compression Model of PyDeltaEncoder.

a_in = LavaPyType(cls=<class 'lava.magma.core.model.py.ports.PyInPortVectorDense'>, d_type=<class 'numpy.int32'>, precision=24)

implements_process: alias of DeltaEncoder

implements_protocol: alias of LoihiProtocol

model_id: ty.Optional[int]

proc_params: ty.Type['ProcessParameters']

process_to_service: ty.Optional[CspSendPort]

py_ports: ty.List[AbstractPyPort]

run_spk(): Function that runs in Spiking Phase

s_out = LavaPyType(cls=<class 'lava.magma.core.model.py.ports.PyOutPortVectorDense'>, d_type=<class 'numpy.int32'>, precision=24)

service_to_process: ty.Optional[CspRecvPort]

tags: ty.List[str] = ['dense_out']

var_id_to_var_map: ty.Dict[int, ty.Any]

var_ports: ty.List[PyVarPort]

class lava.proc.io.encoder.PyDeltaEncoderModelSparse(proc_params=None)

Bases: AbstractPyDeltaEncoderModel

Sparse compression Model of PyDeltaEncoder.

Based on compression mode, it can be * SPARSE: 32 bit data and 32 bit index used for messaging sparse data. * PACKED_4: Four 8 bit data packed into one 32 bit data for messaging. * DELTA_SPARSE_8: 8 bit index and 8 bit data messaging.

a_in = LavaPyType(cls=<class 'lava.magma.core.model.py.ports.PyInPortVectorDense'>, d_type=<class 'numpy.int32'>, precision=24)

encode_delta_sparse_8(s_out): 8 bit compressed data and index encoding.

encode_packed_4(s_out): 4x 8bit data encodig into one 32 bit data.

encode_sparse(s_out): Basic sparse encoding.

implements_process: alias of DeltaEncoder

implements_protocol: alias of LoihiProtocol

model_id: ty.Optional[int]

proc_params: ty.Type['ProcessParameters']

process_to_service: ty.Optional[CspSendPort]

py_ports: ty.List[AbstractPyPort]

run_spk(): Function that runs in Spiking Phase

s_out = LavaPyType(cls=<class 'lava.magma.core.model.py.ports.PyOutPortVectorSparse'>, d_type=<class 'numpy.int32'>, precision=24)

service_to_process: ty.Optional[CspRecvPort]

tags: ty.List[str] = ['sparse_out']

var_id_to_var_map: ty.Dict[int, ty.Any]

var_ports: ty.List[PyVarPort]

lava.proc.io.reset

class lava.proc.io.reset.AbstractPyReset(proc_params=None)

Bases: PyLoihiProcessModel

Abstract Reset process implementation.

interval: ndarray = LavaPyType(cls=<class 'numpy.ndarray'>, d_type=<class 'int'>, precision=None)

offset: ndarray = LavaPyType(cls=<class 'numpy.ndarray'>, d_type=<class 'int'>, precision=None)

post_guard()

Guard function that determines if post lrn mgmt phase will get executed or not for the current timestep.

Return type: None

reset_value: ndarray = LavaPyType(cls=<class 'numpy.ndarray'>, d_type=<class 'int'>, precision=None)

run_post_mgmt()

Function that runs in Post Lrn Mgmt Phase

Return type: None

state: Optional[PyRefPort] = None

class lava.proc.io.reset.PyResetFixed(proc_params=None)

Bases: AbstractPyReset

Reset process implementation for int type.

implements_process: alias of Reset

implements_protocol: alias of LoihiProtocol

required_resources: ty.List[ty.Type[AbstractResource]] = [<class 'lava.magma.core.resources.CPU'>]

state: PyRefPort = LavaPyType(cls=<class 'lava.magma.core.model.py.ports.PyRefPortVectorDense'>, d_type=<class 'int'>, precision=None)

tags: ty.List[str] = ['fixed_pt']

class lava.proc.io.reset.PyResetFloat(proc_params=None)

Bases: AbstractPyReset

Reset process implementation for float type.

implements_process: alias of Reset

implements_protocol: alias of LoihiProtocol

required_resources: ty.List[ty.Type[AbstractResource]] = [<class 'lava.magma.core.resources.CPU'>]

state: PyRefPort = LavaPyType(cls=<class 'lava.magma.core.model.py.ports.PyRefPortVectorDense'>, d_type=<class 'float'>, precision=None)

tags: ty.List[str] = ['floating_pt']

class lava.proc.io.reset.Reset(*, reset_value=0, interval=1, offset=0)

Bases: AbstractProcess

Resets it’s internal state at a set interval and offset (phase).

Parameters

reset_value (int or float) – reset value, by default 0
interval (int, optional) – reset interval, by default 1
offset (int, optional) – reset offset (phase), by default 0

connect_var(var)

Return type: None

lava.proc.io.sink

class lava.proc.io.sink.AbstractPyRead(proc_params)

Bases: PyLoihiProcessModel

Abstract Read Var process implementation.

data = None

interval: ndarray = LavaPyType(cls=<class 'numpy.ndarray'>, d_type=<class 'int'>, precision=None)

offset: ndarray = LavaPyType(cls=<class 'numpy.ndarray'>, d_type=<class 'int'>, precision=None)

post_guard()

Guard function that determines if post lrn mgmt phase will get executed or not for the current timestep.

Return type: None

run_post_mgmt()

Function that runs in Post Lrn Mgmt Phase

Return type: None

state: Optional[PyRefPort] = None

class lava.proc.io.sink.AbstractPyReceiveModel(proc_params=None)

Bases: PyLoihiProcessModel

Abstract ring buffer receive process model.

a_in = None

data = None

model_id: ty.Optional[int]

proc_params: ty.Type['ProcessParameters']

process_to_service: ty.Optional[CspSendPort]

py_ports: ty.List[AbstractPyPort]

run_spk()

Receive spikes and store in an internal variable

Return type: None

service_to_process: ty.Optional[CspRecvPort]

var_id_to_var_map: ty.Dict[int, ty.Any]

var_ports: ty.List[PyVarPort]

class lava.proc.io.sink.PyReadFixed(proc_params)

Bases: AbstractPyRead

Read Var process implementation for int type.

data: ndarray = LavaPyType(cls=<class 'numpy.ndarray'>, d_type=<class 'numpy.int32'>, precision=None)

implements_process: alias of Read

implements_protocol: alias of LoihiProtocol

required_resources: ty.List[ty.Type[AbstractResource]] = [<class 'lava.magma.core.resources.CPU'>]

state: PyRefPort = LavaPyType(cls=<class 'lava.magma.core.model.py.ports.PyRefPortVectorDense'>, d_type=<class 'numpy.int32'>, precision=None)

tags: ty.List[str] = ['fixed_pt']

class lava.proc.io.sink.PyReadFloat(proc_params)

Bases: AbstractPyRead

Read Var process implementation for float type.

data: ndarray = LavaPyType(cls=<class 'numpy.ndarray'>, d_type=<class 'float'>, precision=None)

implements_process: alias of Read

implements_protocol: alias of LoihiProtocol

required_resources: ty.List[ty.Type[AbstractResource]] = [<class 'lava.magma.core.resources.CPU'>]

state: PyRefPort = LavaPyType(cls=<class 'lava.magma.core.model.py.ports.PyRefPortVectorDense'>, d_type=<class 'float'>, precision=None)

tags: ty.List[str] = ['floating_pt']

class lava.proc.io.sink.PyReceiveModelFixed(proc_params=None)

Bases: AbstractPyReceiveModel

Fixed point ring buffer receive process model.

a_in: PyInPort = LavaPyType(cls=<class 'lava.magma.core.model.py.ports.PyInPortVectorDense'>, d_type=<class 'numpy.int32'>, precision=None)

data: ndarray = LavaPyType(cls=<class 'numpy.ndarray'>, d_type=<class 'numpy.int32'>, precision=None)

implements_process: alias of RingBuffer

implements_protocol: alias of LoihiProtocol

required_resources: ty.List[ty.Type[AbstractResource]] = [<class 'lava.magma.core.resources.CPU'>]

tags: ty.List[str] = ['fixed_pt']

class lava.proc.io.sink.PyReceiveModelFloat(proc_params=None)

Bases: AbstractPyReceiveModel

Float ring buffer receive process model.

a_in: PyInPort = LavaPyType(cls=<class 'lava.magma.core.model.py.ports.PyInPortVectorDense'>, d_type=<class 'float'>, precision=None)

data: ndarray = LavaPyType(cls=<class 'numpy.ndarray'>, d_type=<class 'float'>, precision=None)

implements_process: alias of RingBuffer

implements_protocol: alias of LoihiProtocol

required_resources: ty.List[ty.Type[AbstractResource]] = [<class 'lava.magma.core.resources.CPU'>]

tags: ty.List[str] = ['floating_pt']

class lava.proc.io.sink.Read(*, buffer, interval=1, offset=0)

Bases: AbstractProcess

Reads and logs the data of it’s internal state at a set interval and offset (phase).

Parameters

buffer (int) – number of samples to buffer
interval (int, optional) – reset interval, by default 1
offset (int, optional) – reset offset (phase), by default 0

connect_var(var)

Return type: None

class lava.proc.io.sink.RingBuffer(*, shape, buffer)

Bases: AbstractProcess

Process for receiving arbitrarily shaped data into a ring buffer memory. Works as a substitute for probing.

Parameters

shape (tuple) – shape of the process
buffer (int) – size of data sink buffer

lava.proc.io.source

class lava.proc.io.source.AbstractPyRingBuffer(proc_params=None)

Bases: PyLoihiProcessModel

Abstract ring buffer process model.

data = None

model_id: ty.Optional[int]

proc_params: ty.Type['ProcessParameters']

process_to_service: ty.Optional[CspSendPort]

py_ports: ty.List[AbstractPyPort]

run_spk()

Function that runs in Spiking Phase

Return type: None

s_out = None

service_to_process: ty.Optional[CspRecvPort]

var_id_to_var_map: ty.Dict[int, ty.Any]

var_ports: ty.List[PyVarPort]

class lava.proc.io.source.PySendModelFixed(proc_params=None)

Bases: AbstractPyRingBuffer

Fixed point ring buffer send process model.

data: ndarray = LavaPyType(cls=<class 'numpy.ndarray'>, d_type=<class 'numpy.int32'>, precision=None)

implements_process: alias of RingBuffer

implements_protocol: alias of LoihiProtocol

required_resources: ty.List[ty.Type[AbstractResource]] = [<class 'lava.magma.core.resources.CPU'>]

s_out: PyOutPort = LavaPyType(cls=<class 'lava.magma.core.model.py.ports.PyOutPortVectorDense'>, d_type=<class 'numpy.int32'>, precision=None)

tags: ty.List[str] = ['fixed_pt']

class lava.proc.io.source.PySendModelFloat(proc_params=None)

Bases: AbstractPyRingBuffer

Float ring buffer send process model.

data: ndarray = LavaPyType(cls=<class 'numpy.ndarray'>, d_type=<class 'float'>, precision=None)

implements_process: alias of RingBuffer

implements_protocol: alias of LoihiProtocol

required_resources: ty.List[ty.Type[AbstractResource]] = [<class 'lava.magma.core.resources.CPU'>]

s_out: PyOutPort = LavaPyType(cls=<class 'lava.magma.core.model.py.ports.PyOutPortVectorDense'>, d_type=<class 'float'>, precision=None)

tags: ty.List[str] = ['floating_pt']

class lava.proc.io.source.RingBuffer(*, data)

Bases: AbstractProcess

Spike generator process from circular data buffer.

Parameters: data (np array) – data to generate spike from. Last dimension is assumed as time.