cleo.ioproc module#

Classes and functions for constructing and configuring an IOProcessor.

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

Bases: 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

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: 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: NDArray[Any, UInt], **kwargs) → NDArray[Any, 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]

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

Bases: 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 = 1, sampling: str = 'fixed', processing: str = 'parallel')[source]#

Bases: IOProcessor

IOProcessor capable of delivering stimulation some time after measurement.

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.

Method generated by attrs for class LatencyIOProcessor.

get_ctrl_signals(query_time_ms)[source]#

Get per-stimulator control signal from the IOProcessor.

Parameters:

query_time_ms (float) – Current simulation time.

Returns:

A {‘stimulator_name’: ctrl_signal} 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

out_buffer: deque[Tuple[dict, float]]#
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]

processing: str#

“serial” or “parallel”.

“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.

Type:

Processing scheme

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.

sampling: str#

“fixed” or “when idle”.

“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.

Type:

Sampling scheme

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: 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

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: 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: LatencyIOProcessor

Take samples without performing any control.

Use this if all you are doing is recording.

Method generated by attrs for class LatencyIOProcessor.

out_buffer: deque[Tuple[dict, float]]#
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]

processing: str#

“serial” or “parallel”.

“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.

Type:

Processing scheme

sampling: str#

“fixed” or “when idle”.

“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.

Type:

Sampling scheme

t_samp_ms: list[float]#

Record of sampling times—each time put_state() is called.