# 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
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# 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 <http://www.gnu.org/licenses/>.
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)