Restores the Brian Network to where it was when the CLSimulator was last modified (last injection, IOProcessor change). Calls reset() on devices and io_processor.

run(duration: Quantity, **kwparams) → None[source]¶

Run simulation.

Parameters:

duration (brian2 temporal Quantity) – Length of simulation
**kwparams (additional arguments passed to brian2.run()) – level has a default value of 1

set_io_processor(io_processor: IOProcessor, communication_period=None) → CLSimulator[source]¶

Set simulator IO processor

Will replace any previous IOProcessor so there is only one at a time. A Brian NetworkOperation is created to govern communication between the Network and the IOProcessor.

Parameters:: io_processor (IOProcessor)
Returns:: self
Return type:: CLSimulator

stimulators: dict[str, Stimulator]¶

to_neo() → Block[source]¶

Exports simulator data to a Neo Block

Returns:: Neo Block containing signals representing each device’s data
Return type:: neo.core.Block

update_stimulators(stim_values: dict[str, Any]) → None[source]¶

Update stimulators with output from the IOProcessor

Parameters:: stim_values (dict) – {stimulator_name: stim_value} dictionary with values to update each stimulator.

class cleo.IOProcessor(sample_period: Quantity = 1. * msecond)[source]¶

Bases: ABC

Abstract class for implementing sampling, signal processing and control

This must be implemented by the user with their desired closed-loop use case, though most users will find the LatencyIOProcessor() class more useful, since delay handling is already defined.

Method generated by attrs for class IOProcessor.

abstract get_ctrl_signals(t_query: Quantity) → dict[source]¶

Get per-stimulator control signal from the IOProcessor.

Parameters:: t_query (Quantity) – Current simulation time.
Returns:: A {‘stimulator_name’: ctrl_signal} dictionary for updating stimulators.
Return type:: dict

get_stim_values(t_query: Quantity) → dict[source]¶

abstract is_sampling_now(t_now: Quantity) → bool[source]¶

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

Parameters:: t_now (Quantity) – Current time.
Return type:: bool

latest_ctrl_signal: dict¶: The most recent control signal returned by get_ctrl_signals()

preprocess_ctrl_signals(latest_ctrl_signals: dict, t_query: Quantity) → dict[source]¶

Preprocess control signals as needed to control stimulator waveforms between samples.

I.e., if a control signal defines the frequency of a periodic light stimulus, this function computes the current intensity given the latest frequency and the current time. This is called immediately after get_ctrl_signals() and on every timestep to update the stimulator waveform between samples.

This only needs to be implemented when a stimulus that varies between samples is desired. Otherwise, the control signal returned by get_ctrl_signals() is used directly. If not all stimulators need this functionality, only return a dict for those that do. The original, unprocessed control signal is used for the others.

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

abstract put_state(state_dict: dict, t_samp: Quantity) → None[source]¶

Deliver network state to the IOProcessor.

Parameters:

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

reset(**kwargs) → None[source]¶: Gets called on reset(). The user should implement this if their processor has a state that needs to be reset.

sample_period: Quantity¶: Determines how frequently the processor takes samples

class cleo.InterfaceDevice(*, name: str = NOTHING, save_history: bool = True)[source]¶

Bases: ABC

Base class for devices to be injected into the network

Method generated by attrs for class InterfaceDevice.

add_self_to_plot(ax: Axes3D, axis_scale_unit: Unit, **kwargs) → list[Artist][source]¶

Add device to an existing plot

Should only be called by plot().

Parameters:

ax (Axes3D) – The existing matplotlib Axes object
axis_scale_unit (Unit) – The unit used to label axes and define chart limits
**kwargs (optional)

Returns:

A list of artists used to render the device. Needed for use in conjunction with VideoVisualizer.

Return type:

list[Artist]

brian_objects: set¶: All the Brian objects added to the network by this device. Must be kept up-to-date in connect_to_neuron_group() and other functions so that those objects can be automatically added to the network when the device is injected.

