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
- 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
- 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
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]
- 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]
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
- class lava.proc.io.reset.PyResetFixed(proc_params=None)
Bases:
AbstractPyReset
Reset process implementation for int type.
- 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_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
- 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]
- 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_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_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]
- 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.