text:
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# Copyright 2018 Xanadu Quantum Technologies Inc.

# 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

#     http://www.apache.org/licenses/LICENSE-2.0

# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""
Unit tests for qmlt learners
============================

.. codeauthor:: Maria Schuld <maria@xanadu.ai>

"""
import os
import unittest
from io import StringIO
import sys
from copy import deepcopy
import shutil
import tensorflow as tf
import numpy as np
import matplotlib as mpl
if os.environ.get('DISPLAY', '') == '':
    print('no display found. Using non-interactive Agg backend')
    mpl.use('Agg')
import matplotlib.pyplot as plt
from qmlt.numerical import CircuitLearner as Learner_num
from qmlt.tf import CircuitLearner as Learner_tf
from qmlt.numerical.helpers import make_param as make_param_num
from qmlt.tf.helpers import make_param as make_param_tf
import strawberryfields as sf
from strawberryfields.ops import Dgate, BSgate
from qmlt.tf.learner import (_check_Y as tf_check_Y,
                             _check as tf_check,
                             _check_shuffle as tf_check_shuffle,
                             _check_X as tf_check_X,
                             _check_batch_size as tf_check_batch_size,
                             _check_steps as tf_check_steps)
from qmlt.numerical.learner import (_check_Y as num_check_Y,
                                    _check as num_check,
                                    _check_X as num_check_X,
                                    _check_batch_size as num_check_batch_size,
                                    _check_steps as num_check_steps,
                                    _check_logs as num_check_logs)

class BaseLearnerTest(unittest.TestCase):
    """
    Baseclass for learner tests.
    """

    def setUp(self):
        return

    def assertAlmostEqual(self, one, two, msg=None, delta=1e-6):
        """
        Assert that ``one`` almost equal to ``two``. Default threshold is 1e-6.
        """
        if one == two:
            return
        if np.abs(one - two) <= delta:
            return
        standardMsg = 'The values differ more than threshold delta = {}'.format(delta)
        msg = self._formatMessage(msg, standardMsg)
        raise self.failureException(msg)

    def assertAlmostEqualArray(self, one_arr, two_arr, msg=None, delta=1e-6):
        """
        Assert that the elements in 1-d array ``one`` are almost equal to those in 1-d array ``two``.
        Default threshold is 1e-6.
        """
        if np.allclose(one_arr, two_arr, atol=delta):
            return
        standardMsg = 'The values differ more than threshold delta = {}'.format(delta)
        msg = self._formatMessage(msg, standardMsg)
        raise self.failureException(msg)

    def assertDictHasKey(self, dict, key, msg=None):
        """
        Assert that dict has the key.
        """
        if key in dict:
            return
        standardMsg = 'The key {} is not in the dictionary'.format(key)
        msg = self._formatMessage(msg, standardMsg)
        raise self.failureException(msg)

    def assertEquals(self, one, two, msg=None):
        """
        Assert that one == two. Compares lists, tuples and strings.
        """
        if one == two:
            return
        standardMsg = 'The strings {} and {} are not equal'.format(one, two)
        msg = self._formatMessage(msg, standardMsg)
        raise self.failureException(msg)

    def assertStringNotEmpty(self, mystring, msg=None):
        """
        Assert that str is a nonempty string.
        """
        if not mystring:
            standardMsg = 'String is empty'
            msg = self._formatMessage(msg, standardMsg)
            raise self.failureException(msg)
        else:
            return

    def assertPathExists(self, path, msg=None):
        if os.path.exists(os.path.dirname(os.path.abspath(__file__))+"/"+path):
            return
        standardMsg = 'The path {} does not exist'.format(path)
        msg = self._formatMessage(msg, standardMsg)
        raise self.failureException(msg)

class TestNumericalLearnerOptimization(BaseLearnerTest):
    """
    Test the numeric learner for optimization tasks
    """

    def setUp(self):
        super().setUp()

        my_init_params = [make_param_num(constant=0.1, name='alpha')]

        def circuit(params):
            eng, q = sf.Engine(1)
            with eng:
                Dgate(params[0]) | q[0]
            state = eng.run('gaussian')
            circuit_output = state.fock_prob([1])
            return circuit_output

        def myloss(circuit_output):
            return -circuit_output

        self.hyperparams = {'circuit': circuit,
                            'init_circuit_params': my_init_params,
                            'task': 'optimization',
                            'loss': myloss,
                            'print_log': False
                            }

        self.dummy_input = np.array([[0., 1.]])

    def test_basic_optimization_num(self):
        """
        Test to optimize a StrawberryFields Dgate with minimal parameters.
        """
        learner = Learner_num(hyperparams=self.hyperparams)
        learner.train_circuit(steps=10)
        final_par = learner.get_circuit_parameters()['alpha']
        self.assertAlmostEqual(final_par, 0.1213333101, delta=0.0001)

    def test_optimization_custom_modeldir(self):
        """
        Test to optimize a StrawberryFields Dgate with custom model directory.
        """
        _ = Learner_num(hyperparams=self.hyperparams, model_dir="newpath/modeldir")
        self.assertPathExists('../newpath/')

    def test_optimization_with_regul_num(self):
        """
        Test to optimize a StrawberryFields Dgate with regularization.
        """
        def myregularizer(regularized_params):
            return sum(regularized_params)

        new_hyperparams = deepcopy(self.hyperparams)
        new_hyperparams['regularizer'] = myregularizer
        new_hyperparams['regularization_strength'] = 1.
        learner = Learner_num(hyperparams=new_hyperparams)
        learner.train_circuit(steps=10)
        final_par = learner.get_circuit_parameters()['alpha']
        self.assertAlmostEqual(final_par, 0.1213333101, delta=0.0001)

    def test_optimization_with_learningrate_num(self):
        """
        Test to optimize a StrawberryFields Dgate with a custom learning rate.
        """
        new_hyperparams = deepcopy(self.hyperparams)
        new_hyperparams['init_learning_rate'] = 2.
        learner = Learner_num(hyperparams=new_hyperparams)
        learner.train_circuit(steps=10)
        final_par = learner.get_circuit_parameters()['alpha']
        self.assertAlmostEqual(final_par, 1.449175616872, delta=0.0001)

    def test_optimization_with_decay_num(self):
        """
        Test to optimize a StrawberryFields Dgate with learning rate decay.
        """
        new_hyperparams = deepcopy(self.hyperparams)
        new_hyperparams['decay'] = 0.5
        learner = Learner_num(hyperparams=new_hyperparams)
        learner.train_circuit(steps=10)
        final_par = learner.get_circuit_parameters()['alpha']
        self.assertAlmostEqual(final_par, 0.108182953535, delta=0.0001)

    def test_optimization_with_warmstart_num(self):
        """
        Test to optimize a StrawberryFields Dgate with warm start.
        """
        learner = Learner_num(hyperparams=self.hyperparams)
        learner.train_circuit(steps=10)
        new_hyperparams = deepcopy(self.hyperparams)
        new_hyperparams['warm_start'] = True
        learner2 = Learner_num(hyperparams=new_hyperparams)
        learner2.train_circuit(steps=1)
        final_par = int(np.loadtxt("logsNUM/global_step.txt"))
        self.assertAlmostEqual(final_par, 11)

    def test_optimization_with_warmstart_raises_error_num(self):
        """
        Test if optimizing a StrawberryFields Dgate with warm start but no parameter file raises error.
        """
        with self.assertRaises(IOError):
            shutil.rmtree('logsNUM', ignore_errors=True)
            new_hyperparams = deepcopy(self.hyperparams)
            new_hyperparams['warm_start'] = True
            learner = Learner_num(hyperparams=new_hyperparams)
            learner.train_circuit(steps=1)
        with self.assertRaises(IOError):
            new_hyperparams['warm_start'] = False
            learner = Learner_num(hyperparams=new_hyperparams)
            learner.train_circuit(steps=1)
            shutil.rmtree('logsNUM/global_step.txt', ignore_errors=False)
            learner.train_circuit(steps=1)
        with self.assertRaises(IOError):
            new_hyperparams['warm_start'] = True
            new_hyperparams['init_circuit_params'] = [make_param_num(constant=0.1), make_param_num(constant=0.1)]
            learner = Learner_num(hyperparams=new_hyperparams)
            learner.train_circuit(steps=1)

    def test_optimization_with_adaptivelr_num(self):
        """
        Test to optimize a StrawberryFields Dgate with adaptive learning
        rate.
        """
        new_hyperparams = deepcopy(self.hyperparams)
        new_hyperparams['adaptive_learning_rate_threshold'] = 0.5
        learner = Learner_num(hyperparams=new_hyperparams)
        learner.train_circuit(steps=10)
        final_par = learner.get_circuit_parameters()['alpha']
        self.assertAlmostEqual(final_par, 0.346251338090, delta=0.0001)

    def test_optimization_with_logging_num(self):
        """
        Test to optimize a StrawberryFields Dgate while saving a log.
        """

        def circuit2(params):
            eng, q = sf.Engine(1)
            with eng:
                Dgate(params[0]) | q[0]
            state = eng.run('gaussian')
            circuit_output = state.fock_prob([1])
            log = {'log_value':1.}
            return circuit_output, log

        new_hyperparams = deepcopy(self.hyperparams)
        new_hyperparams['circuit'] = circuit2
        new_hyperparams['log_every'] = 2
        learner = Learner_num(hyperparams=new_hyperparams)
        learner.train_circuit(steps=10)
        final_par = np.genfromtxt("logsNUM/log.csv", delimiter=',', skip_header=1)
        self.assertAlmostEqual(final_par[0, 5], 1)
        learner.run_circuit()
        learner.score_circuit()

    def test_optimization_with_printlog_SGD_regul_num(self):
        """
        Test to optimize a StrawberryFields Dgate with printing the log.
        """
        def circuit2(params):
            eng, q = sf.Engine(1)
            with eng:
                Dgate(params[0]) | q[0]
            state = eng.run('gaussian')
            circuit_output = state.fock_prob([1])
            log = {'log_value':1.}
            return circuit_output, log

        def myregularizer(regularized_params):
            return 0

        capturedOutput = StringIO()
        sys.stdout = capturedOutput
        new_hyperparams = deepcopy(self.hyperparams)
        new_hyperparams['circuit'] = circuit2
        new_hyperparams['print_log'] = True
        new_hyperparams['regularizer'] = myregularizer
        learner = Learner_num(hyperparams=new_hyperparams)
        learner.train_circuit(steps=1)
        sys.stdout = sys.__stdout__
        self.assertStringNotEmpty(capturedOutput.getvalue())

    def test_optimization_with_printlog_CG_num(self):
        """
        Test to optimize a StrawberryFields Dgate with printing the log.
        """
        def circuit2(params):
            eng, q = sf.Engine(1)
            with eng:
                Dgate(params[0]) | q[0]
            state = eng.run('gaussian')
            circuit_output = state.fock_prob([1])
            log = {'log_value':1.}
            return circuit_output, log

        capturedOutput = StringIO()
        sys.stdout = capturedOutput
        new_hyperparams = deepcopy(self.hyperparams)
        new_hyperparams['circuit'] = circuit2
        new_hyperparams['optimizer'] = 'CG'
        new_hyperparams['print_log'] = True
        learner = Learner_num(hyperparams=new_hyperparams)
        learner.train_circuit(steps=1)
        sys.stdout = sys.__stdout__
        self.assertStringNotEmpty(capturedOutput.getvalue())

    def test_optimization_SGD_with_monitoring_num(self):
        """
        Test to optimize a StrawberryFields Dgate while saving a log of the variable using the SGD optimizer.
        """
        new_hyperparams = deepcopy(self.hyperparams)
        new_hyperparams['init_circuit_params'] = [make_param_num(constant=0.1,
                                                                 name='alpha',
                                                                 monitor=True)]

        new_hyperparams['log_every'] = 2
        learner = Learner_num(hyperparams=new_hyperparams)
        learner.train_circuit(steps=10)
        final_par = np.genfromtxt("logsNUM/log.csv", delimiter=',', skip_header=1)
        self.assertAlmostEqual(final_par[0, 5], 0.1)

    def test_optimization_CG_with_monitoring_num(self):
        """
        Test to optimize a StrawberryFields Dgate while saving a log of the variable using the CG optimizer.
        """
        new_hyperparams = deepcopy(self.hyperparams)
        new_hyperparams['init_circuit_params'] = [make_param_num(constant=0.1,
                                                                 name='alpha',
                                                                 monitor=True)]

        new_hyperparams['log_every'] = 2
        new_hyperparams['optimizer'] = 'CG'
        learner = Learner_num(hyperparams=new_hyperparams)
        learner.train_circuit(steps=10)
        final_par = np.genfromtxt("logsNUM/log.csv", delimiter=',', skip_header=1)
        self.assertAlmostEqual(final_par[0, 5], 0.1)

    def test_optimization_score_num(self):
        """
        Test to optimize a StrawberryFields Dgate and score the circuit.
        """
        learner = Learner_num(hyperparams=self.hyperparams)
        learner.train_circuit(steps=10)
        test_score = learner.score_circuit()
        res = test_score['loss']
        self.assertAlmostEqual(res, -0.01450662909, delta=0.0001)
        with self.assertRaises(ValueError):
            learner.score_circuit(X=self.dummy_input)

    def test_optimization_run_num(self):
        """
        Test to optimize a StrawberryFields Dgate and run the circuit.
        """
        learner = Learner_num(hyperparams=self.hyperparams)
        learner.train_circuit(steps=10)
        test_score = learner.run_circuit()
        res = test_score['outputs']
        self.assertAlmostEqual(res, 0.014506629095, delta=0.0001)
        with self.assertRaises(ValueError):
            learner.run_circuit(X=self.dummy_input)

    def test_optimization_with_printing_params_num(self):
        """
        Test the get_circuit_parameters function printing.
        """

        capturedOutput = StringIO()
        sys.stdout = capturedOutput
        learner = Learner_num(hyperparams=self.hyperparams)
        learner.train_circuit(steps=1)
        learner.get_circuit_parameters(only_print=True)
        sys.stdout = sys.__stdout__
        self.assertStringNotEmpty(capturedOutput.getvalue())

class TestNumericalLearnerUnsupervised(BaseLearnerTest):
    """
    Test the numeric learner for unsupervised learning tasks
    """

    def setUp(self):
        super().setUp()

        my_init_params = [make_param_num(name='alpha', constant=1.)]

        def circuit(params):
            eng, q = sf.Engine(2)
            with eng:
                Dgate(params[0]) | q[0]
            state = eng.run('fock', cutoff_dim=5)
            circuit_output = state.all_fock_probs()
            return circuit_output

        def myloss(circuit_output, X):
            probs = [circuit_output[x[0], x[1]] for x in X]
            prob_total = sum(np.reshape(probs, -1))
            return -prob_total

        self.X_train = np.array([[0, 0]])

        self.hyperparams = {'circuit': circuit,
                            'init_circuit_params': my_init_params,
                            'task': 'unsupervised',
                            'optimizer': 'CG',
                            'loss': myloss,
                            'print_log': False
                            }

    def test_basic_unsupervised_num(self):
        """
        Basic test to train a variational circuit in unsupervised fashion.
        """
        learner = Learner_num(hyperparams=self.hyperparams)
        learner.train_circuit(X=self.X_train, steps=2)
        final_par = learner.get_circuit_parameters()
        final_par = final_par['alpha']
        self.assertAlmostEqual(final_par, 0., delta=0.0001)

    def test_basic_unsupervised_SGD_num(self):
        """
        Basic test to train a variational circuit in unsupervised fashion with SGD optimizer.
        """
        new_hyperparams = deepcopy(self.hyperparams)
        new_hyperparams['optimizer'] = 'SGD'
        learner = Learner_num(hyperparams=new_hyperparams)
        learner.train_circuit(X=self.X_train, steps=2)
        final_par = learner.get_circuit_parameters()
        final_par = final_par['alpha']
        self.assertAlmostEqual(final_par, 0.98523, delta=0.01)

    def test_unsupervised_score_num(self):
        """
        Test to to train a variational circuit in unsupervised fashion
        and score the circuit.
        """
        learner = Learner_num(hyperparams=self.hyperparams)
        learner.train_circuit(X=self.X_train, steps=2)
        test_score = learner.score_circuit(X=self.X_train)
        res = test_score['loss']
        self.assertAlmostEqual(res, -1., delta=0.0001)
        with self.assertRaises(ValueError):
            learner.score_circuit()

    def test_unsupervised_run_num(self):
        """
        Test to to train a variational circuit in unsupervised fashion
        and run the circuit.
        """
        learner = Learner_num(hyperparams=self.hyperparams)
        learner.train_circuit(X=self.X_train, steps=30)
        outp = learner.run_circuit()['outputs']
        self.assertAlmostEqualArray(outp[0, 0], 1.)
        with self.assertRaises(ValueError):
            learner.run_circuit(X=self.X_train)

    def test_unsupervised_with_logging_num(self):
        """
        Test to to train a variational circuit in unsupervised fashion
         while saving a log.
        """

        def circuit2(params):
            eng, q = sf.Engine(2)
            with eng:
                Dgate(params[0]) | q[0]
            state = eng.run('fock', cutoff_dim=5)
            circuit_output = state.all_fock_probs()
            log = {'log_value': 1.}
            return circuit_output, log

        new_hyperparams = deepcopy(self.hyperparams)
        new_hyperparams['circuit'] = circuit2
        new_hyperparams['log_every'] = 2
        learner = Learner_num(hyperparams=new_hyperparams)
        learner.train_circuit(X=self.X_train, steps=10)
        final_par = np.genfromtxt("logsNUM/log.csv", delimiter=',', skip_header=1)
        self.assertAlmostEqual(final_par[0, 5], 1)
        learner.run_circuit()
        learner.score_circuit(X=self.X_train)

    def test_unsupervised_batch_num(self):
        """
        Basic test to train a variational circuit in supervised fashion
        with batched SGD.
        """
        learner = Learner_num(hyperparams=self.hyperparams)
        learner.train_circuit(X=self.X_train, steps=2, batch_size=1, seed=1)
        final_par = learner.get_circuit_parameters()
        final_par = final_par['alpha']
        self.assertAlmostEqual(final_par, 0., delta=0.0001)

class TestNumericalLearnerSupervised(BaseLearnerTest):
    """
    Test the numeric learner for supervised learning tasks.
    """

    def setUp(self):
        super().setUp()

        my_init_params = [make_param_num(name='alpha', constant=2.)]

        def circuit(X, params):
            eng, q = sf.Engine(2)

            def single_input_circuit(x):
                eng.reset()
                with eng:
                    Dgate(x[0], 0.) | q[0]
                    Dgate(x[1], 0.) | q[1]
                    BSgate(phi=params[0]) | (q[0], q[1])
                    BSgate() | (q[0], q[1])
                state = eng.run('fock', cutoff_dim=10, eval=True)
                p0 = state.fock_prob([0, 2])
                p1 = state.fock_prob([2, 0])
                normalisation = p0 + p1 + 1e-10
                outp = p1 / normalisation
                return outp

            circuit_output = [single_input_circuit(x) for x in X]
            return circuit_output

        def myloss(circuit_output, targets):
            diff = circuit_output - targets
            return 0.5 * sum(np.dot(d, d) for d in diff)

        self.X_train = np.array([[0.2, 0.4], [0.6, 0.8], [0.4, 0.2], [0.8, 0.6]])
        self.Y_train = np.array([1., 1., 0., 0.])
        self.X_test = np.array([[0.25, 0.5], [0.5, 0.25]])
        self.Y_test = np.array([1., 0.])
        self.X_pred = np.array([[0.4, 0.5], [0.5, 0.4]])

        def outputs_to_predictions(circuit_output):
            return round(circuit_output)

        self.otp = outputs_to_predictions

        self.hyperparams = {'circuit': circuit,
                            'init_circuit_params': my_init_params,
                            'task': 'supervised',
                            'loss': myloss,
                            'init_learning_rate': 0.5,
                            'print_log': False
                            }

    def test_basic_supervised_num(self):
        """
        Basic test to to train a variational circuit in supervised fashion
        and score the result.
        """
        learner = Learner_num(hyperparams=self.hyperparams)
        learner.train_circuit(X=self.X_train, Y=self.Y_train, steps=30)
        final_par = learner.get_circuit_parameters()['alpha']
        self.assertAlmostEqual(final_par, 0.1709563877, delta=0.0001)

    def test_supervised_score_num(self):
        """
        Test to to train a variational circuit in supervised fashion
        and score the result.
        """
        learner = Learner_num(hyperparams=self.hyperparams)
        learner.train_circuit(X=self.X_train, Y=self.Y_train, steps=40)
        test_score = learner.score_circuit(X=self.X_test, Y=self.Y_test,
                                           outputs_to_predictions=self.otp)
        self.assertAlmostEqual(test_score['accuracy'], 1., delta=0.0001)
        self.assertAlmostEqual(test_score['loss'], 0., delta=0.1)
        with self.assertRaises(ValueError):
            learner.score_circuit(X=None, Y=None, outputs_to_predictions=self.otp)

    def test_supervised_run_num(self):
        """
        Test to to train a variational circuit in supervised fashion
        and predict the result.
        """
        learner = Learner_num(hyperparams=self.hyperparams)
        learner.train_circuit(X=self.X_train, Y=self.Y_train, steps=30)
        outcomes = learner.run_circuit(X=self.X_test, outputs_to_predictions=self.otp)
        self.assertAlmostEqualArray(outcomes['predictions'], [1., 0.])
        with self.assertRaises(ValueError):
            learner.run_circuit(X=None, outputs_to_predictions=self.otp)

    def test_supervised_with_logging_num(self):
        """
        Test to to train a variational circuit in supervised fashion
         while saving a log.
        """

        def circuit2(X, params):
            eng, q = sf.Engine(2)

            def single_input_circuit(x):
                eng.reset()
                with eng:
                    Dgate(x[0], 0.) | q[0]
                    Dgate(x[1], 0.) | q[1]
                    BSgate(phi=params[0]) | (q[0], q[1])
                    BSgate() | (q[0], q[1])
                state = eng.run('fock', cutoff_dim=10, eval=True)
                p0 = state.fock_prob([0, 2])
                p1 = state.fock_prob([2, 0])
                normalisation = p0 + p1 + 1e-10
                outp = p1 / normalisation
                return outp

            circuit_output = [single_input_circuit(x) for x in X]
            log = {'log_value': 1.}
            return circuit_output, log

        new_hyperparams = deepcopy(self.hyperparams)
        new_hyperparams['circuit'] = circuit2
        new_hyperparams['log_every'] = 2
        learner = Learner_num(hyperparams=new_hyperparams)
        learner.train_circuit(X=self.X_train, Y=self.Y_train, steps=10)
        final_par = np.genfromtxt("logsNUM/log.csv", delimiter=',', skip_header=1)
        self.assertAlmostEqual(final_par[0, 5], 1)
        learner.run_circuit(X=self.X_test)
        learner.score_circuit(X=self.X_test, Y=self.Y_test)

    def test_supervised_batch_num(self):
        """
        Basic test to train a variational circuit in supervised fashion
        with batched SGD.
        """
        learner = Learner_num(hyperparams=self.hyperparams)
        learner.train_circuit(X=self.X_train, Y=self.Y_train, steps=30, batch_size=2)
        final_par = learner.get_circuit_parameters()['alpha']
        self.assertAlmostEqual(final_par, 1.74412942216, delta=0.2)

class TestTfLearnerOptimization(BaseLearnerTest):
    """
    Test the TensorFlow learner for optimization tasks
    """

    def setUp(self):
        super().setUp()

        def circuit():
            alpha = make_param_tf(name='alpha', constant=0.1)
            eng, q = sf.Engine(1)
            with eng:
                Dgate(alpha) | q[0]
            state = eng.run('tf', cutoff_dim=7, eval=False)
            prob = state.fock_prob([1])
            circuit_output = tf.identity(prob, name="prob")
            return circuit_output

        def myloss(circuit_output):
            return -circuit_output

        self.hyperparams = {'circuit': circuit,
                            'task': 'optimization',
                            'loss': myloss,
                            'print_log': False
                            }
        self.dummy_input = np.array([[0., 1.]])

    def test_basic_optimization_tf(self):
        """
        Test to optimize a StrawberryFields Dgate with minimal parameters.
        """
        learner = Learner_tf(hyperparams=self.hyperparams)
        learner.train_circuit(steps=10)
        final_par = learner.get_circuit_parameters()['alpha']
        self.assertAlmostEqual(final_par, 0.1213333101, delta=0.0001)

    def test_optimization_with_regul_tf(self):
        """
        Test to optimize a StrawberryFields Dgate with regularization.
        """
        def myregularizer(regularized_params):
            return sum(regularized_params)

        new_hyperparams = deepcopy(self.hyperparams)
        new_hyperparams['regularizer'] = myregularizer
        new_hyperparams['regularization_strength'] = 1.
        learner = Learner_tf(hyperparams=new_hyperparams)
        learner.train_circuit(steps=10)
        final_par = learner.get_circuit_parameters()['alpha']
        self.assertAlmostEqual(final_par, 0.1213333101, delta=0.0001)

    def test_optimization_with_learningrate_tf(self):
        """
        Test to optimize a StrawberryFields Dgate with a custom learning rate.
        """
        new_hyperparams = deepcopy(self.hyperparams)
        new_hyperparams['init_learning_rate'] = 2.
        learner = Learner_tf(hyperparams=new_hyperparams)
        learner.train_circuit(steps=10)
        final_par = learner.get_circuit_parameters()['alpha']
        self.assertAlmostEqual(final_par, 1.449175616872, delta=0.0001)

    def test_optimization_with_decay_tf(self):
        """
        Test to optimize a StrawberryFields Dgate with learning rate decay.
        """
        new_hyperparams = deepcopy(self.hyperparams)
        new_hyperparams['decay'] = 0.5
        learner = Learner_tf(hyperparams=new_hyperparams)
        learner.train_circuit(steps=10)
        final_par = learner.get_circuit_parameters()['alpha']
        self.assertAlmostEqual(final_par, 0.117403194, delta=0.0001)

    def test_optimization_with_warmstart_tf(self):
        """
        Test to optimize a StrawberryFields Dgate with warm start.
        """
        learner = Learner_tf(hyperparams=self.hyperparams)
        learner.train_circuit(steps=10)
        new_hyperparams = deepcopy(self.hyperparams)
        new_hyperparams['warm_start'] = True
        learner2 = Learner_tf(hyperparams=new_hyperparams)
        learner2.train_circuit(steps=1)
        final_par = learner.get_circuit_parameters()['global_step']
        self.assertAlmostEqual(final_par, 11)

    def test_optimization_with_logging_not_throwing_errors_tf(self):
        """
        Test to optimize a StrawberryFields Dgate while saving a log.
        """
        def circuit2():
            alpha = make_param_tf(name='alpha', constant=0.1, monitor=True)
            eng, q = sf.Engine(1)
            with eng:
                Dgate(alpha) | q[0]
            state = eng.run('tf', cutoff_dim=7, eval=False)
            prob = state.fock_prob([1])
            circuit_output = tf.identity(prob, name="prob")
            return circuit_output

        new_hyperparams = deepcopy(self.hyperparams)
        new_hyperparams['print_log'] = True
        new_hyperparams['log_every'] = 2
        new_hyperparams['circuit'] = circuit2
        learner = Learner_tf(hyperparams=new_hyperparams)
        learner.train_circuit(steps=4, tensors_to_log={'Prob': 'prob'})

    def test_optimization_with_monitoring_tf(self):
        """
        Test to optimize a StrawberryFields Dgate while saving a log of the variable.
        """

        def circuit2():
            alpha = make_param_tf(name='alpha', constant=0.1, monitor=True)
            eng, q = sf.Engine(1)
            with eng:
                Dgate(alpha) | q[0]
            state = eng.run('tf', cutoff_dim=7, eval=False)
            return state.fock_prob([1])

        new_hyperparams = deepcopy(self.hyperparams)
        new_hyperparams['circuit'] = circuit2
        new_hyperparams['log_every'] = 2
        learner = Learner_tf(hyperparams=new_hyperparams)
        learner.train_circuit(steps=10)

    def test_optimization_score_tf(self):
        """
        Test to optimize a StrawberryFields Dgate and score the circuit.
        """
        learner = Learner_tf(hyperparams=self.hyperparams)
        learner.train_circuit(steps=10)
        test_score = learner.score_circuit()
        res = test_score['loss']
        self.assertAlmostEqual(res, -0.01450662909, delta=0.0001)
        with self.assertRaises(ValueError):
            learner.score_circuit(X=self.dummy_input)

    def test_optimization_run_tf(self):
        """
        Test to optimize a StrawberryFields Dgate and run the circuit.
        """
        learner = Learner_tf(hyperparams=self.hyperparams)
        learner.train_circuit(steps=10)
        test_score = learner.run_circuit()
        res = test_score['outputs']
        self.assertAlmostEqual(res, 0.014506629095, delta=0.0001)
        with self.assertRaises(ValueError):
            learner.run_circuit(X=self.dummy_input)

    def test_optimization_with_printlog_tf(self):
        """
        Test to optimize a StrawberryFields Dgate with printing the log.
        """

        capturedOutput = StringIO()
        sys.stdout = capturedOutput
        new_hyperparams = deepcopy(self.hyperparams)
        new_hyperparams['print_log'] = True
        learner = Learner_tf(hyperparams=new_hyperparams)
        learner.train_circuit(steps=1)
        sys.stdout = sys.__stdout__
        self.assertStringNotEmpty(capturedOutput.getvalue())

    def test_optimization_with_printing_params_tf(self):
        """
        Test the get_circuit_parameters function prints.
        """

        capturedOutput = StringIO()
        sys.stdout = capturedOutput
        learner = Learner_tf(hyperparams=self.hyperparams)
        learner.train_circuit(steps=1)
        learner.get_circuit_parameters(only_print=True)
        sys.stdout = sys.__stdout__
        self.assertStringNotEmpty(capturedOutput.getvalue())

class TestTfLearnerUnsupervised(BaseLearnerTest):
    """
    Test the TensorFLow learner for unsupervised learning tasks
    """

    def setUp(self):
        super().setUp()

        def circuit():
            alpha = make_param_tf(name='alpha', constant=1.)
            eng, q = sf.Engine(2)
            with eng:
                Dgate(alpha) | q[0]
            state = eng.run('tf', cutoff_dim=5, eval=False)
            circuit_output = state.all_fock_probs()
            return circuit_output

        def myloss(circuit_output, X):
            probs = tf.gather_nd(params=circuit_output, indices=X)
            prob_total = tf.reduce_sum(probs, axis=0)
            return -prob_total

        self.X_train = np.array([[0, 0]])

        self.hyperparams = {'circuit': circuit,
                            'task': 'unsupervised',
                            'optimizer': 'SGD',
                            'loss': myloss,
                            'print_log': False
                            }

    def test_ftrl_prints_warning_tf(self):
        """
        Basic test to check that using the FTRL optimizer prints a warning.
        """
        learner = Learner_tf(hyperparams=self.hyperparams)
        learner.train_circuit(X=self.X_train, steps=30)
        final_par = learner.get_circuit_parameters()['alpha']
        self.assertAlmostEqual(final_par, 0.75931054, delta=0.01)

    def test_unsupervised_score_tf(self):
        """
        Test to to train a variational circuit in unsupervised fashion
        and score the circuit.
        """
        learner = Learner_tf(hyperparams=self.hyperparams)
        learner.train_circuit(X=self.X_train, steps=30)
        test_score = learner.score_circuit(X=self.X_train)
        res = test_score['loss']
        self.assertAlmostEqual(res, -0.561831, delta=0.01)
        with self.assertRaises(ValueError):
            learner.score_circuit(X=None)

    def test_unsupervised_run_tf(self):
        """
        Test to to train a variational circuit in unsupervised fashion
        and run the circuit.
        """
        learner = Learner_tf(hyperparams=self.hyperparams)
        learner.train_circuit(X=self.X_train, steps=30)
        outp = learner.run_circuit()['outputs']
        self.assertAlmostEqual(outp[0, 0], 0.56183195, delta=0.01)
        with self.assertRaises(ValueError):
            learner.run_circuit(X=self.X_train)

    def test_unsupervised_with_logging_tf(self):
        """
        Test to to train a variational circuit in unsupervised fashion
         while saving a log.
        """
        new_hyperparams = deepcopy(self.hyperparams)
        new_hyperparams['log_every'] = 2
        learner = Learner_tf(hyperparams=new_hyperparams)
        learner.train_circuit(X=self.X_train, steps=30)

    def test_unsupervised_batch_tf(self):
        """
        Basic test to train a variational circuit in supervised fashion
        with batched SGD.
        """
        learner = Learner_tf(hyperparams=self.hyperparams)
        learner.train_circuit(X=self.X_train, steps=30, batch_size=1, shuffle_data=False)
        final_par = learner.get_circuit_parameters()
        final_par = final_par['alpha']
        self.assertAlmostEqual(final_par, 0.75931054, delta=0.01)

class TestTfLearnerSupervised(BaseLearnerTest):
    """
    Test the TensorFLow learner for supervised learning tasks
    """

    def setUp(self):
        super().setUp()

        def circuit(X):
            alpha = make_param_tf('alpha', constant=2.)
            eng, q = sf.Engine(2)
            with eng:
                Dgate(X[:, 0], 0.) | q[0]
                Dgate(X[:, 1], 0.) | q[1]
                BSgate(phi=alpha) | (q[0], q[1])
                BSgate() | (q[0], q[1])
            num_inputs = X.get_shape().as_list()[0]
            state = eng.run('tf', cutoff_dim=10, eval=False, batch_size=num_inputs)
            p0 = state.fock_prob([0, 2])
            p1 = state.fock_prob([2, 0])
            normalisation = p0 + p1 + 1e-10
            circuit_output = p1 / normalisation
            return circuit_output

        def myloss(circuit_output, targets):
            return tf.losses.mean_squared_error(labels=circuit_output, predictions=targets)

        self.X_train = np.array([[0.2, 0.4], [0.6, 0.8], [0.4, 0.2], [0.8, 0.6]])
        self.Y_train = np.array([1., 1., 0., 0.])
        self.X_test = np.array([[0.25, 0.5], [0.5, 0.25]])
        self.Y_test = np.array([1., 0.])
        self.X_pred = np.array([[0.4, 0.5], [0.5, 0.4]])

        def outputs_to_predictions(circuit_output):
            return tf.round(circuit_output)

        self.otp = outputs_to_predictions

        self.hyperparams = {'circuit': circuit,
                            'task': 'supervised',
                            'loss': myloss,
                            'init_learning_rate': 0.5,
                            'print_log': False
                            }

    def test_basic_supervised_tf(self):
        """
        Basic test to to train a variational circuit in supervised fashion
        and score the result.
        """
        learner = Learner_tf(hyperparams=self.hyperparams)
        learner.train_circuit(X=self.X_train, Y=self.Y_train, steps=30)
        final_par = learner.get_circuit_parameters()['alpha']
        self.assertAlmostEqual(final_par, 1.7205561, delta=0.01)

    def test_supervised_score_tf(self):
        """
        Test to to train a variational circuit in supervised fashion
        and score the result.
        """
        learner = Learner_tf(hyperparams=self.hyperparams)
        learner.train_circuit(X=self.X_train, Y=self.Y_train, steps=60)
        test_score = learner.score_circuit(X=self.X_test, Y=self.Y_test,
                                           outputs_to_predictions=self.otp)
        self.assertAlmostEqual(test_score['accuracy'], 1., delta=0.0001)
        self.assertAlmostEqual(test_score['loss'], 0., delta=0.1)
        with self.assertRaises(ValueError):
            learner.score_circuit(X=None, Y=None, outputs_to_predictions=self.otp)

    def test_supervised_run_tf(self):
        """
        Test to to train a variational circuit in supervised fashion
        and predict the result.
        """
        learner = Learner_tf(hyperparams=self.hyperparams)
        learner.train_circuit(X=self.X_train, Y=self.Y_train, steps=60)
        outcomes = learner.run_circuit(X=self.X_test, outputs_to_predictions=self.otp)
        self.assertAlmostEqualArray(outcomes['predictions'], [1.,0.])
        with self.assertRaises(ValueError):
            learner.run_circuit(X=None, outputs_to_predictions=self.otp)

    def test_supervised_with_logging_tf(self):
        """
        Test to to train a variational circuit in supervised fashion
         while saving a log.
        """
        new_hyperparams = deepcopy(self.hyperparams)
        new_hyperparams['log_every'] = 2
        learner = Learner_tf(hyperparams=new_hyperparams)
        learner.train_circuit(X=self.X_train, Y=self.Y_train, steps=10)

    def test_supervised_batch_tf(self):
        """
        Basic test to train a variational circuit in supervised fashion
        with batched SGD.
        """
        learner = Learner_tf(hyperparams=self.hyperparams)
        learner.train_circuit(X=self.X_train, Y=self.Y_train, steps=30, batch_size=2)
        final_par = learner.get_circuit_parameters()['alpha']
        self.assertAlmostEqual(final_par, 1.74412942216, delta=0.2)

class TestInputChecks(BaseLearnerTest):
    """
    Test input checks.
    """

    def setUp(self):
        super().setUp()
        self.onedarr = np.array([1., 0., 2.])
        self.onedlis = [1., 0.]
        self.twodarr = np.array([[1., 0.], [0., 1.]])
        self.twodlis = [[1., 0.], [0., 1.]]
        self.threedarr = np.array([[[0., 1.], [1., 2.]], [[0., 1.], [1., 2.]]])
        self.integer = 1
        self.neg = -1
        self.flt = 0.1
        self.string = "bla"
        self.dict = {'a a': 1.}

        def regul(regularized_params):
            return 0
        self.regul = regul

    def test_check_hp_throws_exception(self):
        """
        Test the exceptions of the hyperparameter check for both the TensorFlow and numerical learner.
        """
        num_check({'print_log': False, 'circuit': 1, 'loss': 1, 'task': 'optimization', 'init_circuit_params': 1})
        # No task key
        with self.assertRaises(ValueError):
            num_check({'circuit': 1, 'loss': 1, 'init_circuit_params': [{}]})
        with self.assertRaises(ValueError):
            tf_check({'circuit': 1, 'loss': 1})
        # No loss key
        with self.assertRaises(ValueError):
            num_check({'circuit': 1, 'task': 'optimization', 'init_circuit_params': [{}]})
        with self.assertRaises(ValueError):
            tf_check({'circuit': 1,  'task': 'optimization'})
        # No circuit key
        with self.assertRaises(ValueError):
            num_check({'loss': 1, 'task': 'optimization', 'init_circuit_params': [{}]})
        with self.assertRaises(ValueError):
            tf_check({'loss': 1, 'task': 'optimization'})
        # No init param key
        with self.assertRaises(ValueError):
            num_check({'circuit': 1, 'loss': 1, 'task': 'optimization'})
        # Wrong task
        with self.assertRaises(ValueError):
            num_check({'circuit': 1, 'loss': 1, 'task': self.string, 'init_circuit_params': 1.1})
        with self.assertRaises(ValueError):
            tf_check({'circuit': 1, 'loss': 1, 'task': self.string, 'optimizer': 'sthfunny'})
        # Wrong init params
        with self.assertRaises(ValueError):
            num_check({'circuit': 1, 'loss': 1, 'task': 'optimization', 'init_circuit_params': 1.1})
        # Unrecognized key
        with self.assertRaises(ValueError):
            num_check({'circuit': 1, 'loss': 1, 'task': 'optimization', 'init_circuit_params': [{}], 'funny_key': 1.})
        with self.assertRaises(ValueError):
            tf_check({'circuit': 1, 'loss': 1, 'task': 'optimization', 'funny_key': 1.})
        # Unrecognized optimizer
        with self.assertRaises(ValueError):
            num_check({'circuit': 1, 'loss': 1, 'task': 'optimization', 'init_circuit_params': [{}], 'optimizer': 'sthfunny'})
        with self.assertRaises(ValueError):
            tf_check({'circuit': 1, 'loss': 1, 'task': 'optimization', 'optimizer': 'sthfunny'})

    def test_check_hp_sets_default_regstrength(self):
        """
        Test that when regularizer is given but no regularization strength, the strength is set to 0.1, else it is
        set to user provided value.
        """
        hp = {'circuit': 1, 'loss': 1, 'task': 'optimization', 'init_circuit_params': [{}],
              'regularizer': self.regul}
        num_check(hp)
        self.assertAlmostEqual(hp['regularization_strength'], 0.1)

        hp = {'circuit': 1, 'loss': 1, 'task': 'optimization', 'regularizer': self.regul}
        tf_check(hp)
        self.assertAlmostEqual(hp['regularization_strength'], 0.1)

        hp = {'circuit': 1, 'loss': 1, 'task': 'optimization', 'init_circuit_params': [{}],
              'regularizer': self.regul, 'regularization_strength': 0.5}
        num_check(hp)
        self.assertAlmostEqual(hp['regularization_strength'], 0.5)

        hp = {'circuit': 1, 'loss': 1, 'task': 'optimization', 'regularizer': self.regul, 'regularization_strength': 0.5}
        tf_check(hp)
        self.assertAlmostEqual(hp['regularization_strength'], 0.5)

    def test_check_hp_renames_regularized_params(self):
        """
        Test check that the initial parameter name contains string <regularized> when regularized.
        """
        hp = {'circuit': 1, 'loss': 1, 'task': 'optimization',
              'init_circuit_params': [{'name': 'dummy', 'regul': True}]}
        num_check(hp)
        self.assertEquals(hp['init_circuit_params'][0]['name'], 'regularized/dummy')

    def test_check_hp_prints_warning_for_ftrl_optimizer(self):
        """
        Test Ftrl warning gets printed in hyperparameter check of tf learner.
        """
        captured_output = StringIO()
        sys.stdout = captured_output
        hp = {'circuit': 1, 'loss': 1, 'task': 'optimization', 'optimizer': 'Ftrl', 'print_log':False}
        tf_check(hp)
        sys.stdout = sys.__stdout__
        self.assertStringNotEmpty(captured_output.getvalue())

    def test_checkX_throws_exception(self):
        """
        Test the exceptions of check_X for both the TensorFlow and numerical learner.
        """
        num_check_X(self.twodlis)
        tf_check_X(self.twodlis)

        with self.assertRaises(ValueError):
            num_check_X(self.dict)
        with self.assertRaises(ValueError):
            tf_check_X(self.dict)
        with self.assertRaises(ValueError):
            num_check_X(self.onedarr)
        with self.assertRaises(ValueError):
            tf_check_X(self.onedarr)
        with self.assertRaises(ValueError):
            num_check_X(self.onedlis)
        with self.assertRaises(ValueError):
            tf_check_X(self.onedlis)

    def test_checkY_throws_exception(self):
        """
        Test the exceptions of check_Y for both the TensorFlow and numerical learner.
        """
        with self.assertRaises(ValueError):
            num_check_Y(self.onedarr, self.twodarr)
        with self.assertRaises(ValueError):
            tf_check_Y(self.onedarr, self.twodarr)
        with self.assertRaises(ValueError):
            num_check_Y(self.twodarr, None)
        with self.assertRaises(ValueError):
            tf_check_Y(self.twodarr, None)
        with self.assertRaises(ValueError):
            num_check_Y(self.dict, self.integer)
        with self.assertRaises(ValueError):
            tf_check_Y(self.dict, self.integer)

    def test_checkshuffle_throws_exception(self):
        """
        Test the exceptions of check_shuffle for the TensorFlow learner.
        """
        with self.assertRaises(ValueError):
            tf_check_shuffle(self.integer)

    def test_checkbatches_throws_exception(self):
        """
        Test the exceptions of check_batch_size for both the TensorFlow and numerical learner.
        """
        num_check_batch_size(None, self.twodarr)
        tf_check_batch_size(None, self.twodarr)
        with self.assertRaises(ValueError):
            num_check_batch_size(self.dict, self.twodarr)
        with self.assertRaises(ValueError):
            tf_check_batch_size(self.dict, self.twodarr)
        with self.assertRaises(ValueError):
            num_check_batch_size(5, self.twodarr)
        with self.assertRaises(ValueError):
            tf_check_batch_size(5, self.twodarr)
        with self.assertRaises(ValueError):
            num_check_batch_size(self.neg, self.twodarr)
        with self.assertRaises(ValueError):
            tf_check_batch_size(self.neg, self.twodarr)

    def test_checksteps_throws_exception(self):
        """
        Test the exceptions of check_steps for both the TensorFlow and numerical learner.
        """
        num_check_steps(None)
        tf_check_steps(None)
        with self.assertRaises(ValueError):
            num_check_steps(self.dict)
        with self.assertRaises(ValueError):
            tf_check_steps(self.dict)
        with self.assertRaises(ValueError):
            num_check_X(self.neg)
        with self.assertRaises(ValueError):
            tf_check_X(self.neg)

    def test_check_logs_throws_exception(self):
        """
        Test the exceptions of check_logs for the numerical learner.
        """
        with self.assertRaises(ValueError):
            num_check_logs(self.integer)
        with self.assertRaises(ValueError):
            num_check_logs(self.dict)

if __name__ == "__main__":
    print('Testing learners of the QMLT app.')

    # run the tests in this file
    suite = unittest.TestSuite()
    for t in [TestInputChecks,
              TestNumericalLearnerOptimization,
              TestNumericalLearnerUnsupervised,
              TestNumericalLearnerSupervised,
              TestTfLearnerOptimization,
              TestTfLearnerUnsupervised,
              TestTfLearnerSupervised,
             ]:
        ttt = unittest.TestLoader().loadTestsFromTestCase(t)
        suite.addTests(ttt)

    unittest.TextTestRunner().run(suite)

max_stars_repo_name: stjordanis/QMLT

max_issues_repo_name: stjordanis/QMLT

max_forks_repo_name: stjordanis/QMLT


text:
# import packages

import json
import numpy as np
import os

from landscape import Landscape

# import settings

with open('../settings.json') as f:
    settings = json.load(f)

# extract general settings

directoryLandscapes = settings['directory']['landscapes']

gridN = settings['grid']['n']
gridSize = settings['grid']['size']

# extract variable settings for all continuous variables

varLabel = list()
varSmooth1 = list()
varSmooth2 = list()

for var in settings['var']:
    varLabel.append(settings['var'][var]['label'])
    varSmooth1.append(settings['var'][var]['smooth1'])
    varSmooth2.append(settings['var'][var]['smooth2'])

# read settings for binary variable (used for completely hostile sites)

unsuitableSmooth1 = settings['unsuitable']['smooth1']
unsuitableSmooth2 = settings['unsuitable']['smooth2']
unsuitableThreshold = settings['unsuitable']['threshold']

# main program

# create new directory to store results if it didn't already exist

if not os.path.exists(directoryLandscapes):
    os.makedirs(directoryLandscapes)

# loop over number of grids that are to be generated

for grid in range(0, gridN):

    # print message to keep track of the progress

    print("Generating grid " + str(grid))

    # generate a random seed, use it and store it

    randomSeed = round(np.random.random() * 1000000)
    np.random.seed(randomSeed)
    f = open(directoryLandscapes + 'randomSeeds.txt', 'a')
    f.write(str(grid) + "," + str(randomSeed) + '\n')
    f.close()

    # generate a landscape object with binary variable

    lscape = Landscape(size=gridSize, smooth1=unsuitableSmooth1, smooth2=unsuitableSmooth2, threshold=unsuitableThreshold)

    # simulate all continuous variables

    for index, label in enumerate(varLabel):
        lscape.sim_pred(label=label, smooth1=varSmooth1[index], smooth2=varSmooth2[index])

    # save the object

    lscape.pkl_export(path=directoryLandscapes + 'landscape' + str(grid) + '.pkl')

ext: py

max_stars_repo_name: KoMiel/virtualSpecies

max_issues_repo_name: KoMiel/virtualSpecies

max_forks_repo_name: KoMiel/virtualSpecies

reason: import numpy


text:
from geojson_to_pixel_arr import geojson_to_pixel_arr
from plot_truth_coords import plot_truth_coords
from create_building_mask import create_building_mask
from create_dist_map import create_dist_map
from plot_dist_transform import plot_dist_transform

import matplotlib.pyplot as plt
import numpy as np
from osgeo import gdal,ogr,osr

import keras
from keras.models import Sequential
from keras.layers import Dense, Dropout, Flatten, Activation
from keras.layers import Conv2D, MaxPooling2D
from keras.optimizers import RMSprop, Adam, Adadelta

import os
import glob
import time

def numpy_to_tif(arr, out_path, template):
    no_data = 0
    # First of all, gather some information from the template file
    data = gdal.Open(template)
    [cols, rows] = arr.shape
    trans = data.GetGeoTransform()
    proj = data.GetProjection()
    # nodatav = 0 #data.GetNoDataValue()
    # Create the file, using the information from the template file
    outdriver = gdal.GetDriverByName("GTiff")
    # http://www.gdal.org/gdal_8h.html
    # GDT_Byte = 1, GDT_UInt16 = 2, GDT_UInt32 = 4, GDT_Int32 = 5, GDT_Float32 = 6,
    outdata   = outdriver.Create(str(out_path), rows, cols, 1, gdal.GDT_Float32)
    # Write the array to the file, which is the original array in this example
    outdata.GetRasterBand(1).WriteArray(arr)
    # Set a no data value if required
    outdata.GetRasterBand(1).SetNoDataValue(no_data)
    # Georeference the image
    outdata.SetGeoTransform(trans)
    # Write projection information
    outdata.SetProjection(proj)
    return arr

def find_img_json(img_list, json_dir):
    json_list = []

    for img in img_list:
        img_name = os.path.splitext(os.path.basename(img))[0]
        json_name = img_name.replace('3band', 'Geo') + '.geojson'
        json_path = os.path.join(json_dir, json_name)
        if os.path.exists(json_path):
            json_list.append(json_path)
            # print(img)
            # print(json_path)
    return json_list

def create_dist_tifs(img_list, json_list, out_dir, out_npy, nrows, ncols):

    N_imgs = len(img_list)
    print('Total number of images: ', str(N_imgs))

    arrays = []
    for i, img in enumerate(img_list):
        # print(i)
        img_name = os.path.splitext(os.path.basename(img))[0]
        dist_npy_name = img_name.replace('3band', 'dist') + '.npy'
        dist_tif_name = img_name.replace('3band', 'dist') + '.tif'
        print(str(i), ": ", dist_tif_name)
        dist_npy_path = os.path.join(out_dir, dist_npy_name)
        dist_tif_path = os.path.join(out_dir, dist_tif_name)
        # print(json_list[i])
        # print(dist_npy_path)
        dist_arr = create_dist_map(img,json_list[i])
        dist_arr = dist_arr[0:nrows, 0:ncols]
        # dist_arr.shape = dist_arr.shape + (1,)
        arrays.append(dist_arr)
        # print(dist_arr.shape)
        # tmp = np.load(dist_npy_path)
        numpy_to_tif(dist_arr,dist_tif_path, img)
        dist_arr = None
        # print(tmp.shape)
        # plt.imshow(dist_arr)
        # plt.show()
    data = np.array(arrays)
    data.shape = data.shape + (1,)  # convert 2d array to 3d
    print(data.shape)
    np.save(out_npy, data)
    return  arrays

def images_to_numpy(input_dir, out_file, nrows, ncols, nsamples):
    arrays = []
    filelist = glob.glob(os.path.join(input_dir, '*.tif'))
    filelist = filelist[0:nsamples]
    print("Total number of images: ", len(filelist))
    for i, file in enumerate(filelist):
        print(str(i), ': ', file)
        img = plt.imread(file)[0:nrows, 0:ncols, :]
        # print(img.shape)
        arrays.append(img)
    data = np.array(arrays)
    np.save(out_file, data)
    print(data.shape)
    return data

def train_model(X_train, y_train, batch_size, epochs):

    nrows = train_x.shape[1]
    ncols = train_x.shape[2]
    nbands = train_x.shape[3]
    input_shape = (nrows, ncols, nbands)

    # batch_size = 32
    # # num_classes = 6
    # epochs = 12

    model = Sequential()

    # model.add(Dense(13, kernel_initializer='normal', activation='relu', input_shape=input_shape))
    # model.add(Dense(6, kernel_initializer='normal', activation='relu'))
    # model.add(Dense(1, kernel_initializer='normal'))
    # # Compile model
    # model.compile(loss='mean_squared_error', optimizer='adam')
    # model.summary()

    # model.add(Dense(units=64, activation='relu', input_shape=input_shape))
    # model.add(Dense(units=10, activation='relu'))
    # model.add(Flatten())
    # model.compile(loss='categorical_crossentropy',
    #               optimizer='sgd',
    #               metrics=['accuracy'])
    # model.compile(loss=keras.losses.categorical_crossentropy,
    #               optimizer=keras.optimizers.SGD(lr=0.01, momentum=0.9, nesterov=True))
    # model.summary()
    # model.fit(train_x, train_y, epochs=epochs, batch_size=batch_size)
    # model = Sequential()
    model.add(Conv2D(32, kernel_size=(3, 3), activation='relu', input_shape=input_shape))
    model.add(Conv2D(64, (3, 3), activation='relu'))

    model.add(Flatten())
    model.add(Dense(256))
    model.add(Activation('relu'))
    model.add(Dropout(0.5))
    model.add(Dense(128))
    model.add(Activation('relu'))
    model.add(Dropout(0.5))

    model.add(Dense(nrows*ncols))
    model.add(Activation('linear'))

    model.compile(loss='mean_squared_error', optimizer=Adadelta())
    model.summary()
    model.fit(X_train, y_train, batch_size, epochs)

    # model.add(MaxPooling2D(pool_size=(2, 2)))
    # model.add(Dropout(0.25))
    # model.add(Flatten())
    # model.add(Dense(128, activation='relu'))
    # model.add(Dropout(0.5))
    # # model.add(Dense(num_classes, activation='softmax'))
    # model.summary()
    # model.compile(loss=keras.losses.categorical_crossentropy,
    #               optimizer=keras.optimizers.Adadelta(),
    #               metrics=['accuracy'])
    # model.fit(train_x, train_y,
    #           batch_size=batch_size,
    #           epochs=epochs,
    #           verbose=1)
    #           # validation_data=(x_test, y_test))
    return 0

if __name__ == '__main__':

    start_time = time.time()

    data_dir = "/media/hdd/Data/AWS/AOI 1 - Rio de Janeiro/processedBuildingLabels/"
    json_dir = os.path.join(data_dir, 'vectordata/geojson/')

    out_dir = os.path.join(data_dir, "output")
    if not os.path.exists(out_dir):
        os.makedirs(out_dir)

    sample_dir = os.path.join(data_dir, "3band")
    images = glob.glob(os.path.join(sample_dir,"*.tif"))

    train_dir = os.path.join(data_dir, 'train')
    train_x_path = os.path.join(train_dir, "train_x.npy")
    train_y_path = os.path.join(train_dir, "train_y.npy")

    nrows = 406
    ncols = 438
    nsamples = 1000
    images = images[0:nsamples]
    # for img in images:
    #     print(img)
    json_files = glob.glob(os.path.join(sample_dir,"*.geojson"))
    json_files = find_img_json(images, json_dir)

    # train_x = images_to_numpy(sample_dir, train_x_path, nrows, ncols, nsamples)
    # create_dist_tifs(images, json_files, out_dir, train_y_path, nrows, ncols)

    train_x = np.load(train_x_path)
    train_y = np.load(train_y_path)

    print(train_x.shape)
    print(train_y.shape)

    batch_size = 50
    epochs = 5
    train_model(train_x, train_y, batch_size, epochs)

    # print(train_x.shape)
    # train_x = np.load(train_y_path)
    # print(train_x.shape)
    # print(train_x)
    # train_y = images_to_numpy(out_dir, train_y_path)

    # img = images[0]
    # geojson = json_files[0]

    # pixel_coords, latlon_coords = geojson_to_pixel_arr(img,geojson)
    #
    # for pixels in pixel_coords:
    #     print(pixels)
    #     print("\n")
    #
    # for latlon in latlon_coords:
    #     print(latlon)
    #     print("\n")
    # print(pixel_coords)
    # print(latlon_coords)
    #
    # input_image = plt.imread(img)
    # plot_truth_coords(input_image, pixel_coords)
    #
    # output_mask = "output/mask.tif"
    # create_building_mask(img,geojson,output_mask)
    #
    # mask = plt.imread(output_mask)
    # plt.imshow(mask, cmap='bwr')
    # plt.show()
    #
    # dist_map = "output/dist.npy"
    # create_dist_map(img, geojson, dist_map)
    # dist_image = np.load(dist_map)
    # plot_dist_transform(input_image, pixel_coords, dist_image)

    end_time = time.time()
    print("Total run time: ", str(end_time-start_time))

max_stars_repo_name: giswqs/SpaceNet

max_issues_repo_name: giswqs/SpaceNet

max_forks_repo_name: giswqs/SpaceNet


text:
from __future__ import print_function, division

import sys
sys._running_pytest = True

import pytest
from sympy.core.cache import clear_cache

def pytest_report_header(config):
    from sympy.utilities.misc import ARCH
    s = "architecture: %s\n" % ARCH
    from sympy.core.cache import USE_CACHE
    s += "cache:        %s\n" % USE_CACHE
    from sympy.core.compatibility import GROUND_TYPES, HAS_GMPY
    version = ''
    if GROUND_TYPES =='gmpy':
        if HAS_GMPY == 1:
            import gmpy
        elif HAS_GMPY == 2:
            import gmpy2 as gmpy
        version = gmpy.version()
    s += "ground types: %s %s\n" % (GROUND_TYPES, version)
    return s

def pytest_addoption(parser):
    parser.addoption("--slow", action="store_true",
        help="allow slow tests to run")

def pytest_configure(config):
    # register an additional marker
    config.addinivalue_line("markers", "slow: slow test")

def pytest_runtest_setup(item):
    if not isinstance(item, pytest.Function):
        return
    if item.config.getoption("--slow"):
        if not 'slow' in item.keywords:
            pytest.skip()
    elif 'slow' in item.keywords:
        pytest.skip("slow test: pass --slow to run")

def pytest_terminal_summary(terminalreporter):
    if (terminalreporter.stats.get('error', None) or
            terminalreporter.stats.get('failed', None)):
        terminalreporter.write_sep(
            ' ', 'DO *NOT* COMMIT!', red=True, bold=True)

def pytest_runtest_teardown():
    clear_cache()

reason: from sympy


text:
import numpy as np
import pandas as pd
from typing import Optional

def add_spray_angle(df: pd.DataFrame, adjusted: bool = False) -> pd.DataFrame:
    """Adds spray angle and adjusted spray angle to StatCast DataFrames
    - Spray angle is the raw left-right angle of the hit
    - Adjusted spray angle flips the sign for left handed batters, making it a push/pull angle

    Args:
        df (pd.DataFrame): StatCast pd.DataFrame (retrieved through statcast, statcast_batter, etc)
    Returns:
        pd.DataFrame: Input dataframe with spray angle columns appended
    """

    df["spray_angle"] = np.arctan((df["hc_x"]-125.42)/(198.27-df["hc_y"]))*180/np.pi*.75
    if adjusted:
        df["adj_spray_angle"] = df.apply(lambda row: -row["spray_angle"] if row["stand"] == "L" else row["spray_angle"], axis=1)
        df = df.drop(["spray_angle"], axis=1)
    return df


text:
import nltk
import numpy as np
import random
import string # to process standard python strings
from nltk import WordNetLemmatizer
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.metrics.pairwise import cosine_similarity

#Reading in the corpus
with open('corpus/chatbot.txt','r', encoding='utf8', errors ='ignore') as fin:
    raw = fin.read().lower()

#Tokenisation
sent_tokens = nltk.sent_tokenize(raw) #converts to list of sentences
word_tokens = nltk.word_tokenize(raw) #converts to list of words

# Preprocessing
lemma = WordNetLemmatizer()

def lem_tokens(tokens):
    return [lemma.lemmatize(token) for token in tokens]

remove_punct_dict = dict((ord(punct), None) for punct in string.punctuation)

def lem_normalizer(text):
    return lem_tokens(nltk.word_tokenize(text.lower().translate(remove_punct_dict)))

# Keyword Matching
GREETING_INPUTS = ('hi', 'hello', 'greeting', 'sup', 'hey', 'what\'s up')
GREETING_RESPONSES = ['hi', 'hey', '*nods*', 'hi there', 'hello', 'I am glad! Ypu are talking to me']

def bot_greeting(sentence):
    for word in sentence.split():
        if word.lower() in GREETING_INPUTS:
            return random.choice(GREETING_RESPONSES)

# Generating response
def response(user_response):
    bot_response = ''
    sent_tokens.append(user_response)
    frequency_vector = TfidfVectorizer(tokenizer=lem_normalizer, stop_words='english')
    tf_idf = frequency_vector.fit_transform(sent_tokens)
    vals = cosine_similarity(tf_idf[-1], tf_idf)
    idx = vals.argsort()[0][-2]
    flat = vals.flatten()
    flat.sort()
    req_tfidf = flat[-2]
    if(req_tfidf==0):
        bot_response = bot_response+'sorry, I can\'t seem to understand you'
        return bot_response
    else:
        bot_response = bot_response+sent_tokens[idx]
        return bot_response

flag=True
print('FejiBot: Hello! my name is Feji. I will answer your queries about chatbots.'
      'If you want to exit, type Bye!')

while(flag == True):
    user_response = input()
    user_response = user_response.lower()
    if(user_response != 'bye'):
        if(user_response is 'thanks' or user_response is 'thank you'):
            flag=False
            print('FejiBot: You are welcome..')
        else:
            if(bot_greeting(user_response) != None):
                print('FejiBot: '+bot_greeting(user_response))
            else:
                print('FejiBot: ', end = '')
                print(response(user_response))
                sent_tokens.remove(user_response)
    else:
        flag=False
        print('FejiBot: Bye! Nice chatting with you')


text:
# -*- coding: utf-8 -*-
"""
Tests.
"""
import unittest
import numpy as np

from bruges.transform import time_to_depth, depth_to_time

class TimeDepthTest(unittest.TestCase):
    """
    Tests the basic functionality of the time to depth and depth to
    time conversions in bruges.
    """

    def test_depthtotime(self):
        data = np.zeros((100, 100))
        vmodel = np.zeros((100, 100))

        data[0:50, :] += 100.0
        data[50:, :] += 200.0

        vmodel[0:50, :] += 1500.0
        vmodel[50:, :] += 3000.0

        dt = 0.001
        dz = 1.0

        face_change = int(np.floor(((49 * dz) / 1500.0) / dt))

        output = depth_to_time(data, vmodel, dz, dt, twt=False)

        self.assertTrue((output[face_change+1, 50] -
                         output[face_change, 50]) == 100)

    def test_timetodepth(self):
        data = np.zeros((100, 100))
        vmodel = np.zeros((100, 100))

        data[0:50, :] += 100.0
        data[50:, :] += 200.0

        vmodel[0:50, :] += 1500.0
        vmodel[50:, :] += 5000.0

        dt = 0.001
        dz = 1.0

        output = time_to_depth(data, vmodel, dt, dz, twt=False)

        face_change = int(np.floor(((49.5 * dt) * 1500.0) / dz))

        self.assertTrue((output[face_change+1, 50] -
                         output[face_change, 50]) == 100)

    def test_recip(self):
        data = np.zeros((100, 100))
        vmodel = np.zeros((100, 100))

        data[0:50, :] += 100.0
        data[50:, :] += 200.0

        vmodel[0:50, :] += 1500.0
        vmodel[50:, :] += 1800.0

        dt = 0.001
        dz = 1.

        out1 = depth_to_time(data, vmodel, dz, dt)
        v2 = depth_to_time(vmodel, vmodel, dz, dt)
        out2 = time_to_depth(out1, v2, dt, dz)

        # This test is not finished!

if (__name__ == '__main__'):

    suite = \
      unittest.TestLoader().loadTestsFromTestCase(TimeDepthTest)
    unittest.TextTestRunner(verbosity=2).run(suite)

max_stars_repo_name: mycarta/bruges

max_issues_repo_name: mycarta/bruges

max_forks_repo_name: mycarta/bruges


text:
"""
Pytest for Product

