Source code for spynnaker.pyNN.connections.spynnaker_poisson_control_connection

# Copyright (c) 2017-2019 The University of Manchester
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# GNU General Public License for more details.
# You should have received a copy of the GNU General Public License
# along with this program.  If not, see <>.

from spinn_utilities.overrides import overrides
from data_specification.enums import DataType
from spinn_front_end_common.utilities.connections import LiveEventConnection
from spinn_front_end_common.utilities.exceptions import ConfigurationException
from spinn_front_end_common.utilities.constants import NOTIFY_PORT
import functools

[docs]class SpynnakerPoissonControlConnection(LiveEventConnection): __slots__ = [ "__control_label_extension", "__control_label_to_label", "__label_to_control_label"] def __init__( self, poisson_labels=None, local_host=None, local_port=NOTIFY_PORT, control_label_extension="_control"): """ :param iterable(str) poisson_labels: Labels of Poisson populations to be controlled :param str local_host: Optional specification of the local hostname or\ IP address of the interface to listen on :param int local_port: Optional specification of the local port to listen on. Must match the port that the toolchain will send the notification on (19999 by default) :param str control_label_extension: The extra name added to the label of each Poisson source """ self.__control_label_extension = control_label_extension control_labels = None self.__control_label_to_label = dict() self.__label_to_control_label = dict() if poisson_labels is not None: control_labels = [ self.__convert_to_control_label(label) for label in poisson_labels ] self.__control_label_to_label.update( {control: label for control, label in zip(control_labels, poisson_labels)}) self.__label_to_control_label.update( {label: control for label, control in zip(poisson_labels, control_labels)}) super(SpynnakerPoissonControlConnection, self).__init__( live_packet_gather_label=None, send_labels=control_labels, local_host=local_host, local_port=local_port)
[docs] def add_poisson_label(self, label): """ :param str label: The label of the Poisson source population. """ control = self.__convert_to_control_label(label) self.__control_label_to_label[control] = label self.__label_to_control_label[label] = control self.add_send_label(control)
def __convert_to_control_label(self, label): return "{}{}".format(label, self.__control_label_extension) def __control_label(self, label): # Try to get a control label, but if not just use the label return self.__label_to_control_label.get(label, label) def __label(self, control_label): # Try to get a label, but if not just use the control label return self.__control_label_to_label.get(control_label, control_label) def _start_callback_wrapper(self, callback, label, connection): callback(self.__label(label), connection) def _init_callback_wrapper(self, callback, label, vertex_size, run_time_ms, machine_timestep_ms): callback(self.__label(label), vertex_size, run_time_ms, machine_timestep_ms) def _stop_callback_wrapper(self, callback, label, connection): callback(self.__label(label), connection)
[docs] @overrides(LiveEventConnection.add_start_callback) def add_start_callback(self, label, start_callback): super(SpynnakerPoissonControlConnection, self).add_start_callback( self.__control_label(label), functools.partial( self._start_callback_wrapper, start_callback))
[docs] @overrides(LiveEventConnection.add_start_resume_callback) def add_start_resume_callback(self, label, start_resume_callback): super(SpynnakerPoissonControlConnection, self)\ .add_start_resume_callback( self.__control_label(label), functools.partial( self._start_callback_wrapper, start_resume_callback))
[docs] @overrides(LiveEventConnection.add_init_callback) def add_init_callback(self, label, init_callback): super(SpynnakerPoissonControlConnection, self).add_init_callback( self.__control_label(label), functools.partial( self._init_callback_wrapper, init_callback))
[docs] @overrides(LiveEventConnection.add_receive_callback) def add_receive_callback(self, label, live_event_callback, translate_key=False): raise ConfigurationException( "SpynnakerPoissonControlPopulation can't receive data")
[docs] @overrides(LiveEventConnection.add_pause_stop_callback) def add_pause_stop_callback(self, label, pause_stop_callback): super(SpynnakerPoissonControlConnection, self).add_pause_stop_callback( self.__control_label(label), functools.partial( self._stop_callback_wrapper, pause_stop_callback))
[docs] def set_rate(self, label, neuron_id, rate): """ Set the rate of a Poisson neuron within a Poisson source :param str label: The label of the Population to set the rates of :param int neuron_id: The neuron ID to set the rate of :param float rate: The rate to set in Hz """ self.set_rates(label, [(neuron_id, rate)])
[docs] def set_rates(self, label, neuron_id_rates): """ Set the rates of multiple Poisson neurons within a Poisson source :param str label: The label of the Population to set the rates of :param list(tuple(int,float)) neuron_id_rates: A list of tuples of (neuron ID, rate) to be set """ control = self.__control_label(label) datatype = DataType.S1615 atom_ids_and_payloads = [(nid, datatype.encode_as_int(rate)) for nid, rate in neuron_id_rates] self.send_events_with_payloads(control, atom_ids_and_payloads)