Source code for spynnaker.pyNN.utilities.struct

# 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.

import numpy
from enum import Enum
from pyNN.random import RandomDistribution
from spinn_utilities.helpful_functions import is_singleton
from spinn_front_end_common.utilities.constants import BYTES_PER_WORD
from spynnaker.pyNN.utilities.utility_calls import convert_to
from spynnaker.pyNN.models.common.param_generator_data import (
    get_generator_type, param_generator_id, param_generator_params,


[docs] class StructRepeat(Enum): """ How a structure repeats, or not, in memory. """ #: Indicates a single global struct GLOBAL = 0 #: Indicates a struct that repeats per neuron PER_NEURON = 1
[docs] class Struct(object): """ Represents a C code structure. """ __slots__ = [ "__fields", "__repeat_type", "__default_values" ] def __init__(self, fields, repeat_type=StructRepeat.PER_NEURON, default_values=None): """ :param fields: The types and names of the fields, ordered as they appear in the structure. :type fields: list(~data_specification.enums.DataType, str) :param StructRepeat repeat_type: How the structure repeats :param default_values: Dict of field name -> value to use when values doesn't contain the field :type default_values: dict(str, int or float) or None """ self.__fields = fields self.__repeat_type = repeat_type self.__default_values = default_values or dict() @property def fields(self): """ The types and names of the fields, ordered as they appear in the structure. :rtype: list(~data_specification.enums.DataType, str) """ return self.__fields @property def repeat_type(self): """ How the structure repeats. :rtype: StructRepeat """ return self.__repeat_type @property def numpy_dtype(self): """ The numpy data type of the structure. :rtype: ~numpy.dtype """ return numpy.dtype( [(name, numpy.dtype(data_type.struct_encoding)) for data_type, name in self.__fields], align=True)
[docs] def get_size_in_whole_words(self, array_size=1): """ Get the size of the structure in whole words in an array of given size (default 1 item). :param int array_size: The number of elements in an array of structures :rtype: int """ datatype = self.numpy_dtype size_in_bytes = array_size * datatype.itemsize return (size_in_bytes + (BYTES_PER_WORD - 1)) // BYTES_PER_WORD
[docs] def get_data(self, values, vertex_slice=None): """ Get a numpy array of uint32 of data for the given values. :param values: The values to fill in the data with :type values: dict(str, int or float or AbstractList) :param vertex_slice: The vertex slice to get the data for, or `None` if the structure is global. :type vertex_slice: Slice or None :rtype: ~numpy.ndarray(dtype="uint32") """ n_items = 1 if vertex_slice is None: if self.__repeat_type != StructRepeat.GLOBAL: raise ValueError( "Repeating structures must specify a vertex_slice") elif self.__repeat_type == StructRepeat.GLOBAL: raise ValueError("Global Structures do not have a slice") else: n_items = vertex_slice.n_atoms # Create an array to store values in data = numpy.zeros(n_items, dtype=self.numpy_dtype) if not self.__fields: return data.view("uint32") # Go through and get the values and put them in the array for data_type, name in self.__fields: if name in values: all_vals = values[name] if is_singleton(all_vals): # If there is just one value for everything, use it # everywhere data[name] = convert_to(all_vals, data_type) elif self.__repeat_type == StructRepeat.GLOBAL: # If there is a ranged list for global struct, # we might need to read a single value data[name] = convert_to( all_vals.get_single_value_all(), data_type) else: self.__get_data_for_slice( data, all_vals, name, data_type, vertex_slice) else: # If there is only a default value, get that and use it # everywhere value = self.__default_values[name] data_value = convert_to(value, data_type) data[name] = data_value # Pad to whole number of uint32s overflow = (n_items * self.numpy_dtype.itemsize) % BYTES_PER_WORD if overflow != 0: data = numpy.pad( data.view("uint8"), (0, BYTES_PER_WORD - overflow), "constant") return data.view("uint32")
def __get_data_for_slice( self, data, all_vals, name, data_type, vertex_slice): """ Get the data for a single value from a vertex slice. """ # If there is a list of values, convert it ids = vertex_slice.get_raster_ids() data_pos = 0 for start, stop, value in all_vals.iter_ranges_by_ids(ids): # Get the values and convert to the correct data type n_values = stop - start if isinstance(value, RandomDistribution): r_vals = data_value = [ convert_to(v, data_type) for v in r_vals] else: data_value = convert_to(value, data_type) data[name][data_pos:data_pos + n_values] = data_value data_pos += n_values
[docs] def get_generator_data(self, values, vertex_slice=None): """ Get a numpy array of uint32 of data to generate the given values. :param ~dict-like values: The values to fill in the data with :param vertex_slice: The vertex slice or `None` for a structure with repeat_type global, or where a single value repeats for every neuron. If this is not the case and vertex_slice is `None`, an error will be raised! :type vertex_slice: Slice or None :rtype: ~numpy.ndarray(dtype="uint32") """ # Define n_repeats, which is either the total number of neurons # or a flag to indicate that the data repeats for each neuron if vertex_slice is None: if self.__repeat_type == StructRepeat.GLOBAL: n_repeats = 1 else: n_repeats = REPEAT_PER_NEURON_FLAG else: if self.__repeat_type == StructRepeat.GLOBAL: raise ValueError( "Global Structures cannot repeat more than once") n_repeats = vertex_slice.n_atoms # Start with bytes per repeat, n_repeats (from above), # total size of data written (0 as filled in later), # and number of fields in struct data = [self.numpy_dtype.itemsize, n_repeats, 0, len(self.__fields)] gen_data = list() # Go through all values and add in generator data for each for data_type, name in self.__fields: # Store the writer type based on the data type data.append(get_generator_type(data_type)) # We want the data generated "per neuron" regardless of how many - # there must be a single value for this to work if vertex_slice is None: self.__gen_data_one_for_all(data, gen_data, values, name, n_repeats) # If we know the array size, the values can vary per neuron else: self.__gen_data_for_slice( data, gen_data, values, name, vertex_slice) # Update with size *before* adding generator parameters data[2] = len(data) * BYTES_PER_WORD # Add the generator parameters after the rest of the data all_data = [numpy.array(data, dtype="uint32")] all_data.extend(gen_data) # Make it one return numpy.concatenate(all_data)
def __gen_data_one_for_all(self, data, gen_data, values, name, n_repeats): """ Generate data with a single value for all neurons. """ # How many sub-sets of repeats there are (1 in this case as # that one sub-set covers all neurons) data.append(1) # How many times to repeat the next bit data.append(n_repeats) # Get the value to write, of which there can only be one # (or else there will be an error here ;) if name in values: value = values[name] if not is_singleton(value): value = value.get_single_value_all() else: value = self.__default_values[name] # Write the id of the generator of the parameter data.append(param_generator_id(value)) # Add any parameters required to generate the values gen_data.append(param_generator_params(value)) def __gen_data_for_slice( self, data, gen_data, values, name, vertex_slice): """ Generate data with different values for each neuron. """ # If we have a range list for the value, generate for the range if name in values: vals = values[name] if is_singleton(vals): # If there is a single value, we can just use that # on all atoms data.append(1) data.append(vertex_slice.n_atoms) data.append(param_generator_id(vals)) gen_data.append(param_generator_params(vals)) else: # Store where to update with the number of items and # set to 0 to start n_items_index = len(data) data.append(0) n_items = 0 # Go through and get the data for each value ids = vertex_slice.get_raster_ids() for start, stop, value in vals.iter_ranges_by_ids(ids): n_items += 1 # This is the metadata data.append(stop - start) data.append(param_generator_id(value)) # This data goes after *all* the metadata gen_data.append(param_generator_params(value)) data[n_items_index] = n_items else: # Just a single value for all neurons from defaults value = self.__default_values[name] data.append(1) data.append(vertex_slice.n_atoms) data.append(param_generator_id(value)) gen_data.append(param_generator_params(value)) @property def is_generatable(self): """ Whether the data inside could be generated on machine. :rtype: bool """ return all(type_has_generator(data_type) for data_type, _name in self.__fields)
[docs] def read_data( self, data, values, data_offset=0, vertex_slice=None): """ Read a byte string of data and write to values. :param data: The data to be read :type data: bytes or bytearray :param ~spinn_utilities.ranged.RangeDictionary values: The values to update with the read data :param int data_offset: Index of the byte at the start of the valid data. :param int offset: The first index into values to write to. :param array_size: The number of structure copies to read, or `None` if this is a non-repeating structure. :type array_size: int or None """ n_items = 1 ids = None if vertex_slice is None: if self.__repeat_type != StructRepeat.GLOBAL: raise ValueError( "Repeating structures must specify an array size") elif self.__repeat_type == StructRepeat.GLOBAL: raise ValueError("Global Structures do not have a slice") else: n_items = vertex_slice.n_atoms ids = vertex_slice.get_raster_ids() if not self.__fields: return # Read in the data values numpy_data = numpy.frombuffer( data, offset=data_offset, dtype=self.numpy_dtype, count=n_items) for data_type, name in self.fields: # Ignore fields that can't be set if name in values: # Get the data to set for this item value = data_type.decode_numpy_array(numpy_data[name]) if self.__repeat_type == StructRepeat.GLOBAL: values[name] = value[0] else: values[name].set_value_by_ids(ids, value)