"""

__author__ = "jerome.colin'at'cesbio.cnes.fr"
__license__ = "MIT"
__version__ = "1.0.3"

import os
import pytest
import Product
import Roi
import utilities
import numpy
import osgeo.gdal
from matplotlib import pylab as pl

TEST_DATA_PATH = os.environ['TEST_DATA_PATH']

logger = utilities.get_logger('test_Product', verbose=True)

# TEST REFACTORED PRODUCT FROM HERE

## TESTING PRODUCT_DIR (DEFAULT)
def test_product_dir():
    logger.info("TESTING PRODUCT_DIR (DEFAULT)")
    p_dir = Product.Product(TEST_DATA_PATH + "acix_carpentras/SENTINEL2A_20171007-103241-161_L2A_T31TFJ_C_V1-0",
                                  logger)
    assert type(p_dir.content_list) is list

    p_dir_filename = p_dir.find_band("SRE_B4.")
    assert p_dir_filename == TEST_DATA_PATH + "acix_carpentras/SENTINEL2A_20171007-103241-161_L2A_T31TFJ_C_V1-0/SENTINEL2A_20171007-103241-161_L2A_T31TFJ_C_V1-0_SRE_B4.tif"

    p_dir_band = p_dir.get_band(p_dir.find_band("SRE_B4."))
    assert type(p_dir_band) is numpy.ndarray
    assert p_dir_band[0,0] == 850
    assert p_dir_band[6, 2] == 1249

    b4_subset = p_dir.get_band_subset(p_dir.find_band("SRE_B4."), ulx=649455, uly=4238445, lrx=649465, lry=4238435)
    assert type(b4_subset) is numpy.ndarray

def test_product_mask_use_nan():
    logger.info("TESTING PRODUCT GET_MASK WITH NAN")
    p_dir = Product.Product(TEST_DATA_PATH + "acix_carpentras/SENTINEL2A_20171007-103241-161_L2A_T31TFJ_C_V1-0",
                                  logger)
    clm = numpy.zeros((3,3)) + 33
    edg = numpy.zeros_like(clm) + 1
    clm[1,:] = numpy.nan
    clm[2, :] = numpy.nan
    assert utilities.count_nan(clm) == 6
    edg[:,1] = numpy.nan
    assert utilities.count_nan(edg) == 3
    mask, ratio = p_dir.get_mask(clm,edg, stats=True, use_nodata=True)
    assert numpy.sum(mask) == 2
    assert ratio == pytest.approx(2/9*100)
    assert mask[1,1] == 1
    assert mask[2,1] == 1
    logger.debug("test_product_mask ratio=%6.4f" % ratio)

def test_product_mask_use_zeros():
    logger.info("TESTING PRODUCT GET_MASK WITH ZEROS")
    p_dir = Product.Product(TEST_DATA_PATH + "acix_carpentras/SENTINEL2A_20171007-103241-161_L2A_T31TFJ_C_V1-0",
                                  logger)
    clm = numpy.zeros((3,3)) + 33
    edg = numpy.zeros_like(clm) + 1
    clm[1,:] = 0
    clm[2, :] = 0
    assert numpy.count_nonzero(clm) == 3
    edg[:,1] = 0
    assert numpy.count_nonzero(edg) == 6
    mask, ratio = p_dir.get_mask(clm,edg, stats=True, use_nodata=False)
    assert numpy.sum(mask) == 2
    assert ratio == pytest.approx(2/9*100)
    assert mask[1,1] == 1
    assert mask[2,1] == 1
    logger.debug("test_product_mask ratio=%6.4f" % ratio)

## TEST PRODUCT_ZIP_VENUS
def test_product_zip_venus():
    logger.info("TEST PRODUCT_ZIP_VENUS")
    p_zip_venus = Product.Product_zip_venus(TEST_DATA_PATH + "VENUS-XS_20200402-191352-000_L2A_GALLOP30_D.zip", logger)
    assert p_zip_venus.name == "VENUS-XS_20200402-191352-000_L2A_GALLOP30_D.zip"
    assert p_zip_venus.band_names[0] == "SRE_B1."
    assert p_zip_venus.clm_name == "CLM_XS"
    assert p_zip_venus.edg_name == "EDG_XS"

    logger.info("test_product_zip_venus_get_content_list")
    b4_filename = p_zip_venus.find_band("SRE_B4.")
    assert b4_filename == "VENUS-XS_20200402-191352-000_L2A_GALLOP30_C_V2-2/VENUS-XS_20200402-191352-000_L2A_GALLOP30_C_V2-2_SRE_B4.tif"

    logger.info("test_product_zip_venus_get_band")
    b4 = p_zip_venus.get_band(b4_filename)
    assert type(b4) is numpy.ndarray
    logger.debug("b4 is of size %i, with shape %s" % (numpy.size(b4), str(numpy.shape(b4))))
    assert numpy.shape(b4)[0] == 11686
    assert numpy.shape(b4)[1] == 11711
    b4 = None

    logger.info("test_product_zip_venus_get_band_aot")
    atb_filename = p_zip_venus.find_band("ATB_XS")
    assert atb_filename == "VENUS-XS_20200402-191352-000_L2A_GALLOP30_C_V2-2/VENUS-XS_20200402-191352-000_L2A_GALLOP30_C_V2-2_ATB_XS.tif"
    atb_bands = p_zip_venus.get_band(atb_filename)
    logger.debug("atb_bands is of type %s" % str(type(atb_bands)))
    logger.debug("atb_bands is of size %i, with shape %s" % (numpy.size(atb_bands), str(numpy.shape(atb_bands))))
    assert type(atb_bands) is numpy.ndarray
    assert numpy.shape(atb_bands)[0] == 2
    atb_bands = None

    aot = p_zip_venus.get_band(atb_filename, layer=1)
    logger.debug("aot is of size %i, with shape %s" % (numpy.size(aot), str(numpy.shape(aot))))
    assert numpy.shape(aot)[0] == 11686
    assert numpy.shape(aot)[1] == 11711
    aot = None

    aot = p_zip_venus.get_band(p_zip_venus.find_band(p_zip_venus.aot_name), scalef=p_zip_venus.aot_scalef, layer=p_zip_venus.aot_layer)
    logger.debug("aot is of size %i, with shape %s" % (numpy.size(aot), str(numpy.shape(aot))))
    assert numpy.shape(aot)[0] == 11686
    assert numpy.shape(aot)[1] == 11711
    aot = None

    vap = p_zip_venus.get_band(p_zip_venus.find_band(p_zip_venus.vap_name), scalef=p_zip_venus.vap_scalef, layer=p_zip_venus.vap_layer)
    logger.debug("vap is of size %i, with shape %s" % (numpy.size(vap), str(numpy.shape(vap))))
    assert numpy.shape(vap)[0] == 11686
    assert numpy.shape(vap)[1] == 11711
    vap = None

    logger.info("test_product_zip_venus_get_band_subset")
    b4_subset = p_zip_venus.get_band_subset(b4_filename, ulx=649455, uly=4238445, lrx=649465, lry=4238435)
    assert type(b4_subset) is numpy.ndarray
    assert b4_subset[0, 0] == 93
    assert b4_subset[1, 0] == 86
    assert b4_subset[0, 1] == 113
    assert b4_subset[1, 1] == 94
    assert p_zip_venus.sre_scalef == 1000
    b4_subset_SRE = p_zip_venus.get_band_subset(b4_filename, ulx=649455, uly=4238445, lrx=649465, lry=4238435, scalef=p_zip_venus.sre_scalef)
    assert b4_subset_SRE[0, 0] == 0.093
    assert b4_subset_SRE[1, 0] == 0.086
    assert b4_subset_SRE[0, 1] == 0.113
    assert b4_subset_SRE[1, 1] == 0.094

    logger.info("test_product_zip_venus_get_band_subset_withROI")
    roi = Roi.Roi([99, 649460, 4238440], 10, logger)
    b4_subset_SRE = p_zip_venus.get_band_subset(b4_filename, roi=roi, scalef=p_zip_venus.sre_scalef)
    assert b4_subset_SRE[0, 0] == 0.093
    assert b4_subset_SRE[1, 0] == 0.086
    assert b4_subset_SRE[0, 1] == 0.113
    assert b4_subset_SRE[1, 1] == 0.094

    aot = p_zip_venus.get_band_subset(p_zip_venus.find_band(p_zip_venus.aot_name), ulx=649455, uly=4238445, lrx=649465, lry=4238435, scalef=p_zip_venus.aot_scalef, layer=p_zip_venus.aot_layer)
    logger.debug("aot is of size %i, with shape %s" % (numpy.size(aot), str(numpy.shape(aot))))
    assert aot[0, 0] == 0.195
    assert aot[1, 0] == 0.195
    assert aot[0, 1] == 0.195
    assert aot[1, 1] == 0.195

    vap = p_zip_venus.get_band_subset(p_zip_venus.find_band(p_zip_venus.vap_name), ulx=649455, uly=4238445, lrx=649465, lry=4238435, scalef=p_zip_venus.vap_scalef, layer=p_zip_venus.vap_layer)
    logger.debug("vap is of size %i, with shape %s" % (numpy.size(vap), str(numpy.shape(vap))))
    assert vap[0, 0] == 0.55
    assert vap[1, 0] == 0.55
    assert vap[0, 1] == 0.55
    assert vap[1, 1] == 0.55

## TEST PRODUCT_HDF
def test_product_hdf():
    logger.info("TEST PRODUCT_HDF")
    p_hdf_vermote = Product.Product_hdf(TEST_DATA_PATH + "vermote_carpentras/refsrs2-L1C_T31TFJ_A003037_20171005T104550-Carpentras.hdf", logger)

    assert p_hdf_vermote.ptype == "HDF"

    logger.info("test_product_hdf_content_list")
    logger.debug(type(p_hdf_vermote.content_list))
    logger.debug(p_hdf_vermote.content_list)
    assert type(p_hdf_vermote.content_list) is list

    logger.info("test_product_hdf_find_band")
    subdsid = p_hdf_vermote.find_band("band04-red")
    assert type(subdsid) is int
    assert subdsid == 3

    logger.info("test_product_hdf_get_band")
    assert type(p_hdf_vermote.get_band(p_hdf_vermote.find_band("band04-red"))) is numpy.ndarray
    assert p_hdf_vermote.get_band(p_hdf_vermote.find_band("band04-red"))[0][0] == 596
    assert p_hdf_vermote.get_band(p_hdf_vermote.find_band("band04-red"))[6][2] == 1096

def test_product_hdf_acix():
    p_hdf_acix = Product.Product_hdf_acix(TEST_DATA_PATH + "vermote_carpentras/refsrs2-L1C_T31TFJ_A012260_20171027T103128-Carpentras.hdf", logger)
    assert p_hdf_acix.sre_scalef == 10000
    b7 = p_hdf_acix.get_band(p_hdf_acix.find_band("band07"), scalef=p_hdf_acix.sre_scalef)
    assert type(b7) is numpy.ndarray
    assert b7[12,5] == 0.1829
    assert b7[899,899] == 0.2021

## TESTING PRODUCT_DIR_MAJA
def test_product_dir_maja():
    logger.info("TESTING PRODUCT_DIR_MAJA")
    p_dir_maja = Product.Product_dir_maja(TEST_DATA_PATH + "acix_carpentras/SENTINEL2A_20171007-103241-161_L2A_T31TFJ_C_V1-0",
                                  logger)
    assert type(p_dir_maja.content_list) is list

    p_dir_b4_filename = p_dir_maja.find_band("SRE_B4.")
    assert p_dir_b4_filename == TEST_DATA_PATH + "acix_carpentras/SENTINEL2A_20171007-103241-161_L2A_T31TFJ_C_V1-0/SENTINEL2A_20171007-103241-161_L2A_T31TFJ_C_V1-0_SRE_B4.tif"

    p_dir_band = p_dir_maja.get_band(p_dir_b4_filename, scalef=p_dir_maja.sre_scalef)
    assert type(p_dir_band) is numpy.ndarray
    assert p_dir_band[0,0] == 0.0850
    assert p_dir_band[6, 2] == 0.1249

    roi = Roi.Roi([88, 660260, 4887620], 20, logger)
    b4_subset = p_dir_maja.get_band_subset(p_dir_maja.find_band("SRE_B4."), roi=roi, scalef=p_dir_maja.sre_scalef)
    assert type(b4_subset) is numpy.ndarray
    assert b4_subset[0, 0] == 0.1362
    assert b4_subset[1, 0] == 0.1451
    assert b4_subset[0, 1] == 0.1173
    assert b4_subset[1, 1] == 0.1306

max_stars_repo_name: jerome-colin/mtools

max_issues_repo_name: jerome-colin/mtools

max_forks_repo_name: jerome-colin/mtools


text:
import numpy as np
import pandas as pd
import logging
from collections import OrderedDict
from yggdrasil import serialize, platform
from yggdrasil.drivers.InterpretedModelDriver import InterpretedModelDriver
from yggdrasil.drivers.PythonModelDriver import PythonModelDriver
from yggdrasil.drivers.CModelDriver import CModelDriver
from yggdrasil.languages.R import install

logger = logging.getLogger(__name__)

class RModelDriver(InterpretedModelDriver):  # pragma: R
    r"""Base class for running R models.

    Args:
        name (str): Driver name.
        args (str or list): Argument(s) for running the model in matlab.
            Generally, this should be the full path to a Matlab script.
        **kwargs: Additional keyword arguments are passed to parent class's
            __init__ method.

    """
    _schema_subtype_description = ('Model is written in R.')
    language = 'R'
    language_aliases = ['r']
    language_ext = '.R'
    base_languages = ['python']
    default_interpreter = 'Rscript'
    # Dynamically setting the interface library causes circular a import so
    # it is defined here statically. For dynamic import, use the following:
    #     interface_library = PythonModelDriver.interface_library
    interface_library = 'yggdrasil'
    interface_dependencies = install.requirements_from_description()
    # The Batch version causes output to saved to a file rather than directed to
    # stdout so use Rscript instead. For the batch version, use the following:
    #     default_interpreter_flags = ['CMD', 'BATCH' '--vanilla', '--silent']
    default_interpreter_flags = ['--default-packages=methods,utils']
    send_converters = {'table': serialize.consolidate_array}
    type_map = {
        'int': 'integer, bit64::integer64',
        'float': 'double',
        'string': 'character',
        'array': 'list',
        'object': 'list',
        'boolean': 'logical',
        'null': 'NA',
        'uint': 'integer',
        'complex': 'complex',
        'bytes': 'char (utf-8)',
        'unicode': 'char',
        '1darray': 'list',
        'ndarray': 'list',
        'ply': 'PlyDict',
        'obj': 'ObjDict',
        'schema': 'list'}
    function_param = {
        'import': 'source(\"{filename}\")',
        'istype': 'is({variable}, \"{type}\")',
        'len': 'length({variable})',
        'index': '{variable}[[{index}]]',
        'first_index': 1,
        'interface': 'library(yggdrasil)',
        'input': '{channel} <- YggInterface(\"YggInput\", \"{channel_name}\")',
        'output': '{channel} <- YggInterface(\"YggOutput\", \"{channel_name}\")',
        'python_interface': ('{channel} <- YggInterface(\"{python_interface}\", '
                             '\"{channel_name}\")'),
        'python_interface_format': ('{channel} <- YggInterface('
                                    '\"{python_interface}\", '
                                    '\"{channel_name}\", '
                                    '\"{format_str}\")'),
        'recv_function': '{channel}$recv',
        'send_function': '{channel}$send',
        'multiple_outputs': 'c({outputs})',
        'multiple_outputs_def': 'list({outputs})',
        'true': 'TRUE',
        'false': 'FALSE',
        'not': '!',
        'and': '&&',
        'comment': '#',
        'indent': 2 * ' ',
        'quote': '\"',
        'print_generic': 'print({object})',
        'print': 'print(\"{message}\")',
        'fprintf': 'print(sprintf(\"{message}\", {variables}))',
        'error': 'stop(\"{error_msg}\")',
        'block_end': '}',
        'if_begin': 'if ({cond}) {{',
        'if_elif': '}} else if ({cond}) {{',
        'if_else': '}} else {{',
        'for_begin': 'for ({iter_var} in {iter_begin}:{iter_end}) {{',
        'while_begin': 'while ({cond}) {{',
        'try_begin': 'tryCatch({',
        'try_except': '}}, error = function({error_var}) {{',
        'try_end': '})',
        'assign': '{name} <- {value}',
        'assign_mult': '{name} %<-% {value}',
        'function_def_begin': '{function_name} <- function({input_var}) {{',
        'return': 'return({output_var})',
        'function_def_regex': (
            r'{function_name} *(?:(?:\<-)|(?:=)) *function'
            r'\((?P<inputs>(?:.|(?:\r?\n))*?)\)\s*\{{'
            r'(?P<body>(?:.|(?:\r?\n))*?)'
            r'(?:return\((list\()?'
            r'(?P<outputs>(?:.|(?:\r?\n))*?)(?(3)\))\)'
            r'(?:.|(?:\r?\n))*?\}})'
            r'|(?:\}})'),
        'inputs_def_regex': r'\s*(?P<name>.+?)\s*(?:,|$)',
        'outputs_def_regex': r'\s*(?P<name>.+?)\s*(?:,|$)'}
    brackets = (r'{', r'}')

    @classmethod
    def is_library_installed(cls, lib, **kwargs):
        r"""Determine if a dependency is installed.

        Args:
            lib (str): Name of the library that should be checked.
            **kwargs: Additional keyword arguments are ignored.

        Returns:
            bool: True if the library is installed, False otherwise.

        """
        if lib not in cls._library_cache:
            try:
                cls.run_executable(['-e', 'library(%s)' % lib.split()[0]])
                cls._library_cache[lib] = True
            except RuntimeError:
                cls._library_cache[lib] = False
        return cls._library_cache[lib]

    @classmethod
    def language_version(cls, **kwargs):
        r"""Determine the version of this language.

        Args:
            **kwargs: Keyword arguments are passed to the parent class's
                method.

        Returns:
            str: Version of compiler/interpreter for this language.

        """
        kwargs.setdefault('skip_interpreter_flags', True)
        return super(RModelDriver, cls).language_version(**kwargs)

    # @property
    # def debug_flags(self):
    #     r"""list: Flags that should be prepended to an executable command to
    #     enable debugging."""
    #     if self.with_valgrind:
    #         interp = 'R'.join(self.get_interpreter().rsplit('Rscript', 1))
    #         return [interp, '-d', '"valgrind %s"'
    #                 % ' '.join(self.valgrind_flags), '--vanilla', '-f']
    #     return super(RModelDriver, self).debug_flags

    def set_env(self):
        r"""Get environment variables that should be set for the model process.

        Returns:
            dict: Environment variables for the model process.

        """
        out = super(RModelDriver, self).set_env()
        out['RETICULATE_PYTHON'] = PythonModelDriver.get_interpreter()
        if CModelDriver.is_language_installed():
            c_linker = CModelDriver.get_tool('linker')
            search_dirs = c_linker.get_search_path(env_only=True)
            out = CModelDriver.update_ld_library_path(out, paths_to_add=search_dirs,
                                                      add_to_front=True)
        return out

    @classmethod
    def comm_atexit(cls, comm):  # pragma: no cover
        r"""Operations performed on comm at exit including draining receive.

        Args:
            comm (CommBase): Communication object.

        """
        if comm.direction == 'recv':
            while comm.recv(timeout=0)[0]:
                comm.sleep()
        else:
            comm.send_eof()
        if not getattr(comm, 'dont_backlog', True):
            comm.linger_close()

    @classmethod
    def python2language(cls, pyobj):
        r"""Prepare a python object for transformation in R.

        Args:
            pyobj (object): Python object.

        Returns:
            object: Python object in a form that is R friendly.

        """
        logger.debug("python2language: %s, %s" % (pyobj, type(pyobj)))
        if isinstance(pyobj, tuple):
            return tuple([cls.python2language(x) for x in pyobj])
        elif isinstance(pyobj, list):
            return [cls.python2language(x) for x in pyobj]
        elif isinstance(pyobj, OrderedDict):
            return OrderedDict([(cls.python2language(k),
                                 cls.python2language(v))
                                for k, v in pyobj.items()])
        elif isinstance(pyobj, dict):
            return {cls.python2language(k): cls.python2language(v)
                    for k, v in pyobj.items()}
        elif isinstance(pyobj, np.string_):
            return pyobj.decode("utf-8")
        elif isinstance(pyobj, pd.DataFrame):
            # R dosn't have int64 and will cast 64bit ints as floats if passed
            # without casting them to int32 first
            for n in pyobj.columns:
                if pyobj[n].dtype == np.dtype('int64'):
                    pyobj[n] = pyobj[n].astype('int32')
        return pyobj

    @classmethod
    def write_model_wrapper(cls, model_file, model_function, **kwargs):
        r"""Return the lines required to wrap a model function as an integrated
        model.

        Args:
            model_file (str): Full path to the file containing the model
                function's declaration.
            model_function (str): Name of the model function.
            **kwargs: Additional keyword arguments are passed to the parent
                class's method.

        Returns:
            list: Lines of code wrapping the provided model with the necessary
                code to run it as part of an integration.

        """
        if platform._is_win:  # pragma: windows
            model_file = model_file.replace('\\', '/')
        return super(RModelDriver, cls).write_model_wrapper(
            model_file, model_function, **kwargs)