abstract connect_to_neuron_group(neuron_group: NeuronGroup, **kwparams) → None[source]¶

Connect device to given neuron_group.

If your device introduces any objects which Brian must keep track of, such as a NeuronGroup, Synapses, or Monitor, make sure to add these to brian_objects.

Parameters:

neuron_group (NeuronGroup)
**kwparams (optional) – Passed from inject

init_for_simulator(simulator: CLSimulator) → None[source]¶

Initialize device for simulator on initial injection.

This function is called only the first time a device is injected into a simulator and performs any operations that are independent of the individual neuron groups it is connected to.

Parameters:: simulator (CLSimulator) – simulator being injected into

name: str¶: Identifier for device, used in sampling, plotting, etc. Name of the class by default. Must be unique among recorders and stimulators

reset(**kwargs) → None[source]¶: Reset the device to a neutral state

save_history: bool¶

Determines whether times and inputs/outputs are recorded. True by default.

For stimulators, this is when update() is called. For recorders, it is when get_state() is called.

sim: CLSimulator¶: The simulator the device is injected into

update_artists(artists: list[Artist], *args, **kwargs) → list[Artist][source]¶

Update the artists used to render the device

Used to set the artists’ state at every frame of a video visualization. The current state would be passed in *args or **kwargs

Parameters:: artists (list[Artist]) – the artists used to render the device originally, i.e., which were returned from the first add_self_to_plot() call.
Returns:: The artists that were actually updated. Needed for efficient blit rendering, where only updated artists are re-rendered.
Return type:: list[Artist]

