Using external CMAES library

pypuf/experiments/experiment/reliability_based_cmaes.py

@@ -0,0 +1,148 @@
+"""This module provides an experiment class which learns an instance of LTFArray with reliability based CMAES learner.
+It is based on the work from G. T. Becker in "The Gap Between Promise and Reality: On the Insecurity of
+XOR Arbiter PUFs". The learning algorithm applies Covariance Matrix Adaptation Evolution Strategies from N. Hansen
+in "The CMA Evolution Strategy: A Comparing Review".
+"""
+import numpy as np
+
+from pypuf.experiments.experiment.base import Experiment
+from pypuf.learner.evolution_strategies.reliability_based_cmaes import ReliabilityBasedCMAES
+from pypuf.simulation.arbiter_based.ltfarray import LTFArray, NoisyLTFArray
+from pypuf import tools
+
+
+class ExperimentReliabilityBasedCMAES(Experiment):
+    """This class implements an experiment for executing the reliability based CMAES learner for XOR LTF arrays.
+    It provides all relevant parameters as well as an instance of an LTF array to learn.
+    Furthermore, the learning results are being logged into csv files.
+    """
+
+    def __init__(
+            self, log_name,
+            seed_instance, k, n, transform, combiner, noisiness,
+            seed_challenges, num, reps,
+            seed_model, pop_size, limit_stag, limit_iter
+    ):
+        """Initialize an Experiment using the Reliability based CMAES Learner for modeling LTF Arrays
+        :param log_name:        Log name, Prefix of the name of the experiment log file
+        :param seed_instance:   PRNG seed used to create an LTF array instance to learn
+        :param k:               Width, the number of parallel LTFs in the LTF array
+        :param n:               Length, the number stages within the LTF array
+        :param transform:       Transformation function, the function that modifies the input within the LTF array
+        :param combiner:        Combiner, the function that combines particular chains' outputs within the LTF array
+        :param noisiness:       Noisiness, the relative scale of noise of instance compared to the scale of weights
+        :param seed_challenges: PRNG seed used to sample challenges
+        :param num:             Challenge number, the number of binary inputs (challenges) for the LTF array
+        :param reps:            Repetitions, the number of evaluations of every challenge (part of training_set)
+        :param seed_model:      PRNG seed used by the CMAES algorithm for sampling solution points
+        :param pop_size:        Population size, the number of sampled points of every CMAES iteration
+        :param limit_stag:      Stagnation limit, the maximal number of stagnating iterations within the CMAES
+        :param limit_iter:      Iteration limit, the maximal number of iterations within the CMAES
+        """
+        super().__init__(
+            log_name='%s.0x%x_%i_%i_%i_%i_%i' % (
+                log_name,
+                seed_instance,
+                k,
+                n,
+                num,
+                reps,
+                pop_size,
+            ),
+        )
+        # Instance of LTF array to learn
+        self.seed_instance = seed_instance
+        self.prng_i = np.random.RandomState(seed=self.seed_instance)
+        self.k = k
+        self.n = n
+        self.transform = transform
+        self.combiner = combiner
+        self.noisiness = noisiness
+        # Training set
+        self.seed_challenges = seed_challenges
+        self.prng_c = np.random.RandomState(seed=self.seed_instance)
+        self.num = num
+        self.reps = reps
+        self.training_set = None
+        # Parameters for CMAES
+        self.seed_model = seed_model
+        self.pop_size = pop_size
+        self.limit_s = limit_stag
+        self.limit_i = limit_iter
+        # Basic objects
+        self.instance = None
+        self.learner = None
+        self.model = None
+
+    def run(self):
+        """Initialize the instance, the training set and the learner
+        to then run the Reliability based CMAES with the given parameters
+        """
+        self.instance = NoisyLTFArray(
+            weight_array=LTFArray.normal_weights(self.n, self.k, random_instance=self.prng_i),
+            transform=self.transform,
+            combiner=self.combiner,
+            sigma_noise=NoisyLTFArray.sigma_noise_from_random_weights(self.n, 1, self.noisiness),
+            random_instance=self.prng_i,
+        )
+        self.training_set = tools.TrainingSet(self.instance, self.num, self.prng_c, self.reps)
+        self.learner = ReliabilityBasedCMAES(
+            self.training_set,
+            self.k,
+            self.n,
+            self.transform,
+            self.combiner,
+            self.pop_size,
+            self.limit_s,
+            self.limit_i,
+            self.seed_model,
+            self.progress_logger,
+        )
+        self.model = self.learner.learn()
+
+    def analyze(self):
+        """Analyze the learned model"""
+        assert self.model is not None
+        self.result_logger.info(
+            '0x%x\t0x%x\t0x%x\t%i\t%i\t%i\t%f\t%i\t%i\t%f\t%s\t%f\t%s\t%i\t%i\t%s',
+            self.seed_instance,
+            self.seed_challenges,
+            self.seed_model,
+            self.n,
+            self.k,
+            self.num,
+            self.noisiness,
+            self.reps,
+            self.pop_size,
+            1.0 - tools.approx_dist(self.instance, self.model, min(100000, 2 ** self.n), self.prng_c),
+            ','.join(map(str, self.calc_individual_accs())),
+            self.measured_time,
+            self.learner.stops,
+            self.learner.num_abortions,
+            self.learner.num_iterations,
+            ','.join(map(str, self.model.weight_array.flatten() / np.linalg.norm(self.model.weight_array.flatten()))),
+        )
+
+    def calc_individual_accs(self):
+        """Calculate the accuracies of individual chains of the learned model"""
+        transform = self.model.transform
+        combiner = self.model.combiner
+        accuracies = np.zeros(self.k)
+        poles = np.zeros(self.k)
+        for i in range(self.k):
+            chain_original = LTFArray(self.instance.weight_array[i, np.newaxis, :], transform, combiner)
+            for j in range(self.k):
+                chain_model = LTFArray(self.model.weight_array[j, np.newaxis, :], transform, combiner)
+                accuracy = tools.approx_dist(chain_original, chain_model, min(10000, 2 ** self.n), self.prng_c)
+                pole = 1
+                if accuracy < 0.5:
+                    accuracy = 1.0 - accuracy
+                    pole = -1
+                if accuracy > accuracies[i]:
+                    accuracies[i] = accuracy
+                    poles[i] = pole
+        accuracies *= poles
+        for i in range(self.k):
+            if accuracies[i] < 0:
+                accuracies[i] += 1
+        return accuracies