text:
# -*- coding: utf-8 -*-
"""
Created on Fri Jan 16 01:08:05 2015

@author: hoseung
"""

#%%
import numpy as np
import load
import time
import utils.util as utl
import utils.prettyplot as ptt

base = '/home/hoseung/Work/data/group3826/'
nout=135
#base = '/home/hoseung/Work/data/Lmax_136/'
#nout=136
#base = '/home/hoseung/Work/data/C04466/'
#nout=181
#base = '/home/hoseung/Work/data/AGN2/'
#nout=51

foutput = 'field_Lmax136.npy'

import utils.sampling as smp

#region = smp.set_region(xc=0.205, yc=0.78, zc=0.395, radius=0.01)
#print(region)
#xrs = region["ranges"]
#xrs = [[0.21, 0.22], [0.77, 0.78], [0.39, 0.40]]
s = load.sim.Sim(nout, base) # , ranges=xrs)
#s.set_ranges(xrs)
#%%
s.add_part('dm')
print(" add_part ")
#s.part.dm['px'].ptp()

#region = smp.set_region(centers=[0.6,0.4,0.5],radius=0.005)
#print(" set_region ")

s.part.load()
#print(" set_part ")
#%%
region = s.part.search_zoomin(scale=0.5, load=True)

s.set_ranges(region["ranges"])
# sim.set_ranges does not automatically change sim.part.ranges
# Can I make sim.part.ranges to be t