class cleo.LatencyIOProcessor(sample_period: Quantity = 1. * msecond, 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(t_query)[source]¶

Get per-stimulator control signal from the IOProcessor.

Parameters:: t_query (Quantity) – Current simulation time.
Returns:: A {‘stimulator_name’: ctrl_signal} dictionary for updating stimulators.
Return type:: dict

is_sampling_now(t_query)[source]¶

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

Parameters:: t_now (Quantity) – Current time.
Return type:: bool

out_buffer: deque[Tuple[dict, float]]¶

“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

abstract process(state_dict: dict, t_samp: Quantity) → Tuple[dict, Quantity][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()
t_samp (Quantity) – The time at which the sample was taken.

Returns:

{‘stim_name’: ctrl_signal} dictionary and output time (including unit).

Return type:

Tuple[dict, Quantity]

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.

Type:: Processing scheme

put_state(state_dict: dict, t_samp: Quantity)[source]¶

Deliver network state to the IOProcessor.

Parameters:

state_dict (dict) – A dictionary of recorder measurements, as returned by get_state()
t_samp (Quantity) – 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: Quantity¶: Record of sampling times—each time put_state() is called.

class cleo.NeoExportable[source]¶

Bases: ABC

Mixin class for classes that can be exported to Neo objects

abstract to_neo() → neo.core.BaseNeo[source]¶

Return a Neo signal object with the device’s data

Returns:: Neo object representing exported data
Return type:: neo.core.BaseNeo

class cleo.Recorder(*, name: str = NOTHING, save_history: bool = True)[source]¶

Bases: InterfaceDevice

Device for taking measurements of the network.

Method generated by attrs for class Recorder.

abstract get_state() → Any[source]¶: Return current measurement.

class cleo.Stimulator(*, name: str = NOTHING, save_history: bool = True)[source]¶

Bases: InterfaceDevice, NeoExportable

Device for manipulating the network

Method generated by attrs for class Stimulator.

reset(**kwargs) → None[source]¶: Reset the stimulator device to a neutral state

t: Quantity¶: Times stimulator was updated, stored if save_history

to_neo()[source]¶

Return a Neo signal object with the device’s data

Returns:: Neo object representing exported data
Return type:: neo.core.BaseNeo

update(ctrl_signal) → None[source]¶

Set the stimulator value.

By default this sets value to ctrl_signal and updates saved times and values. You will want to implement this method if your stimulator requires additional logic. Use super.update(self, value) to preserve the self.value and save_history logic

Parameters:: ctrl_signal (any) – The value the stimulator is to take.

value: Any¶: The current value of the stimulator device

values: list[Any]¶: Values taken by the stimulator at each update() call, stored if save_history

class cleo.SynapseDevice(extra_namespace: dict = NOTHING, *, name: str = NOTHING, save_history: bool = True)[source]¶

Bases: InterfaceDevice

Base class for devices that record from/stimulate neurons via a Synapses object with device-specific model. Used for opsin and indicator classes

Method generated by attrs for class SynapseDevice.

connect_to_neuron_group(neuron_group: NeuronGroup, **kwparams) → None[source]¶

Transfect neuron group with device.

Parameters:

neuron_group (NeuronGroup) – The neuron group to transform

Keyword Arguments:

p_expression (float) – Probability (0 <= p <= 1) that a given neuron in the group will express the protein. 1 by default.
i_targets (array-like) – Indices of neurons in the group to transfect. recommended for efficiency when stimulating or imaging a small subset of the group. Incompatible with p_expression.
rho_rel (float) – The expression level, relative to the standard model fit, of the protein. 1 by default. For heterogeneous expression, this would have to be modified in the light-dependent synapse post-injection, e.g., opsin.syns["neuron_group_name"].rho_rel = ...
[default_name]_var_name (str) – See required_vars. Allows for custom variable names.

extra_namespace: dict¶: Additional items (beyond parameters) to be added to the opto synapse namespace

init_syn_vars(syn: Synapses) → None[source]¶

Initializes appropriate variables in Synapses implementing the model

Also called on reset().

Parameters:: syn (Synapses) – The synapses object implementing this model

model: str¶

Basic Brian model equations string.

Should contain a rho_rel term reflecting relative expression levels. Will likely also contain special NeuronGroup-dependent symbols such as V_VAR_NAME to be replaced on injection in modify_model_and_params_for_ng().

modify_model_and_params_for_ng(neuron_group: NeuronGroup, injct_params: dict) → Tuple[Equations, dict][source]¶

Adapt model for given neuron group on injection

This enables the specification of variable names differently for each neuron group, allowing for custom names and avoiding conflicts.

Parameters:

neuron_group (NeuronGroup) – NeuronGroup this opsin model is being connected to
injct_params (dict) – kwargs passed in on injection, could contain variable names to plug into the model

Keyword Arguments:

model (str, optional) – Model to start with, by default that defined for the class. This allows for prior string manipulations before it can be parsed as an Equations object.

Returns:

A tuple containing an Equations object and a parameter dictionary, constructed from model and params, respectively, with modified names for use in synapses

Return type:

Equations, dict

on_pre: str¶: Model string for brian2.synapses.synapses.Synapses reacting to spikes.

property params: dict¶: Returns a dictionary of parameters for the model

per_ng_unit_replacements: list[Tuple[str, str]]¶: List of (UNIT_NAME, neuron_group_specific_unit_name) tuples to be substituted in the model string on injection and before checking required variables.

required_vars: list[Tuple[str, Unit]]¶

Default names of state variables required in the neuron group, along with units, e.g., [(‘Iopto’, amp)].

It is assumed that non-default values can be passed in on injection as a keyword argument [default_name]_var_name=[non_default_name] and that these are found in the model string as [DEFAULT_NAME]_VAR_NAME before replacement.

reset(**kwargs)[source]¶: Reset the device to a neutral state

source_ngs: dict[str, NeuronGroup]¶

{target_ng.name: source_ng} dict of source neuron groups.

The source is the target itself by default or light aggregator neurons for LightDependent.

synapses: dict[str, Synapses]¶: Stores the synapse objects implementing the model, connecting from source (light aggregator neurons or the target group itself) to target neuron groups, of form {target_ng.name: synapses}.