evaluators.py

"""Cell evaluator class"""

"""
Copyright (c) 2016-2020, EPFL/Blue Brain Project

 This file is part of BluePyOpt <https://github.com/BlueBrain/BluePyOpt>

 This library is free software; you can redistribute it and/or modify it under
 the terms of the GNU Lesser General Public License version 3.0 as published
 by the Free Software Foundation.

 This library 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 Lesser General Public License for more
 details.

 You should have received a copy of the GNU Lesser General Public License
 along with this library; if not, write to the Free Software Foundation, Inc.,
 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
"""


# pylint: disable=W0511

import logging
logger = logging.getLogger(__name__)

import bluepyopt as bpopt
import bluepyopt.tools


class CellEvaluator(bpopt.evaluators.Evaluator):

    """Simple cell class"""

    def __init__(
            self,
            cell_model=None,
            param_names=None,
            fitness_protocols=None,
            fitness_calculator=None,
            isolate_protocols=None,
            sim=None,
            use_params_for_seed=False,
            timeout=None):
        """Constructor

        Args:
            cell_model (ephys.models.CellModel): CellModel object to evaluate
            param_names (list of str): names of the parameters
                (parameters will be initialised in this order)
            fitness_protocols (dict of str -> ephys.protocols.Protocol):
                protocols used during the fitness evaluation
            fitness_calculator (ObjectivesCalculator):
                ObjectivesCalculator object used for the transformation of
                Responses into Objective objects
            isolate_protocols (bool): whether to use multiprocessing to
                isolate the simulations
                (disabling this could lead to unexpected behavior, and might
                hinder the reproducability of the simulations)
            sim (ephys.simulators.NrnSimulator): simulator to use for the cell
                evaluation
            use_params_for_seed (bool): use a hashed version of the parameter
                dictionary as a seed for the simulator
            timeout (int): duration in second after which a Process will
                be interrupted when using multiprocessing
        """

        super(CellEvaluator, self).__init__(
            fitness_calculator.objectives,
            cell_model.params_by_names(param_names))

        if sim is None:
            raise ValueError("CellEvaluator: you have to provide a Simulator "
                             "object to the 'sim' argument of the "
                             "CellEvaluator constructor")
        self.sim = sim

        self.cell_model = cell_model
        self.param_names = param_names
        # Stimuli used for fitness calculation
        self.fitness_protocols = fitness_protocols
        # Fitness value calculator
        self.fitness_calculator = fitness_calculator

        self.isolate_protocols = isolate_protocols
        self.timeout = timeout
        self.use_params_for_seed = use_params_for_seed

    def param_dict(self, param_array):
        """Convert param_array in param_dict"""
        param_dict = {}
        for param_name, param_value in \
                zip(self.param_names, param_array):
            param_dict[param_name] = param_value

        return param_dict

    def objective_dict(self, objective_array):
        """Convert objective_array in objective_dict"""
        objective_dict = {}
        objective_names = [objective.name
                           for objective in self.fitness_calculator.objectives]

        if len(objective_names) != len(objective_array):
            raise Exception(
                'CellEvaluator: list given to objective_dict() '
                'has wrong number of objectives')

        for objective_name, objective_value in \
                zip(objective_names, objective_array):
            objective_dict[objective_name] = objective_value

        return objective_dict

    def objective_list(self, objective_dict):
        """Convert objective_dict in objective_list"""
        objective_list = []
        objective_names = [objective.name
                           for objective in self.fitness_calculator.objectives]
        for objective_name in objective_names:
            objective_list.append(objective_dict[objective_name])

        return objective_list

    @staticmethod
    def seed_from_param_dict(param_dict):
        """Return a seed value based on a param_dict"""

        sorted_keys = sorted(param_dict.keys())

        string = ''
        for key in sorted_keys:
            string += '%s%s' % (key, str(param_dict[key]))

        return bluepyopt.tools.uint32_seed(string)

    def run_protocol(
            self,
            protocol,
            param_values,
            isolate=None,
            cell_model=None,
            sim=None,
            timeout=None):
        """Run protocol"""

        sim = self.sim if sim is None else sim

        if self.use_params_for_seed:
            sim.random123_globalindex = self.seed_from_param_dict(param_values)

        # Try/except added for backward compatibility
        try:
            return protocol.run(
                self.cell_model if cell_model is None else cell_model,
                param_values,
                sim=sim,
                isolate=isolate,
                timeout=timeout)
        except TypeError as e:
            if "unexpected keyword" in str(e):
                return protocol.run(
                    self.cell_model if cell_model is None else cell_model,
                    param_values,
                    sim=sim,
                    isolate=isolate)
            else:
                raise

    def run_protocols(self, protocols, param_values):
        """Run a set of protocols"""

        responses = {}

        for protocol in protocols:
            responses.update(self.run_protocol(
                protocol,
                param_values=param_values,
                isolate=self.isolate_protocols,
                timeout=self.timeout))

        return responses

    def evaluate_with_dicts(self, param_dict=None, target='scores'):
        """Run evaluation with dict as input and output"""

        if target not in ['scores', 'values']:
            raise Exception(
                'CellEvaluator: target has to be "scores" or "values".')

        if self.fitness_calculator is None:
            raise Exception(
                'CellEvaluator: need fitness_calculator to evaluate')

        logger.debug('Evaluating %s', self.cell_model.name)

        responses = self.run_protocols(
            self.fitness_protocols.values(),
            param_dict)

        if target == 'scores':
            return self.fitness_calculator.calculate_scores(responses)

        elif target == 'values':
            return self.fitness_calculator.calculate_values(responses)

    def evaluate_with_lists(self, param_list=None, target='scores'):
        """Run evaluation with lists as input and outputs"""

        param_dict = self.param_dict(param_list)

        obj_dict = self.evaluate_with_dicts(
            param_dict=param_dict, target=target
        )

        return self.objective_list(obj_dict)

    def init_simulator_and_evaluate_with_lists(
        self, param_list=None, target='scores'
    ):
        """Set NEURON variables and run evaluation with lists.

        Setting the NEURON variables is necessary when using ipyparallel,
        since the new subprocesses have pristine NEURON.
        """
        self.sim.initialize()
        return self.evaluate_with_lists(param_list=param_list, target=target)

    def evaluate(self, param_list=None, target='scores'):
        """Run evaluation with lists as input and outputs"""

        return self.evaluate_with_lists(param_list, target=target)

    def __str__(self):

        content = 'cell evaluator:\n'

        content += '  cell model:\n'
        if self.cell_model is not None:
            content += '    %s\n' % str(self.cell_model)

        content += '  fitness protocols:\n'
        if self.fitness_protocols is not None:
            for fitness_protocol in self.fitness_protocols.values():
                content += '    %s\n' % str(fitness_protocol)

        content += '  fitness calculator:\n'
        if self.fitness_calculator is not None:
            content += '    %s\n' % str(self.fitness_calculator)

        return content