pypuf/learner/evolution_strategies/reliability_based_cmaes.py

@@ -0,0 +1,241 @@
+"""This module provides a learner exploiting different reliabilities of challenges evaluated several times on an
+XOR Arbiter PUF. It is based on the work from G. T. Becker in "The Gap Between Promise and Reality: On the Insecurity
+of XOR Arbiter PUFs". The learning algorithm applies Covariance Matrix Adaptation Evolution Strategies from
+N. Hansen in "The CMA Evolution Strategy: A Comparing Review".
+"""
+import sys
+import contextlib
+import numpy as np
+from scipy.special import gamma
+from scipy.linalg import norm
+import cma
+
+from pypuf import tools
+from pypuf.learner.base import Learner
+from pypuf.simulation.arbiter_based.ltfarray import LTFArray
+
+
+class ReliabilityBasedCMAES(Learner):
+    """This class implements a learner for XOR LTF arrays. Thus, by means of a CMAES algorithm a model
+    is created similar to the original LTF array. This process is based on the information of reliability
+    originating from multiple repeatedly evaluated challenges.
+    """
+
+    # Constants
+    CONST_EPSILON = 0.1
+    FREQ_ABORTION_CHECK = 50
+    FREQ_LOGGING = 1
+    THRESHOLD_DIST = 0.25
+
+    # Semi-constant :-)
+    approx_challenge_num = 10000
+
+    def __init__(self, training_set, k, n, transform, combiner,
+                 pop_size, limit_stag, limit_iter, random_seed, logger):
+        """Initialize a Reliability based CMAES Learner for the specified LTF array
+
+        :param training_set:    Training set, a data structure containing repeated challenge response pairs
+        :param k:               Width, the number of parallel LTFs in the LTF array
+        :param n:               Length, the number stages within the LTF array
+        :param transform:       Transformation function, the function that modifies the input within the LTF array
+        :param combiner:        Combiner, the function that combines particular chains' outputs within the LTF array
+        :param pop_size:        Population size, the number of sampled points of every CMAES iteration
+        :param limit_stag:      Stagnation limit, the maximal number of stagnating iterations within the CMAES
+        :param limit_iter:      Iteration limit, the maximal number of iterations within the CMAES
+        :param random_seed:     PRNG seed used by the CMAES algorithm for sampling solution points
+        :param logger:          Logger, the instance that logs detailed information every learning iteration
+        """
+        self.training_set = training_set
+        self.k = k
+        self.n = n
+        self.transform = transform
+        self.combiner = combiner
+        self.pop_size = pop_size
+        self.limit_s = limit_stag
+        self.limit_i = limit_iter
+        self.prng = np.random.RandomState(random_seed)
+        self.chains_learned = np.zeros((self.k, self.n))
+        self.num_iterations = 0
+        self.stops = ''
+        self.num_abortions = 0
+        self.num_learned = 0
+        self.logger = logger
+
+        if 2**n < self.approx_challenge_num:
+            self.approx_challenge_num = 2 ** n
+
+    def learn(self):
+        """Compute a model according to the given LTF Array parameters and training set
+        Note that this function can take long to return
+        :return: pypuf.simulation.arbiter_based.LTFArray
+                 The computed model.
+        """
+
+        def log_state(ellipsoid):
+            """Log a snapshot of learning variables while running"""
+            if self.logger is None:
+                return
+            self.logger.debug(
+                '%i\t%f\t%f\t%i\t%i\t%s',
+                self.num_iterations,
+                ellipsoid.sigma,
+                fitness(ellipsoid.mean),
+                self.num_learned,
+                self.num_abortions,
+                ','.join(map(str, list(ellipsoid.mean))),
+            )
+
+        # Preparation
+        epsilon = np.sqrt(self.n) * self.CONST_EPSILON
+        fitness = self.create_fitness_function(
+            challenges=self.training_set.challenges,
+            measured_rels=self.measure_rels(self.training_set.responses),
+            epsilon=epsilon,
+            transform=self.transform,
+            combiner=self.combiner,
+        )
+        normalize = np.sqrt(2) * gamma(self.n / 2) / gamma((self.n - 1) / 2)
+        mean_start = np.zeros(self.n)
+        step_size_start = 1
+        options = {
+            'seed': 0,
+            'pop': self.pop_size,
+            'maxiter': self.limit_i,
+            'tolstagnation': self.limit_s,
+        }
+
+        # Learn all individual LTF arrays (chains)
+        with self.avoid_printing():
+            while self.num_learned < self.k:
+                aborted = False
+                options['seed'] = self.prng.randint(2 ** 32)
+                is_same_solution = self.create_abortion_function(
+                    chains_learned=self.chains_learned,
+                    num_learned=self.num_learned,
+                    transform=self.transform,
+                    combiner=self.combiner,
+                    threshold=self.THRESHOLD_DIST,
+                )
+                search = cma.CMAEvolutionStrategy(x0=mean_start, sigma0=step_size_start, inopts=options)
+                counter = 1
+                # Learn individual LTF array using abortion if evolutionary search approximates previous a solution
+                while not search.stop():
+                    curr_points = search.ask()  # Sample new solution points
+                    search.tell(curr_points, [fitness(point) for point in curr_points])
+                    self.num_iterations += 1
+                    if counter % self.FREQ_LOGGING == 0:
+                        log_state(search)
+                    counter += 1
+                    if counter % self.FREQ_ABORTION_CHECK == 0:
+                        if is_same_solution(search.mean):
+                            self.num_abortions += 1
+                            aborted = True
+                            break
+                solution = search.result.xbest
+
+                # Include normalized solution, if it is different from previous ones
+                if not aborted:
+                    self.chains_learned[self.num_learned] = normalize * solution / norm(solution)
+                    self.num_learned += 1
+                    if self.stops != '':
+                        self.stops += ','
+                    self.stops += '_'.join(list(search.stop()))
+
+        # Polarize the learned combined LTF array
+        majority_responses = self.majority_responses(self.training_set.responses)
+        self.chains_learned = self.polarize_chains(
+            chains_learned=self.chains_learned,
+            challenges=self.training_set.challenges,
+            majority_responses=majority_responses,
+            transform=self.transform,
+            combiner=self.combiner,
+        )
+        return LTFArray(self.chains_learned, self.transform, self.combiner)
+
+    @staticmethod
+    @contextlib.contextmanager
+    def avoid_printing():
+        """Avoid printing on sys.stdout while learning"""
+        save_stdout = sys.stdout
+        sys.stdout = open('/dev/null', 'w')
+        yield
+        sys.stdout.close()
+        sys.stdout = save_stdout
+
+    @staticmethod
+    def create_fitness_function(challenges, measured_rels, epsilon, transform, combiner):
+        """Return a fitness function on a fixed set of challenges and corresponding reliabilities"""
+        this = ReliabilityBasedCMAES
+
+        def fitness(individual):
+            """Return individuals sorted by their correlation coefficient as fitness"""
+            ltf_array = LTFArray(individual[np.newaxis, :], transform, combiner)
+            delay_diffs = ltf_array.val(challenges)
+            reliabilities = np.zeros(np.shape(delay_diffs))
+            indices_of_reliable = np.abs(delay_diffs[:]) > epsilon
+            reliabilities[indices_of_reliable] = 1
+            correlation = this.calc_corr(reliabilities, measured_rels)
+            obj_vals = 1 - (1 + correlation)/2
+            return obj_vals
+
+        return fitness
+
+    @staticmethod
+    def calc_corr(reliabilities, measured_rels):
+        """Return pearson correlation coefficient between reliability arrays of individual and instance"""
+        if np.var(reliabilities[:]) == 0:  # Avoid dividing by zero
+            return -1
+        else:
+            return np.corrcoef(reliabilities[:], measured_rels)[0, 1]
+
+    @staticmethod
+    def create_abortion_function(chains_learned, num_learned, transform, combiner, threshold):
+        """Return an abortion function on a fixed set of challenges and LTFs"""
+        this = ReliabilityBasedCMAES
+        weight_arrays = chains_learned[:num_learned, :]
+        learned_ltf_arrays = list(this.build_individual_ltf_arrays(weight_arrays, transform, combiner))
+
+        def is_same_solution(solution):
+            """Return True, if the current solution mean within CMAES is similar to a previously learned LTF array"""
+            if num_learned == 0:
+                return False
+            new_ltf_array = LTFArray(solution[np.newaxis, :], transform, combiner)
+            for current_ltf_array in learned_ltf_arrays:
+                dist = tools.approx_dist(current_ltf_array, new_ltf_array, this.approx_challenge_num)
+                if dist < threshold or dist > (1 - threshold):
+                    return True
+            return False
+
+        return is_same_solution
+
+    @staticmethod
+    def polarize_chains(chains_learned, challenges, majority_responses, transform, combiner):
+        """Return the correctly polarized combined LTF array"""
+        model = LTFArray(chains_learned, transform, combiner)
+        responses_model = model.eval(challenges)
+        num, _ = np.shape(challenges)
+        accuracy = np.count_nonzero(responses_model == majority_responses) / num
+        polarized_chains = chains_learned
+        if accuracy < 0.5:
+            polarized_chains[0, :] *= -1
+        return polarized_chains
+
+    @staticmethod
+    def build_individual_ltf_arrays(weight_arrays, transform, combiner):
+        """Return iterator over LTF arrays created out of every individual"""
+        pop_size, _ = np.shape(weight_arrays)
+        for i in range(pop_size):
+            yield LTFArray(weight_arrays[i, np.newaxis, :], transform, combiner)
+
+    @staticmethod
+    def majority_responses(responses):
+        """Return the common responses out of repeated responses"""
+        return np.sign(np.sum(responses, axis=0))
+
+    @staticmethod
+    def measure_rels(responses):
+        """Return array of measured reliabilities of instance"""
+        measured_rels = np.abs(np.sum(responses, axis=0))
+        if np.var(measured_rels) == 0:
+            raise Exception('The challenges\' reliabilities evaluated on the instance to learn are to high!')
+        return measured_rels