print(" ------cc ")
s.part.load()
#%%

# part
x = s.part.dm['px']
y = s.part.dm['py']  # These are views. right?
z = s.part.dm['py']
m = s.part.dm['m'] * s.info.msun # part must be normalized already!

npix = 100

from draw import pp
import matplotlib.pyplot as plt
field = pp.den2d(x, y, z, m, npix, s.info, cic=True, norm_integer=True)
np.save(foutput, field)
# Save

#%%
import matplotlib.pyplot as plt
field = np.random.rand(300,300)
from matplotlib.colors import LogNorm
cname = 'brg'
vmin = field.max() * 1e-6
vmax = field.max()

p = plt.imshow(field, cmap=plt.get_cmap(cname), norm=LogNorm(vmin=vmin, vmax=vmax))
"""
nticks = 2
xr = np.asarray(region['xr'])
yr = np.asarray(region['yr'])
x_old, x_new = ptt.tickInPboxInt(xr * s.info.pboxsize, field.shape[0], nticks=nticks)
print(xr * s.info.pboxsize)
print(x_new)
print(x_old * npix)
y_old, y_new = ptt.tickInPboxInt(yr * s.info.pboxsize, field.shape[1], nticks=nticks)

p.xticks(x_old * npix, x_new)
p.yticks(y_old * npix, y_new)
"""

