Source code for bmtk.simulator.pointnet.sonata_adaptors

import numpy as np
from collections import Counter
import numbers
import nest
import types
import pandas as pd

from bmtk.simulator.core.sonata_reader import NodeAdaptor, SonataBaseNode, EdgeAdaptor, SonataBaseEdge
from bmtk.simulator.pointnet.io_tools import io
from bmtk.simulator.pointnet.pyfunction_cache import py_modules
from bmtk.simulator.pointnet.glif_utils import convert_aibs2nest
from bmtk.simulator.pointnet.nest_utils import nest_version

NEST_SYNAPSE_MODEL_PROP = 'model' if nest_version[0] == 2 else 'synapse_model'

[docs]def all_null(node_group, column_name): """Helper function to determine if a column has any non-NULL values""" types_table = node_group.parent.types_table non_null_vals = [types_table[ntid][column_name] for ntid in np.unique(node_group.node_type_ids) if types_table[ntid][column_name] is not None] return len(non_null_vals) == 0
[docs]class PointNodeBatched(object): def __init__(self, node_ids, gids, node_types_table, node_type_id): self._n_nodes = len(node_ids) self._node_ids = node_ids self._gids = gids self._nt_table = node_types_table self._nt_id = node_type_id self._nest_objs = [] self._nest_ids = [] @property def n_nodes(self): return self._n_nodes @property def node_ids(self): return self._node_ids @property def gids(self): return self._gids @property def nest_ids(self): return self._nest_ids @property def nest_model(self): return self._nt_table[self._nt_id]['model_template'].split(':')[1] @property def nest_params(self): return self._nt_table[self._nt_id]['dynamics_params'] @property def model_type(self): return self._nt_table[self._nt_id]['model_type']
[docs] def build(self): self._nest_objs = nest.Create(self.nest_model, self.n_nodes, self.nest_params) self._nest_ids = self._nest_objs.tolist() if nest_version[0] >= 3 else self._nest_objs
[docs]class PointNode(SonataBaseNode): def __init__(self, node, prop_adaptor): super(PointNode, self).__init__(node, prop_adaptor) self._nest_objs = [] self._nest_ids = [] @property def n_nodes(self): return 1 @property def node_ids(self): return [self._prop_adaptor.node_id(self._node)] @property def gids(self): return [self._prop_adaptor.gid(self._node)] @property def nest_ids(self): return self._nest_ids @property def nest_model(self): return self._prop_adaptor.model_template(self._node)[1] @property def nest_params(self): return self.dynamics_params
[docs] def build(self): nest_model = self.nest_model dynamics_params = self.dynamics_params fnc_name = self._node['model_processing'] if fnc_name is None: self._nest_objs = nest.Create(nest_model, 1, dynamics_params) else: cell_fnc = py_modules.cell_processor(fnc_name) self._nest_objs = cell_fnc(nest_model, self._node, dynamics_params) self._nest_ids = self._nest_objs.tolist() if nest_version[0] >= 3 else self._nest_objs
[docs]class PointNodeAdaptor(NodeAdaptor): def __init__(self, network): super(PointNodeAdaptor, self).__init__(network) # Flag for determining if we can build multiple NEST nodes at once. If each individual node has unique # NEST params or a model_processing function is being called then we must nest.Create for each individual cell. # Otherwise we can try to call nest.Create for a batch of nodes that share the same properties self._can_batch = True @property def batch_process(self): return self._can_batch @batch_process.setter def batch_process(self, flag): self._can_batch = flag
[docs] def get_node(self, sonata_node): return PointNode(sonata_node, self)
[docs] def get_batches(self, node_group): node_ids = node_group.node_ids node_type_ids = node_group.node_type_ids node_gids = node_group.gids if node_gids is None: node_gids = node_ids ntids_counter = Counter(node_type_ids) nid_groups = {nt_id: np.zeros(ntids_counter[nt_id], dtype=np.uint32) for nt_id in ntids_counter} gid_groups = {nt_id: np.zeros(ntids_counter[nt_id], dtype=np.uint32) for nt_id in ntids_counter} node_groups_counter = {nt_id: 0 for nt_id in ntids_counter} for node_id, gid, node_type_id in zip(node_ids, node_gids, node_type_ids): grp_indx = node_groups_counter[node_type_id] nid_groups[node_type_id][grp_indx] = node_id gid_groups[node_type_id][grp_indx] = gid node_groups_counter[node_type_id] += 1 return [PointNodeBatched(nid_groups[nt_id], gid_groups[nt_id], node_group.parent.node_types_table, nt_id) for nt_id in ntids_counter]
[docs] @staticmethod def preprocess_node_types(network, node_population): NodeAdaptor.preprocess_node_types(network, node_population) node_types_table = node_population.types_table if 'model_template' in node_types_table.columns and 'dynamics_params' in node_types_table.columns: node_type_ids = np.unique(node_population.type_ids) for nt_id in node_type_ids: node_type_attrs = node_types_table[nt_id] mtemplate = node_type_attrs['model_template'] dyn_params = node_type_attrs['dynamics_params'] if mtemplate.startswith('nest:glif') and dyn_params.get('type', None) == 'GLIF': model_template, dynamics_params = convert_aibs2nest(mtemplate, dyn_params) node_type_attrs['model_template'] = model_template node_type_attrs['dynamics_params'] = dynamics_params
[docs] @staticmethod def patch_adaptor(adaptor, node_group, network): node_adaptor = NodeAdaptor.patch_adaptor(adaptor, node_group, network) # If dynamics params is stored in the nodes.h5 then we have to build each node separate if node_group.has_dynamics_params: node_adaptor.batch_process = False # If there is a non-null value in the model_processing column then it potentially means that every cell is # uniquly built (currently model_processing is applied to each individ. cell) and nodes can't be batched if 'model_processing' in node_group.columns: node_adaptor.batch_process = False elif 'model_processing' in node_group.all_columns and not all_null(node_group, 'model_processing'): node_adaptor.batch_process = False if node_adaptor.batch_process: io.log_info('Batch processing nodes for {}/{}.'.format(, node_group.group_id)) return node_adaptor
[docs]class PointEdge(SonataBaseEdge): @property def source_node_ids(self): return [self._edge.source_node_id] @property def target_node_ids(self): return [self._edge.target_node_id] @property def nest_params(self): if self.model_template in py_modules.synapse_models: src_node = self._prop_adaptor._network.get_node_id(self.source_population, self.source_node_id) trg_node = self._prop_adaptor._network.get_node_id(self.target_population, self.target_node_id) syn_model_fnc = py_modules.synapse_model(self.model_template) else: src_node = None trg_node = None syn_model_fnc = py_modules.synapse_models('default') return syn_model_fnc(self, src_node, trg_node)
[docs]class PointEdgeBatched(object): def __init__(self, source_nids, target_nids, nest_params): self._src_nids = source_nids self._trg_nids = target_nids self._nest_params = nest_params @property def source_node_ids(self): return self._src_nids @property def target_node_ids(self): return self._trg_nids @property def nest_params(self): return self._nest_params
[docs]class PointEdgeAdaptor(EdgeAdaptor): def __init__(self, network): super(PointEdgeAdaptor, self).__init__(network) self._can_batch = True @property def batch_process(self): return self._can_batch @batch_process.setter def batch_process(self, flag): self._can_batch = flag
[docs] def synaptic_params(self, edge): # TODO: THIS NEEDS to be replaced with call to synapse_models params_dict = {'weight': self.syn_weight(edge, None, None), 'delay': edge.delay} params_dict.update(edge.dynamics_params) return params_dict
[docs] def get_edge(self, sonata_node): return PointEdge(sonata_node, self)
[docs] @staticmethod def preprocess_edge_types(network, edge_population): # Fix for sonata/300_pointneurons EdgeAdaptor.preprocess_edge_types(network, edge_population) edge_types_table = edge_population.types_table edge_type_ids = np.unique(edge_population.type_ids) for et_id in edge_type_ids: edge_type = edge_types_table[et_id] if 'model_template' in edge_types_table.columns: model_template = edge_type['model_template'] if model_template.startswith('nest'): edge_type['model_template'] = model_template[5:]
[docs] def get_batches(self, edge_group): src_ids = {} trg_ids = {} edge_types_table = edge_group.parent.edge_types_table edge_type_ids = edge_group.edge_type_ids et_id_counter = Counter(edge_type_ids) tmp_df = pd.DataFrame({'etid': edge_type_ids, 'src_nids': edge_group.src_node_ids, 'trg_nids': edge_group.trg_node_ids}) if 'nsyns' in edge_group.columns: tmp_df['nsyns'] = edge_group.get_dataset('nsyns') if 'syn_weight' in edge_group.columns: tmp_df['syn_weight'] = edge_group.get_dataset('syn_weight') if 'delay' in edge_group.columns: tmp_df['delay'] = edge_group.get_dataset('delay') #for et_id, grp_vals in tmp_df.groupby('etid'): # src_ids[et_id] = np.array(grp_vals['src_nids']) # trg_ids[et_id] = np.array(grp_vals['trg_nids']) type_params = {edge_id: {} for edge_id in et_id_counter.keys()} src_pop = edge_group.parent.source_population trg_pop = edge_group.parent.target_population grp_df = None src_nodes_df = None trg_nodes_df = None for edge_id, grp_vals in tmp_df.groupby('etid'): edge_props = edge_types_table[edge_id] n_edges = len(grp_vals) # Get the model type type_params[edge_id][NEST_SYNAPSE_MODEL_PROP] = edge_props['model_template'] # Add dynamics params # TODO: Add to dataframe and if a part of hdf5 we can return any dynamics params as a list type_params[edge_id].update(edge_props['dynamics_params']) # get the delay parameter if 'delay' in grp_vals.columns: type_params[edge_id]['delay'] = grp_vals['delay'] elif 'delay' in edge_props.keys(): delay = edge_props['delay'] # For NEST 2.* 'delay' can be a single value, but for 3.* it requires a full array for each edge delay = np.full(n_edges, delay) if nest_version[0] >= 3 else delay type_params[edge_id]['delay'] = delay weight_function = edge_types_table[edge_id].get('weight_function', None) if weight_function is not None: if grp_df is None: grp_df = edge_group.to_dataframe() src_nodes_df = self._network.get_nodes_df(src_pop) trg_nodes_df = self._network.get_nodes_df(trg_pop) edges = grp_df[grp_df['edge_type_id'] == edge_id] target_nodes = trg_nodes_df.loc[edges['target_node_id'].values] source_nodes = src_nodes_df.loc[edges['source_node_id'].values] if not py_modules.has_synaptic_weight(weight_function): err_msg = 'Unable to calculate synaptic weight for "{}" edges, missing "weight_function" ' \ 'attribute value {} function.'.format(, weight_function) io.log_exception(err_msg) weight_fnc = py_modules.synaptic_weight(weight_function) type_params[edge_id]['weight'] = weight_fnc(edges, source_nodes, target_nodes) else: # Get nsyns as either an array or a constant. If not explcitiy specified assume nsyns = 1 if 'nsyns' in grp_vals.columns: nsyns = grp_vals['nsyns'].values else: nsyns = edge_props.get('nsyns', 1) # get syn_weight as either an array or constant. If not explicity stated throw an error if 'syn_weight' in grp_vals.columns: syn_weight = grp_vals['syn_weight'].values elif 'syn_weight' in edge_props.keys(): syn_weight = edge_props['syn_weight'] else: # TODO: Make more explicity. Or default to syn_weight of 0 raise Exception('Could not find syn_weight value') # caluclate weight type_params[edge_id]['weight'] = nsyns * syn_weight yield PointEdgeBatched(source_nids=grp_vals['src_nids'].values, target_nids=grp_vals['trg_nids'].values, nest_params=type_params[edge_id])
[docs] @staticmethod def patch_adaptor(adaptor, edge_group): edge_adaptor = EdgeAdaptor.patch_adaptor(adaptor, edge_group) if 'weight_function' not in edge_group.all_columns and 'syn_weight' in edge_group.all_columns: adaptor.syn_weight = types.MethodType(point_syn_weight, adaptor) #else: # edge_adaptor.batch_process = False return edge_adaptor
[docs]def point_syn_weight(self, edge, src_node, trg_node): return edge['syn_weight']*edge.nsyns