cleo.ioproc module#

Classes and functions for constructing and configuring an IOProcessor.

class cleo.ioproc.ConstantDelay(delay_ms: float)[source]#

Bases: cleo.ioproc.delays.Delay

Simply adds a constant delay to the computation

Parameters: delay_ms (float) – Desired delay in milliseconds

compute()[source]#: Compute delay.

class cleo.ioproc.Delay[source]#

Bases: abc.ABC

Abstract base class for computing delays.

abstract compute() → float[source]#: Compute delay.

class cleo.ioproc.FiringRateEstimator(tau_ms: float, sample_period_ms: float, **kwargs)[source]#

Bases: cleo.ioproc.base.ProcessingBlock

Exponential filter to estimate firing rate.

Requires sample_time_ms kwarg when process is called.

Parameters

tau_ms (float) – Time constant of filter
sample_period_ms (float) – Sampling period in milliseconds

compute_output(input: nptyping.types._ndarray.NDArray[Any, nptyping.types._number.UInt], **kwargs) → nptyping.types._ndarray.NDArray[Any, nptyping.types._number.Float][source]#

Estimate firing rate given past and current spikes.

Parameters: input (NDArray[(n,), np.uint]) – n-length vector of spike counts
Keyword Arguments: sample_time_ms (float) – Time measurement was taken in milliseconds
Returns: n-length vector of firing rates
Return type: NDArray[(n,), float]

delay: Delay#: The delay object determining compute latency for the block

save_history: bool#: Whether to record t_in_ms, t_out_ms, and values with every timestep

t_in_ms: list[float]#: The walltime the block received each input. Only recorded if save_history

t_out_ms: list[float]#: The walltime of each of the block’s outputs. Only recorded if save_history

values: list[Any]#: Each of the block’s outputs. Only recorded if save_history

class cleo.ioproc.GaussianDelay(loc: float, scale: float)[source]#

Bases: cleo.ioproc.delays.Delay

Generates normal-distributed delay.

Will return 0 when a negative value is sampled.

Parameters

loc (float) – Center of distribution
scale (float) – Standard deviation of delay distribution

compute() → float[source]#: Compute delay.

class cleo.ioproc.LatencyIOProcessor(sample_period_ms: float, **kwargs)[source]#

Bases: cleo.base.IOProcessor

IOProcessor capable of delivering stimulation some time after measurement.

Parameters

sample_period_ms (float) – Determines how frequently samples are taken from the network.

Keyword Arguments

sampling (str) –
“fixed” or “when idle”; “fixed” by default

”fixed” sampling means samples are taken on a fixed schedule, with no exceptions.

”when idle” sampling means no samples are taken before the previous sample’s output has been delivered. A sample is taken ASAP after an over-period computation: otherwise remains on schedule.
processing (str) –
“parallel” or “serial”; “parallel” by default

”parallel” computes the output time by adding the delay for a sample onto the sample time, so if the delay is 2 ms, for example, while the sample period is only 1 ms, some of the processing is happening in parallel. Output order matches input order even if the computed output time for a sample is sooner than that for a previous sample.

”serial” computes the output time by adding the delay for a sample onto the output time of the previous sample, rather than the sampling time. Note this may be of limited utility because it essentially means the entire round trip cannot be in parallel at all. More realistic is that simply each block or phase of computation must be serial. If anyone cares enough about this, it will have to be implemented in the future.

Note

It doesn’t make much sense to combine parallel computation with “when idle” sampling, because “when idle” sampling only produces one sample at a time to process.

Raises: ValueError – For invalid sampling or processing kwargs

get_ctrl_signal(query_time_ms)[source]#

Get per-stimulator control signal from the IOProcessor.

Parameters: query_time_ms (float) – Current simulation time.
Returns: A {‘stimulator_name’: value} dictionary for updating stimulators.
Return type: dict

is_sampling_now(query_time_ms)[source]#

Determines whether the processor will take a sample at this timestep.

Parameters: time (Brian 2 temporal Unit) – Current timestep.
Return type: bool

abstract process(state_dict: dict, sample_time_ms: float) → Tuple[dict, float][source]#

Process network state to generate output to update stimulators.

This is the function the user must implement to define the signal processing pipeline.

Parameters

state_dict (dict) – {recorder_name: state} dictionary from get_state()
time_ms (float) –

Returns

{‘stim_name’: ctrl_signal} dictionary and output time in milliseconds.

Return type

Tuple[dict, float]

put_state(state_dict: dict, sample_time_ms: float)[source]#

Deliver network state to the IOProcessor.

Parameters

state_dict (dict) – A dictionary of recorder measurements, as returned by get_state()
sample_time_ms (float) – The current simulation timestep. Essential for simulating control latency and for time-varying control.