#legend
cbar = plt.colorbar(p)
cbar.ax.set_yticklabels(['0','1','2','>3'])
cbar.set_label(r'$Log\Sigma$ $(M_{\odot} kpc^{-2})$', rotation=270, labelpad=25)
plt.show()
plt.savefig('test.png', dpi=72)
plt.savefig('test.eps', dpi=200)
#%%
plt.close()

#plt.hist(field.flatten(), 50, normed=1, facecolor='green', alpha=0.75)
#plt.show()
# flatten 2D array before passing to .hist()

max_stars_repo_name: Hoseung/pyRamAn

max_issues_repo_name: Hoseung/pyRamAn

max_forks_repo_name: Hoseung/pyRamAn


text:
'''
array.py
This code is made for array manipulations

MODIFICATION HISTORY:
    dsj, 21, JUL, 2015: VERSION 1.00
    dsj, 12, JUL, 2019: VERSION 2.00
      - add print_dict

'''

import numpy as np

def dict_sum(dict,list):

    for i, key in enumerate( list ):

        if i == 0:
            total = dict[key]
        else:
            total = total + dict[key]

    return np.copy( total )

'''
dict = {}
for aa in ['A1','B1','C1']:
    dict[aa] = {}
    for bb in ['A2','B2','C2']:
        dict[aa][bb] = {}
        for cc in ['A3','B3','C3']:
            dict[aa][bb][cc] = {}
            for dd in ['A4','B4','C4']:
                dict[aa][bb][cc][dd] = {}
                for ee in ['A5','B5','C5']:
                    dict[aa][bb][cc][dd][ee] = aa + bb + cc + dd + ee
'''

