Source code for spynnaker.pyNN.connections.spynnaker_poisson_control_connection

# Copyright (c) 2017 The University of Manchester
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# See the License for the specific language governing permissions and
# limitations under the License.

from spinn_utilities.overrides import overrides
from spinn_front_end_common.interface.ds 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

class SpynnakerPoissonControlConnection(LiveEventConnection):
    A connection used to control a Poisson-distributed random event source's
    firing rate at runtime.
    __slots__ = [

    def __init__(
            self, poisson_labels=None, local_host=None, local_port=NOTIFY_PORT,
        :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 = [
                for label in poisson_labels
                {control: label
                 for control, label in zip(control_labels, poisson_labels)})
                {label: control
                 for label, control in zip(poisson_labels, control_labels)})

            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 f"{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().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().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().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().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)