t_samp_ms: list[float]#: Record of sampling times—each time put_state() is called.

class cleo.ioproc.PIController(ref_signal: callable, Kp: float, Ki: float = 0, sample_period_ms: float = 0, **kwargs: Any)[source]#

Bases: cleo.ioproc.base.ProcessingBlock

Simple PI controller.

compute_output() requires a sample_time_ms keyword argument. Only tested on controlling scalar values, but could be easily adapted to controlling a multi-dimensional state.

Parameters

ref_signal (callable) – Must return the target as a function of time in ms
Kp (float) – Gain on the proportional error
Ki (float, optional) – Gain on the integral error, by default 0
sample_period_ms (float, optional) – Rate at which processor takes samples, by default 0. Only used to compute integrated error on first sample

compute_output(input: float, **kwargs) → float[source]#

Compute control input to the system using previously specified gains.

Parameters: input (Any) – Current system state
Returns: Control signal
Return type: float

ref_signal: callable[[float], Any]#: Callable returning the target as a function of time in ms

class cleo.ioproc.ProcessingBlock(**kwargs)[source]#

Bases: abc.ABC

Abstract signal processing stage or control block.

It’s important to use super().__init__(**kwargs) in the base class to use the parent-class logic here.

Keyword Arguments: delay (Delay) – Delay object which adds to the compute time
Raises: TypeError – When delay is not a Delay object.

abstract compute_output(input: Any, **kwargs) → Any[source]#

Computes output for given input.

This is where the user will implement the desired functionality of the ProcessingBlock without regard for latency.

Parameters

input (Any) – Data to be processed. Passed in from process().
**kwargs (Any) – optional key-value argument pairs passed from process(). Could be used to pass in such values as the IO processor’s walltime or the measurement time for time- dependent functions.

Returns

output

Return type

Any

delay: cleo.ioproc.delays.Delay#: The delay object determining compute latency for the block

process(input: Any, t_in_ms: float, **kwargs) → Tuple[Any, float][source]#

Computes and saves output and output time given input and input time.

The user should implement compute_output() for their child classes, which performs the computation itself without regards for timing or saving variables.

Parameters

input (Any) – Data to be processed
t_in_ms (float) – Time the block receives the input data
**kwargs (Any) – Key-value list of arguments passed to compute_output()

Returns

output, out time in milliseconds

Return type

Tuple[Any, float]

save_history: bool#: Whether to record t_in_ms, t_out_ms, and values with every timestep

t_in_ms: list[float]#: The walltime the block received each input. Only recorded if save_history

t_out_ms: list[float]#: The walltime of each of the block’s outputs. Only recorded if save_history

values: list[Any]#: Each of the block’s outputs. Only recorded if save_history

class cleo.ioproc.RecordOnlyProcessor(sample_period_ms, **kwargs)[source]#

Bases: cleo.ioproc.base.LatencyIOProcessor

Take samples without performing any control.

Use this if all you are doing is recording.

Parameters

sample_period_ms (float) – Determines how frequently samples are taken from the network.

Keyword Arguments

sampling (str) –
“fixed” or “when idle”; “fixed” by default

”fixed” sampling means samples are taken on a fixed schedule, with no exceptions.

”when idle” sampling means no samples are taken before the previous sample’s output has been delivered. A sample is taken ASAP after an over-period computation: otherwise remains on schedule.
processing (str) –
“parallel” or “serial”; “parallel” by default

”parallel” computes the output time by adding the delay for a sample onto the sample time, so if the delay is 2 ms, for example, while the sample period is only 1 ms, some of the processing is happening in parallel. Output order matches input order even if the computed output time for a sample is sooner than that for a previous sample.

”serial” computes the output time by adding the delay for a sample onto the output time of the previous sample, rather than the sampling time. Note this may be of limited utility because it essentially means the entire round trip cannot be in parallel at all. More realistic is that simply each block or phase of computation must be serial. If anyone cares enough about this, it will have to be implemented in the future.

Note

It doesn’t make much sense to combine parallel computation with “when idle” sampling, because “when idle” sampling only produces one sample at a time to process.

Raises: ValueError – For invalid sampling or processing kwargs

process(state_dict: dict, sample_time_ms: float) → Tuple[dict, float][source]#

Process network state to generate output to update stimulators.

This is the function the user must implement to define the signal processing pipeline.

Parameters

state_dict (dict) – {recorder_name: state} dictionary from get_state()
time_ms (float) –

Returns

{‘stim_name’: ctrl_signal} dictionary and output time in milliseconds.

Return type

Tuple[dict, float]

sample_period_ms: float#: Determines how frequently the processor takes samples

t_samp_ms: list[float]#: Record of sampling times—each time put_state() is called.