def print_dict(dict,pos,names):

    keystring1 = 'list(dict'
    keystring2 = '.keys())'
    keystring3 = '[0]'
    keystring = {}
    keystring[0] = keystring1 + keystring2
    for i in np.arange(pos)+1:
        keystring[i] = keystring1
        for ii in np.arange(i):
            keystring[i] += '[' + keystring[ii] + keystring3 + ']'
        keystring[i] += keystring2

    forstring = ''
    printstring = 'print('
    printstring_ind = 'dict'
    availstring = ''
    for i in np.arange(pos):
        forstring += '    '*i + 'for key' + str(i) + ' in ' + keystring[i] + ':\n'
        printstring += 'key' + str(i) + ','
    for i in np.arange(pos):
        printstring_ind += '[key' + str(i) + ']'
        availstring += 'key' + str(i) + ','
    printstring += 'name,' + printstring_ind + '[name])'
    forstring += '    '*(pos+1) + 'for name in names:\n'
    forstring += '    '*(pos+2) + 'try:\n'
    forstring += '    '*(pos+3) + printstring + '\n'
    forstring += '    '*(pos+2) + 'except:\n'
    forstring += '    '*(pos+3) + 'print("not available:",' + availstring + 'name' + ')'

    exec(forstring)


text:
import os
import numpy as np
import pandas as pd
import datetime
import random
import pytest

