Source code for pyNN.utility

# encoding: utf-8
"""
A collection of utility functions and classes.

Functions:
    notify()          - send an e-mail when a simulation has finished.
    get_script_args() - get the command line arguments to the script, however
                        it was run (python, nrniv, mpirun, etc.).
    init_logging()    - convenience function for setting up logging to file and
                        to the screen.

    Timer    - a convenience wrapper around the time.time() function from the
               standard library.

:copyright: Copyright 2006-2020 by the PyNN team, see AUTHORS.
:license: CeCILL, see LICENSE for details.

"""

from __future__ import print_function, division
# If there is a settings.py file on the path, defaults will be
# taken from there.
try:
    from settings import SMTPHOST, EMAIL
except ImportError:
    SMTPHOST = None
    EMAIL = None
try:
    unicode
except NameError:
    unicode = str
import sys
import logging
import time
import os
from datetime import datetime
import functools
import numpy
from importlib import import_module

from pyNN.core import deprecated


[docs]def notify(msg="Simulation finished.", subject="Simulation finished.", smtphost=SMTPHOST, address=EMAIL): """Send an e-mail stating that the simulation has finished.""" if not (smtphost and address): print("SMTP host and/or e-mail address not specified.\nUnable to send notification message.") else: import smtplib import datetime msg = ("From: %s\r\nTo: %s\r\nSubject: %s\r\n\r\n") % (address, address, subject) + msg msg += "\nTimestamp: %s" % datetime.datetime.now().strftime("%H:%M:%S, %F") server = smtplib.SMTP(smtphost) server.sendmail(address, address, msg) server.quit()
def get_script_args(n_args, usage=''): """ Get command line arguments. This works by finding the name of the main script and assuming any arguments after this in sys.argv are arguments to the script. It would be nicer to use optparse, but this doesn't seem to work too well with nrniv or mpirun. """ calling_frame = sys._getframe(1) if '__file__' in calling_frame.f_locals: script = calling_frame.f_locals['__file__'] try: script_index = sys.argv.index(script) except ValueError: try: script_index = sys.argv.index(os.path.abspath(script)) except ValueError: script_index = 0 else: script_index = 0 args = sys.argv[script_index + 1:script_index + 1 + n_args] if len(args) != n_args: usage = usage or "Script requires %d arguments, you supplied %d" % (n_args, len(args)) raise Exception(usage) return args
[docs]def get_simulator(*arguments): """ Import and return a PyNN simulator backend module based on command-line arguments. The simulator name should be the first positional argument. If your script needs additional arguments, you can specify them as (name, help_text) tuples. If you need more complex argument handling, you should use argparse directly. Returns (simulator, command-line arguments) """ import argparse parser = argparse.ArgumentParser() parser.add_argument("simulator", help="neuron, nest, brian or another backend simulator") for argument in arguments: arg_name, help_text = argument[:2] extra_args = {} if len(argument) > 2: extra_args = argument[2] parser.add_argument(arg_name, help=help_text, **extra_args) args = parser.parse_args() sim = import_module("pyNN.%s" % args.simulator) return sim, args
[docs]def init_logging(logfile, debug=False, num_processes=1, rank=0, level=None): """ Simple configuration of logging. """ # allow logfile == None # which implies output to stderr # num_processes and rank should be obtained using mpi4py, rather than having them as arguments if logfile: if num_processes > 1: logfile += '.%d' % rank logfile = os.path.abspath(logfile) # prefix log messages with mpi rank mpi_prefix = "" if num_processes > 1: mpi_prefix = 'Rank %d of %d: ' % (rank, num_processes) if debug: log_level = logging.DEBUG else: log_level = logging.INFO # allow user to override exact log_level if level: log_level = level logging.basicConfig(level=log_level, format=mpi_prefix + '%(asctime)s %(levelname)-8s [%(name)s] %(message)s (%(pathname)s[%(lineno)d]:%(funcName)s)', filename=logfile, filemode='w') return logging.getLogger("PyNN")
[docs]def save_population(population, filename, variables=None): """ Saves the spike_times of a population and the size, structure, labels such that one can load it back into a SpikeSourceArray population using the load_population() function. """ import shelve s = shelve.open(filename) s['spike_times'] = population.getSpikes() s['label'] = population.label s['size'] = population.size s['structure'] = population.structure # should perhaps just save the positions? variables_dict = {} if variables: for variable in variables: variables_dict[variable] = getattr(population, variable) s['variables'] = variables_dict s.close()
[docs]def load_population(filename, sim): """ Loads a population that was saved with the save_population() function into SpikeSourceArray. """ import shelve s = shelve.open(filename) ssa = getattr(sim, "SpikeSourceArray") population = getattr(sim, "Population")(s['size'], ssa, structure=s['structure'], label=s['label']) # set the spiketimes spikes = s['spike_times'] for neuron in range(s['size']): spike_times = spikes[spikes[:, 0] == neuron][:, 1] neuron_in_new_population = neuron + population.first_id index = population.id_to_index(neuron_in_new_population) population[index].set_parameters(**{'spike_times': spike_times}) # set the variables for variable, value in s['variables'].items(): setattr(population, variable, value) s.close() return population
def normalized_filename(root, basename, extension, simulator, num_processes=None, use_iso8601=False): """ Generate a file path containing a timestamp and information about the simulator used and the number of MPI processes. The date is used as a sub-directory name, the date & time are included in the filename. If use_iso8601 is True, follow https://en.wikipedia.org/wiki/ISO_8601 """ timestamp = datetime.now() if use_iso8601: date = timestamp.strftime("%Y-%m-%d") date_time = timestamp.strftime("%Y-%m-%dT%H:%M:%S") else: date = timestamp.strftime("%Y%m%d") date_time = timestamp.strftime("%Y%m%d-%H%M%S") if num_processes: np = "_np%d" % num_processes else: np = "" return os.path.join(root, date, "%s_%s%s_%s.%s" % (basename, simulator, np, date_time, extension)) def connection_plot(projection, positive='O', zero='.', empty=' ', spacer=''): """ """ connection_array = projection.get('weight', format='array') image = numpy.zeros_like(connection_array, dtype=unicode) old_settings = numpy.seterr(invalid='ignore') # ignore the complaint that x > 0 is invalid for NaN image[connection_array > 0] = positive image[connection_array == 0] = zero numpy.seterr(**old_settings) # restore original floating point error settings image[numpy.isnan(connection_array)] = empty return '\n'.join([spacer.join(row) for row in image])
[docs]class Timer(object): """ For timing script execution. Timing starts on creation of the timer. """ def __init__(self): self.start() self.marks = []
[docs] def start(self): """Start/restart timing.""" self._start_time = time.time() self._last_check = self._start_time
[docs] def elapsed_time(self, format=None): """ Return the elapsed time in seconds but keep the clock running. If called with ``format="long"``, return a text representation of the time. Examples:: >>> timer.elapsed_time() 987 >>> timer.elapsed_time(format='long') 16 minutes, 27 seconds """ current_time = time.time() elapsed_time = current_time - self._start_time if format == 'long': elapsed_time = Timer.time_in_words(elapsed_time) self._last_check = current_time return elapsed_time
@deprecated('elapsed_time()') def elapsedTime(self, format=None): return self.elapsed_time(format)
[docs] def reset(self): """Reset the time to zero, and start the clock.""" self.start()
[docs] def diff(self, format=None): # I think delta() would be a better name for this method. """ Return the time since the last time :meth:`elapsed_time()` or :meth:`diff()` was called. If called with ``format='long'``, return a text representation of the time. """ current_time = time.time() time_since_last_check = current_time - self._last_check self._last_check = current_time if format == 'long': time_since_last_check = Timer.time_in_words(time_since_last_check) return time_since_last_check
[docs] @staticmethod def time_in_words(s): """ Formats a time in seconds as a string containing the time in days, hours, minutes, seconds. Examples:: >>> Timer.time_in_words(1) 1 second >>> Timer.time_in_words(123) 2 minutes, 3 seconds >>> Timer.time_in_words(24*3600) 1 day """ # based on http://mail.python.org/pipermail/python-list/2003-January/181442.html T = {} T['year'], s = divmod(s, 31556952) min, T['second'] = divmod(s, 60) h, T['minute'] = divmod(min, 60) T['day'], T['hour'] = divmod(h, 24) def add_units(val, units): return "%d %s" % (int(val), units) + (val > 1 and 's' or '') return ', '.join([add_units(T[part], part) for part in ('year', 'day', 'hour', 'minute', 'second') if T[part] > 0])
[docs] def mark(self, label): """ Store the time since the last time since the last time :meth:`elapsed_time()`, :meth:`diff()` or :meth:`mark()` was called, together with the provided label, in the attribute 'marks'. """ self.marks.append((label, self.diff()))
[docs]class ProgressBar(object): """ Create a progress bar in the shell. """ def __init__(self, width=77, char="#", mode="fixed"): self.char = char self.mode = mode if self.mode not in ['fixed', 'dynamic']: self.mode = 'fixed' self.width = width
[docs] def set_level(self, level): """ Rebuild the bar string based on `level`, which should be a number between 0 and 1. """ if level < 0: level = 0 if level > 1: level = 1 # figure the proper number of 'character' make up the bar all_full = self.width - 2 num_hashes = int(round(level * all_full)) if self.mode == 'dynamic': # build a progress bar with self.char (to create a dynamic bar # where the percent string moves along with the bar progress. bar = self.char * num_hashes else: # build a progress bar with self.char and spaces (to create a # fixed bar (the percent string doesn't move) bar = self.char * num_hashes + ' ' * (all_full - num_hashes) bar = u'[ %s ] %3.0f%%' % (bar, 100 * level) print(bar, end=u' \r') sys.stdout.flush()
def __call__(self, level): self.set_level(level)
class SimulationProgressBar(ProgressBar): def __init__(self, interval, t_stop, char="#", mode="fixed"): super(SimulationProgressBar, self).__init__(width=int(t_stop / interval), char=char, mode=mode) self.interval = interval self.t_stop = t_stop def __call__(self, t): self.set_level(t / self.t_stop) return t + self.interval def assert_arrays_equal(a, b): import numpy assert isinstance(a, numpy.ndarray), "a is a %s" % type(a) assert isinstance(b, numpy.ndarray), "b is a %s" % type(b) assert a.shape == b.shape, "%s != %s" % (a, b) assert (a.flatten() == b.flatten()).all(), "%s != %s" % (a, b) def assert_arrays_almost_equal(a, b, threshold): import numpy assert isinstance(a, numpy.ndarray), "a is a %s" % type(a) assert isinstance(b, numpy.ndarray), "b is a %s" % type(b) assert a.shape == b.shape, "%s != %s" % (a, b) assert (abs(a - b) < threshold).all(), "max(|a - b|) = %s" % (abs(a - b)).max() def sort_by_column(a, col): # see stackoverflow.com/questions/2828059/ return a[a[:, col].argsort(), :] # based on http://wiki.python.org/moin/PythonDecoratorLibrary#Memoize class forgetful_memoize(object): """ Decorator that caches the result from the last time a function was called. If the next call uses the same arguments, the cached value is returned, and not re-evaluated. If the next call uses different arguments, the cached value is overwritten. The use case is when the same, heavy-weight function is called repeatedly with the same arguments in different places. """ def __init__(self, func): self.func = func self.cached_args = None self.cached_value = None def __call__(self, *args): if args == self.cached_args: print("using cached value") return self.cached_value else: #print("calculating value") value = self.func(*args) self.cached_args = args self.cached_value = value return value def __get__(self, obj, objtype): """Support instance methods.""" return functools.partial(self.__call__, obj)