from skmultiflow.data.temporal_data_stream import TemporalDataStream

def test_temporal_data_stream(test_path):
    test_file = os.path.join(test_path, 'sea_stream_file.csv')
    raw_data = pd.read_csv(test_file)
    stream = TemporalDataStream(raw_data, name='Test')

    assert stream.n_remaining_samples() == 40

    expected_names = ['attrib1', 'attrib2', 'attrib3']
    assert stream.feature_names == expected_names

    expected_targets = [0, 1]
    assert stream.target_values == expected_targets

    assert stream.target_names == ['class']

    assert stream.n_features == 3

    assert stream.n_cat_features == 0

    assert stream.n_num_features == 3

    assert stream.n_targets == 1

    assert stream.get_data_info() == 'Test: 1 target(s), 2 classes'

    assert stream.has_more_samples() is True

    assert stream.is_restartable() is True

    # Load test data corresponding to first 10 instances
    test_file = os.path.join(test_path, 'sea_stream_file.npz')
    data = np.load(test_file)
    X_expected = data['X']
    y_expected = data['y']

    X, y, _, _, _ = stream.next_sample()
    assert np.alltrue(X[0] == X_expected[0])
    assert np.alltrue(y[0] == y_expected[0])

    X, y, _, _, _ = stream.last_sample()
    assert np.alltrue(X[0] == X_expected[0])
    assert np.alltrue(y[0] == y_expected[0])

    stream.restart()
    X, y, _, _, _ = stream.next_sample(10)
    assert np.alltrue(X == X_expected)
    assert np.alltrue(y == y_expected)

    assert stream.n_targets == np.array(y).ndim

    assert stream.n_features == X.shape[1]

    assert 'stream' == stream._estimator_type

    expected_info = "DataStream(n_targets=-1, target_idx=1, cat_features=None, name='Test')"
    assert stream.get_info() == expected_info

def test_temporal_data_stream_X_y(test_path):
    test_file = os.path.join(test_path, 'sea_stream_file.csv')
    raw_data = pd.read_csv(test_file)
    y = raw_data.iloc[:, -1:]
    X = raw_data.iloc[:, :-1]
    stream = TemporalDataStream(X, y)

    assert stream._Y_is_defined

    assert stream.n_remaining_samples() == 40

    expected_names = ['attrib1', 'attrib2', 'attrib3']
    assert stream.feature_names == expected_names

    expected_targets = [0, 1]
    assert stream.target_values == expected_targets

    assert stream.target_names == ['class']

    assert stream.n_features == 3

    assert stream.n_cat_features == 0

    assert stream.n_num_features == 3

    assert stream.n_targets == 1

    assert stream.get_data_info() == '1 target(s), 2 classes'

    assert stream.has_more_samples() is True

    assert stream.is_restartable() is True

    # Load test data corresponding to first 10 instances
    test_file = os.path.join(test_path, 'sea_stream_file.npz')
    data = np.load(test_file)
    X_expected = data['X']
    y_expected = data['y']

    X, y, _, _, _ = stream.next_sample()
    assert np.alltrue(X[0] == X_expected[0])
    assert np.alltrue(y[0] == y_expected[0])

    X, y, _, _, _ = stream.last_sample()
    assert np.alltrue(X[0] == X_expected[0])
    assert np.alltrue(y[0] == y_expected[0])

    stream.restart()
    X, y, _, _, _ = stream.next_sample(10)
    assert np.alltrue(X == X_expected)
    assert np.alltrue(y == y_expected)

    assert stream.n_targets == np.array(y).ndim

    assert stream.n_features == X.shape[1]

def test_check_data():
    # Test if data contains non-numeric values
    data = pd.DataFrame(np.array([[1, 2, 3, 4, 5],
                                  [6, 7, 8, 9, 10],
                                  [11, 'invalid', 13, 14, 15]]))

    with pytest.raises(ValueError):
        TemporalDataStream(data=data, allow_nan=False)

    # Test if data contains NaN values
    data = pd.DataFrame(np.array([[1, 2, 3, 4, 5],
                                  [6, 7, 8, 9, 10],
                                  [11, np.nan, 13, 14, 15]]))

    with pytest.raises(ValueError):
        TemporalDataStream(data=data, allow_nan=False)

    # Test warning for NaN values

    with pytest.warns(UserWarning):
        TemporalDataStream(data=data, allow_nan=True)

def test_temporal_data_stream_time_ordered(test_path):
    test_file = os.path.join(test_path, 'sea_stream_file.csv')
    raw_data = pd.read_csv(test_file)
    X = raw_data[["attrib1", "attrib2", "attrib3"]].values
    y = raw_data["class"].values
    time = generate_random_dates(1, X.shape[0])

    stream = TemporalDataStream(X, y, time=time, ordered=True)

def test_temporal_data_stream_time_not_ordered(test_path):
    test_file = os.path.join(test_path, 'sea_stream_file.csv')
    raw_data = pd.read_csv(test_file)
    X = raw_data[["attrib1", "attrib2", "attrib3"]].values
    y = raw_data["class"].values
    time = generate_random_dates(1, X.shape[0])

    stream = TemporalDataStream(X, y, time=time, ordered=False)

    # check if time is ordered
    ordered = (np.diff(stream.time)>=0).all()
    assert ordered == True

def test_temporal_data_stream_time_with_delay(test_path):
    test_file = os.path.join(test_path, 'sea_stream_file.csv')
    raw_data = pd.read_csv(test_file)
    X = raw_data[["attrib1", "attrib2", "attrib3"]].values
    y = raw_data["class"].values
    time = generate_random_dates(1, X.shape[0])
    delay = generate_random_delays(seed=1, samples=time)

    stream = TemporalDataStream(X, y, time=time, sample_delay=delay, ordered=False)

def test_temporal_data_stream_time_with_weight(test_path):
    test_file = os.path.join(test_path, 'sea_stream_file.csv')
    raw_data = pd.read_csv(test_file)
    X = raw_data[["attrib1", "attrib2", "attrib3"]].values
    y = raw_data["class"].values
    time = generate_random_dates(1, X.shape[0])
    delay = generate_random_delays(seed=1, samples=time)
    random.seed(1)
    weight = np.array([random.random() for i in range(X.shape[0])])

    stream = TemporalDataStream(X, y, time=time, sample_delay=delay, sample_weight=weight, ordered=False)

def generate_random_dates(seed, samples):
    start = datetime.datetime(2020, 4, 30)
    end = datetime.datetime(2020, 7, 30)
    random.seed(seed)
    time = [random.random() * (end - start) + start for _ in range(samples)]
    return np.array(time, dtype="datetime64")

def generate_random_delays(seed, samples):
    random.seed(seed)
    delays = []
    for d in samples:
        delays.append(d + np.timedelta64(int(random.random() * 30),"D"))
    return np.array(delays, dtype="datetime64")

max_stars_repo_name: denisesato/scikit-multiflow

max_issues_repo_name: denisesato/scikit-multiflow

max_forks_repo_name: denisesato/scikit-multiflow