text:
# Copyright 2019 IBM Corporation
#
# 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.

import unittest
import lale.pretty_print

class TestToGraphviz(unittest.TestCase):
    def test_with_operator_choice(self):
        from lale.operators import make_choice
        from lale.lib.lale import NoOp
        from lale.lib.sklearn import KNeighborsClassifier
        from lale.lib.sklearn import LogisticRegression
        from lale.lib.sklearn import Nystroem
        from lale.lib.sklearn import PCA
        kernel_tfm_or_not =  NoOp | Nystroem
        tfm = PCA
        clf = make_choice(LogisticRegression, KNeighborsClassifier)
        clf.visualize(ipython_display=False)
        optimizable = kernel_tfm_or_not >> tfm >> clf
        optimizable.visualize(ipython_display=False)

    def test_invalid_input(self):
        from sklearn.linear_model import LogisticRegression as SklearnLR
        scikit_lr = SklearnLR()
        from lale.helpers import to_graphviz
        with self.assertRaises(TypeError):
            to_graphviz(scikit_lr)

class TestPrettyPrint(unittest.TestCase):
    def _roundtrip(self, expected, printed):
        self.maxDiff = None
        self.assertEqual(expected, printed)
        globals2 = {}
        exec(printed, globals2)
        pipeline2 = globals2['pipeline']

    def test_indiv_op_1(self):
        from lale.lib.sklearn import LogisticRegression
        pipeline = LogisticRegression(solver=LogisticRegression.enum.solver.saga, C=0.9)
        expected = """from lale.lib.sklearn import LogisticRegression
import lale
lale.wrap_imported_operators()

pipeline = LogisticRegression(solver='saga', C=0.9)"""
        self._roundtrip(expected, lale.pretty_print.to_string(pipeline))

    def test_indiv_op_2(self):
        from lale.lib.sklearn import LogisticRegression
        pipeline = LogisticRegression()
        expected = """from lale.lib.sklearn import LogisticRegression
import lale
lale.wrap_imported_operators()

pipeline = LogisticRegression()"""
        self._roundtrip(expected, lale.pretty_print.to_string(pipeline))

    def test_reducible(self):
        from lale.lib.sklearn import MinMaxScaler
        from lale.lib.lale import NoOp
        from lale.lib.sklearn import PCA
        from lale.lib.sklearn import Nystroem
        from lale.lib.lale import ConcatFeatures
        from lale.lib.sklearn import KNeighborsClassifier
        from lale.lib.sklearn import LogisticRegression
        pca = PCA(copy=False)
        logistic_regression = LogisticRegression(solver='saga', C=0.9)
        pipeline = (MinMaxScaler | NoOp) >> (pca & Nystroem) >> ConcatFeatures >> (KNeighborsClassifier | logistic_regression)
        expected = \
"""from lale.lib.sklearn import MinMaxScaler
from lale.lib.lale import NoOp
from lale.lib.sklearn import PCA
from lale.lib.sklearn import Nystroem
from lale.lib.lale import ConcatFeatures
from lale.lib.sklearn import KNeighborsClassifier
from lale.lib.sklearn import LogisticRegression
import lale
lale.wrap_imported_operators()

pca = PCA(copy=False)
logistic_regression = LogisticRegression(solver='saga', C=0.9)
pipeline = (MinMaxScaler | NoOp) >> (pca & Nystroem) >> ConcatFeatures >> (KNeighborsClassifier | logistic_regression)"""
        self._roundtrip(expected, lale.pretty_print.to_string(pipeline))

    def test_no_combinators(self):
        from lale.lib.sklearn import MinMaxScaler
        from lale.lib.lale import NoOp
        from lale.lib.sklearn import PCA
        from lale.lib.sklearn import Nystroem
        from lale.lib.lale import ConcatFeatures
        from lale.lib.sklearn import KNeighborsClassifier
        from lale.lib.sklearn import LogisticRegression
        pca = PCA(copy=False)
        logistic_regression = LogisticRegression(solver='saga', C=0.9)
        pipeline = (MinMaxScaler | NoOp) >> (pca & Nystroem & NoOp) >> ConcatFeatures >> (KNeighborsClassifier | logistic_regression)
        expected = \
"""from lale.lib.sklearn import MinMaxScaler
from lale.lib.lale import NoOp
from lale.operators import make_choice
from lale.lib.sklearn import PCA
from lale.lib.sklearn import Nystroem
from lale.operators import make_union
from lale.lib.sklearn import KNeighborsClassifier
from lale.lib.sklearn import LogisticRegression
from lale.operators import make_pipeline

choice_0 = make_choice(MinMaxScaler, NoOp)
pca = PCA(copy=False)
union = make_union(pca, Nystroem, NoOp)
logistic_regression = LogisticRegression(solver='saga', C=0.9)
choice_1 = make_choice(KNeighborsClassifier, logistic_regression)
pipeline = make_pipeline(choice_0, union, choice_1)"""
        printed = lale.pretty_print.to_string(pipeline, combinators=False)
        self._roundtrip(expected, printed)

    def test_import_as_1(self):
        from lale.lib.sklearn import LogisticRegression as LR
        pipeline = LR(solver='saga', C=0.9)
        expected = """from lale.lib.sklearn import LogisticRegression as LR
import lale
lale.wrap_imported_operators()

pipeline = LR(solver='saga', C=0.9)"""
        self._roundtrip(expected, lale.pretty_print.to_string(pipeline))

    def test_import_as_2(self):
        from lale.lib.sklearn import MinMaxScaler as Scaler
        from lale.lib.lale import NoOp
        from lale.lib.sklearn import PCA
        from lale.lib.sklearn import Nystroem
        from lale.lib.lale import ConcatFeatures as Concat
        from lale.lib.sklearn import KNeighborsClassifier as KNN
        from lale.lib.sklearn import LogisticRegression as LR
        pca = PCA(copy=False)
        lr = LR(solver='saga', C=0.9)
        pipeline = (Scaler | NoOp) >> (pca & Nystroem) >> Concat >> (KNN | lr)
        expected = \
"""from lale.lib.sklearn import MinMaxScaler as Scaler
from lale.lib.lale import NoOp
from lale.lib.sklearn import PCA
from lale.lib.sklearn import Nystroem
from lale.lib.lale import ConcatFeatures as Concat
from lale.lib.sklearn import KNeighborsClassifier as KNN
from lale.lib.sklearn import LogisticRegression as LR
import lale
lale.wrap_imported_operators()

pca = PCA(copy=False)
lr = LR(solver='saga', C=0.9)
pipeline = (Scaler | NoOp) >> (pca & Nystroem) >> Concat >> (KNN | lr)"""
        self._roundtrip(expected, lale.pretty_print.to_string(pipeline))

    def test_operator_choice(self):
        from lale.lib.sklearn import PCA
        from lale.lib.sklearn import MinMaxScaler as Scl
        pipeline = PCA | Scl
        expected = \
"""from lale.lib.sklearn import PCA
from lale.lib.sklearn import MinMaxScaler as Scl
import lale
lale.wrap_imported_operators()

pipeline = PCA | Scl"""
        self._roundtrip(expected, lale.pretty_print.to_string(pipeline))

    def test_higher_order(self):
        from lale.lib.lale import Both
        from lale.lib.sklearn import PCA
        from lale.lib.sklearn import Nystroem
        pipeline = Both(op1=PCA(n_components=2), op2=Nystroem)
        expected = """from lale.lib.lale import Both
from lale.lib.sklearn import PCA
from lale.lib.sklearn import Nystroem
import lale
lale.wrap_imported_operators()

pca = PCA(n_components=2)
pipeline = Both(op1=pca, op2=Nystroem)"""
        self._roundtrip(expected, lale.pretty_print.to_string(pipeline))

    def test_higher_order_2(self):
        from lale.lib.sklearn import VotingClassifier as Vote
        from lale.lib.sklearn import KNeighborsClassifier as KNN
        from lale.lib.sklearn import PCA
        from lale.lib.sklearn import LogisticRegression as LR
        pipeline = Vote(estimators=[('knn',KNN), ('pipeline',PCA()>>LR)],
                        voting='soft')
        expected = """from lale.lib.sklearn import VotingClassifier as Vote
from lale.lib.sklearn import KNeighborsClassifier as KNN
from lale.lib.sklearn import PCA
from lale.lib.sklearn import LogisticRegression as LR
import lale
lale.wrap_imported_operators()

pipeline = Vote(estimators=[('knn', KNN), ('pipeline', PCA() >> LR)], voting='soft')"""
        self._roundtrip(expected, lale.pretty_print.to_string(pipeline))

    def test_multimodal(self):
        from lale.lib.lale import Project
        from lale.lib.sklearn import Normalizer as Norm
        from lale.lib.sklearn import OneHotEncoder as OneHot
        from lale.lib.lale import ConcatFeatures as Cat
        from lale.lib.sklearn import LinearSVC
        project_0 = Project(columns={'type': 'number'})
        project_1 = Project(columns={'type': 'string'})
        linear_svc = LinearSVC(C=29617.4, dual=False, tol=0.005266)
        pipeline = ((project_0 >> Norm()) & (project_1 >> OneHot())) >> Cat >> linear_svc
        expected = \
"""from lale.lib.lale import Project
from lale.lib.sklearn import Normalizer as Norm
from lale.lib.sklearn import OneHotEncoder as OneHot
from lale.lib.lale import ConcatFeatures as Cat
from lale.lib.sklearn import LinearSVC
import lale
lale.wrap_imported_operators()

project_0 = Project(columns={'type': 'number'})
project_1 = Project(columns={'type': 'string'})
linear_svc = LinearSVC(C=29617.4, dual=False, tol=0.005266)
pipeline = ((project_0 >> Norm()) & (project_1 >> OneHot())) >> Cat >> linear_svc"""
        self._roundtrip(expected, lale.pretty_print.to_string(pipeline))

    def test_irreducible_1(self):
        from lale.lib.sklearn import PCA
        from lale.lib.sklearn import Nystroem
        from lale.lib.sklearn import MinMaxScaler
        from lale.lib.sklearn import LogisticRegression
        from lale.lib.sklearn import KNeighborsClassifier
        from lale.operators import get_pipeline_of_applicable_type
        choice = PCA | Nystroem
        pipeline = get_pipeline_of_applicable_type(
            steps=[choice, MinMaxScaler, LogisticRegression, KNeighborsClassifier],
            edges=[(choice,LogisticRegression), (MinMaxScaler,LogisticRegression), (MinMaxScaler,KNeighborsClassifier)])
        expected = \
"""from lale.lib.sklearn import PCA
from lale.lib.sklearn import Nystroem
from lale.lib.sklearn import MinMaxScaler
from lale.lib.sklearn import LogisticRegression
from lale.lib.sklearn import KNeighborsClassifier
from lale.operators import get_pipeline_of_applicable_type
import lale
lale.wrap_imported_operators()

choice = PCA | Nystroem
pipeline = get_pipeline_of_applicable_type(steps=[choice, MinMaxScaler, LogisticRegression, KNeighborsClassifier], edges=[(choice,LogisticRegression), (MinMaxScaler,LogisticRegression), (MinMaxScaler,KNeighborsClassifier)])"""
        self._roundtrip(expected, lale.pretty_print.to_string(pipeline))

    def test_irreducible_2(self):
        from lale.lib.sklearn import PCA
        from lale.lib.sklearn import MinMaxScaler as MMS
        from lale.lib.lale import ConcatFeatures as HStack
        from lale.lib.sklearn import KNeighborsClassifier as KNN
        from lale.lib.sklearn import LogisticRegression as LR
        from lale.operators import get_pipeline_of_applicable_type
        pipeline_0 = HStack >> LR
        pipeline = get_pipeline_of_applicable_type(
            steps=[PCA, MMS, KNN, pipeline_0],
            edges=[(PCA, KNN), (PCA, pipeline_0), (MMS, pipeline_0)])
        expected = \
"""from lale.lib.sklearn import PCA
from lale.lib.sklearn import MinMaxScaler as MMS
from lale.lib.sklearn import KNeighborsClassifier as KNN
from lale.lib.lale import ConcatFeatures as HStack
from lale.lib.sklearn import LogisticRegression as LR
from lale.operators import get_pipeline_of_applicable_type
import lale
lale.wrap_imported_operators()

pipeline_0 = HStack >> LR
pipeline = get_pipeline_of_applicable_type(steps=[PCA, MMS, KNN, pipeline_0], edges=[(PCA,KNN), (PCA,pipeline_0), (MMS,pipeline_0)])"""
        self._roundtrip(expected, lale.pretty_print.to_string(pipeline))

    def test_nested(self):
        from lale.lib.sklearn import PCA
        from lale.lib.sklearn import LogisticRegression as LR
        from lale.lib.lale import NoOp
        lr_0 = LR(C=0.09)
        lr_1 = LR(C=0.19)
        pipeline = PCA >> (lr_0 | NoOp >> lr_1)
        expected = \
"""from lale.lib.sklearn import PCA
from lale.lib.sklearn import LogisticRegression as LR
from lale.lib.lale import NoOp
import lale
lale.wrap_imported_operators()

lr_0 = LR(C=0.09)
lr_1 = LR(C=0.19)
pipeline = PCA >> (lr_0 | NoOp >> lr_1)"""
        self._roundtrip(expected, lale.pretty_print.to_string(pipeline))

    def test_autoai_libs_cat_encoder(self):
        from autoai_libs.transformers.exportable import CatEncoder
        import numpy as np
        from lale.lib.sklearn import LogisticRegression as LR
        cat_encoder = CatEncoder(categories='auto', dtype=np.float64, encoding='ordinal', handle_unknown='error')
        pipeline = cat_encoder >> LR()
        expected = \
"""from autoai_libs.transformers.exportable import CatEncoder
import numpy as np
from lale.lib.sklearn import LogisticRegression as LR
import lale
lale.wrap_imported_operators()

cat_encoder = CatEncoder(categories='auto', dtype=np.float64, encoding='ordinal', handle_unknown='error')
pipeline = cat_encoder >> LR()"""
        self._roundtrip(expected, lale.pretty_print.to_string(pipeline))

    def test_autoai_libs_numpy_replace_missing_values(self):
        from autoai_libs.transformers.exportable import NumpyReplaceMissingValues
        from lale.lib.sklearn import LogisticRegression as LR
        numpy_replace_missing_values = NumpyReplaceMissingValues(
            filling_values=float('nan'), missing_values=['?'])
        pipeline = numpy_replace_missing_values >> LR()
        expected = \
"""from autoai_libs.transformers.exportable import NumpyReplaceMissingValues
from lale.lib.sklearn import LogisticRegression as LR
import lale
lale.wrap_imported_operators()

numpy_replace_missing_values = NumpyReplaceMissingValues(filling_values=float('nan'), missing_values=['?'])
pipeline = numpy_replace_missing_values >> LR()"""
        self._roundtrip(expected, lale.pretty_print.to_string(pipeline))

    def test_autoai_libs_tam_1(self):
        from autoai_libs.cognito.transforms.transform_utils import TAM
        import autoai_libs.cognito.transforms.transform_extras
        import numpy as np
        from lale.lib.sklearn import LogisticRegression as LR
        tam = TAM(tans_class=autoai_libs.cognito.transforms.transform_extras.IsolationForestAnomaly, name='isoforestanomaly', col_names=['a', 'b', 'c'], col_dtypes=[np.dtype('float32'), np.dtype('float32'), np.dtype('float32')])
        pipeline = tam >> LR()
        expected = \
"""from autoai_libs.cognito.transforms.transform_utils import TAM
import autoai_libs.cognito.transforms.transform_extras
import numpy as np
from lale.lib.sklearn import LogisticRegression as LR
import lale
lale.wrap_imported_operators()

tam = TAM(tans_class=autoai_libs.cognito.transforms.transform_extras.IsolationForestAnomaly, name='isoforestanomaly', col_names=['a', 'b', 'c'], col_dtypes=[np.dtype('float32'), np.dtype('float32'), np.dtype('float32')])
pipeline = tam >> LR()"""
        self._roundtrip(expected, lale.pretty_print.to_string(pipeline))

    def test_autoai_libs_tam_2(self):
        from lale.lib.autoai_libs import TAM
        import numpy as np
        from lightgbm import LGBMClassifier
        from sklearn.decomposition import PCA
        from lale.operators import make_pipeline
        pca = PCA(copy=False)
        tam = TAM(tans_class=pca, name='pca', col_names=['a', 'b', 'c'], col_dtypes=[np.dtype('float32'), np.dtype('float32'), np.dtype('float32')])
        lgbm_classifier = LGBMClassifier(class_weight='balanced', learning_rate=0.18)
        pipeline = make_pipeline(tam, lgbm_classifier)
        expected = \
"""from lale.lib.autoai_libs import TAM
import sklearn.decomposition.pca
import numpy as np
from lightgbm import LGBMClassifier
from lale.operators import make_pipeline

tam = TAM(tans_class=sklearn.decomposition.pca.PCA(copy=False, iterated_power='auto', n_components=None, random_state=None,   svd_solver='auto', tol=0.0, whiten=False), name='pca', col_names=['a', 'b', 'c'], col_dtypes=[np.dtype('float32'), np.dtype('float32'), np.dtype('float32')])
lgbm_classifier = LGBMClassifier(class_weight='balanced', learning_rate=0.18)
pipeline = make_pipeline(tam, lgbm_classifier)"""
        self._roundtrip(expected, lale.pretty_print.to_string(pipeline, combinators=False))

    def test_autoai_libs_tam_3(self):
        import autoai_libs.cognito.transforms.transform_utils
        import lale.helpers
        import numpy as np
        import sklearn.cluster.hierarchical
        import sklearn.linear_model
        import sklearn.pipeline
        sklearn_pipeline = sklearn.pipeline.make_pipeline(
            autoai_libs.cognito.transforms.transform_utils.TAM(tans_class=sklearn.cluster.hierarchical.FeatureAgglomeration(affinity='euclidean', compute_full_tree='auto', connectivity=None, linkage='ward', memory=None, n_clusters=2, pooling_func=np.mean), name='featureagglomeration', col_names=['a', 'b', 'c'], col_dtypes=[np.dtype('float32'), np.dtype('float32'), np.dtype('float32')]),
            sklearn.linear_model.LogisticRegression(solver='liblinear', multi_class='ovr'))
        pipeline = lale.helpers.import_from_sklearn_pipeline(sklearn_pipeline)
        expected = \
"""from lale.lib.autoai_libs import TAM
from lale.lib.sklearn import FeatureAgglomeration
import numpy as np
from lale.lib.sklearn import LogisticRegression
import lale
lale.wrap_imported_operators()

tam = TAM(tans_class=FeatureAgglomeration(), name='featureagglomeration', col_names=['a', 'b', 'c'], col_dtypes=[np.dtype('float32'), np.dtype('float32'), np.dtype('float32')])
pipeline = tam >> LogisticRegression()"""
        self._roundtrip(expected, lale.pretty_print.to_string(pipeline))

    def test_autoai_libs_tam_4(self):
        import autoai_libs.cognito.transforms.transform_utils
        import lale.helpers
        import numpy as np
        import sklearn.cluster.hierarchical
        import sklearn.linear_model
        import sklearn.pipeline
        sklearn_pipeline = sklearn.pipeline.make_pipeline(
            autoai_libs.cognito.transforms.transform_utils.TAM(tans_class=sklearn.decomposition.PCA(), name='pca', col_names=['a', 'b', 'c'], col_dtypes=[np.dtype('float32'), np.dtype('float32'), np.dtype('float32')]),
            sklearn.linear_model.LogisticRegression(solver='liblinear', multi_class='ovr'))
        pipeline = lale.helpers.import_from_sklearn_pipeline(sklearn_pipeline, fitted=False)
        expected = \
"""from lale.lib.autoai_libs import TAM
from lale.lib.sklearn import PCA
import numpy as np
from lale.lib.sklearn import LogisticRegression
import lale
lale.wrap_imported_operators()

tam = TAM(tans_class=PCA(), name='pca', col_names=['a', 'b', 'c'], col_dtypes=[np.dtype('float32'), np.dtype('float32'), np.dtype('float32')])
pipeline = tam >> LogisticRegression()"""
        self._roundtrip(expected, lale.pretty_print.to_string(pipeline))
        import pandas as pd
        import numpy as np
        test = pd.DataFrame(np.random.randint(0,100,size=(15, 3)), columns=['a','b','c'], dtype=np.dtype('float32'))
        trained = pipeline.fit(test.to_numpy(), [0,1,1,0,0,0,1,1,1,1,0,0,1,0,1])
        trained.predict(test.to_numpy())

    def test_autoai_libs_ta1(self):
        from autoai_libs.cognito.transforms.transform_utils import TA1
        import numpy as np
        import autoai_libs.utils.fc_methods
        from lale.lib.sklearn import LogisticRegression as LR
        ta1 = TA1(fun=np.rint, name='round', datatypes=['numeric'], feat_constraints=[autoai_libs.utils.fc_methods.is_not_categorical], col_names=['a', 'b', 'c'], col_dtypes=[np.dtype('float32'), np.dtype('float32'), np.dtype('float32')])
        pipeline = ta1 >> LR()
        expected = \
"""from autoai_libs.cognito.transforms.transform_utils import TA1
import numpy as np
import autoai_libs.utils.fc_methods
from lale.lib.sklearn import LogisticRegression as LR
import lale
lale.wrap_imported_operators()

ta1 = TA1(fun=np.rint, name='round', datatypes=['numeric'], feat_constraints=[autoai_libs.utils.fc_methods.is_not_categorical], col_names=['a', 'b', 'c'], col_dtypes=[np.dtype('float32'), np.dtype('float32'), np.dtype('float32')])
pipeline = ta1 >> LR()"""
        self._roundtrip(expected, lale.pretty_print.to_string(pipeline))

    def test_autoai_libs_t_no_op(self):
        from lale.lib.autoai_libs import TNoOp
        from lightgbm import LGBMClassifier
        from lale.operators import make_pipeline
        t_no_op = TNoOp(name='no_action', datatypes='x', feat_constraints=[])
        lgbm_classifier = LGBMClassifier(class_weight='balanced', learning_rate=0.18)
        pipeline = make_pipeline(t_no_op, lgbm_classifier)
        expected = \
"""from lale.lib.autoai_libs import TNoOp
from lightgbm import LGBMClassifier
from lale.operators import make_pipeline

t_no_op = TNoOp(name='no_action', datatypes='x', feat_constraints=[])
lgbm_classifier = LGBMClassifier(class_weight='balanced', learning_rate=0.18)
pipeline = make_pipeline(t_no_op, lgbm_classifier)"""
        self._roundtrip(expected, lale.pretty_print.to_string(pipeline, combinators=False))

class TestToAndFromJSON(unittest.TestCase):
    def test_trainable_individual_op(self):
        self.maxDiff = None
        from lale.json_operator import to_json, from_json
        from lale.lib.sklearn import LogisticRegression as LR
        operator = LR(LR.solver.sag, C=0.1)
        json_expected = {
            'class': 'lale.lib.sklearn.logistic_regression.LogisticRegressionImpl',
            'state': 'trainable',
            'operator': 'LogisticRegression', 'label': 'LR',
            'documentation_url': 'https://lale.readthedocs.io/en/latest/modules/lale.lib.sklearn.logistic_regression.html',
            'hyperparams': {'C': 0.1, 'solver': 'sag'},
            'is_frozen_trainable': False}
        json = to_json(operator)
        self.assertEqual(json, json_expected)
        operator_2 = from_json(json)
        json_2 = to_json(operator_2)
        self.assertEqual(json_2, json_expected)

    def test_operator_choice(self):
        self.maxDiff = None
        from lale.json_operator import to_json, from_json
        from lale.lib.sklearn import PCA
        from lale.lib.sklearn import MinMaxScaler as Scl
        operator = PCA | Scl
        json_expected = {
          'class': 'lale.operators.OperatorChoice',
          'operator': 'OperatorChoice',
          'state': 'planned',
          'steps': {
            'pca': {
              'class': 'lale.lib.sklearn.pca.PCAImpl',
              'state': 'planned',
              'operator': 'PCA', 'label': 'PCA',
              'documentation_url': 'https://lale.readthedocs.io/en/latest/modules/lale.lib.sklearn.pca.html'},
            'scl': {
              'class': 'lale.lib.sklearn.min_max_scaler.MinMaxScalerImpl',
              'state': 'planned',
              'operator': 'MinMaxScaler', 'label': 'Scl',
              'documentation_url': 'https://lale.readthedocs.io/en/latest/modules/lale.lib.sklearn.min_max_scaler.html'}}}
        json = to_json(operator)
        self.assertEqual(json, json_expected)
        operator_2 = from_json(json)
        json_2 = to_json(operator_2)
        self.assertEqual(json_2, json_expected)

    def test_pipeline_1(self):
        self.maxDiff = None
        from lale.json_operator import to_json, from_json
        from lale.lib.lale import ConcatFeatures, NoOp
        from lale.lib.sklearn import LogisticRegression as LR
        from lale.lib.sklearn import PCA
        operator = (PCA & NoOp) >> ConcatFeatures >> LR
        json_expected = {
          'class': 'lale.operators.PlannedPipeline',
          'state': 'planned',
          'edges': [
              ['pca', 'concat_features'],
              ['no_op', 'concat_features'],
              ['concat_features', 'lr']],
          'steps': {
            'pca': {
              'class': 'lale.lib.sklearn.pca.PCAImpl',
              'state': 'planned',
              'operator': 'PCA', 'label': 'PCA',
              'documentation_url': 'https://lale.readthedocs.io/en/latest/modules/lale.lib.sklearn.pca.html'},
            'no_op': {
              'class': 'lale.lib.lale.no_op.NoOpImpl',
              'state': 'trained',
              'operator': 'NoOp', 'label': 'NoOp',
              'documentation_url': 'https://lale.readthedocs.io/en/latest/modules/lale.lib.lale.no_op.html',
              'hyperparams': None,
              'coefs': None,
              'is_frozen_trainable': True, 'is_frozen_trained': True},
            'concat_features': {
              'class': 'lale.lib.lale.concat_features.ConcatFeaturesImpl',
              'state': 'trained',
              'operator': 'ConcatFeatures', 'label': 'ConcatFeatures',
              'documentation_url': 'https://lale.readthedocs.io/en/latest/modules/lale.lib.lale.concat_features.html',
              'hyperparams': None,
              'coefs': None,
              'is_frozen_trainable': True, 'is_frozen_trained': True},
            'lr': {
              'class': 'lale.lib.sklearn.logistic_regression.LogisticRegressionImpl',
              'state': 'planned',
              'operator': 'LogisticRegression', 'label': 'LR',
              'documentation_url': 'https://lale.readthedocs.io/en/latest/modules/lale.lib.sklearn.logistic_regression.html'}}}
        json = to_json(operator)
        self.assertEqual(json, json_expected)
        operator_2 = from_json(json)
        json_2 = to_json(operator_2)
        self.assertEqual(json, json_2)

    def test_pipeline_2(self):
        from lale.lib.lale import NoOp
        from lale.lib.sklearn import Nystroem
        from lale.lib.sklearn import PCA
        from lale.lib.sklearn import LogisticRegression
        from lale.lib.sklearn import KNeighborsClassifier
        from lale.operators import make_choice, make_pipeline
        from lale.json_operator import to_json, from_json
        kernel_tfm_or_not =  make_choice(NoOp, Nystroem)
        tfm = PCA
        clf = make_choice(LogisticRegression, KNeighborsClassifier)
        operator = make_pipeline(kernel_tfm_or_not, tfm, clf)
        json = to_json(operator)
        operator_2 = from_json(json)
        json_2 = to_json(operator_2)
        self.assertEqual(json, json_2)

    def test_higher_order_1(self):
        from lale.lib.lale import Both
        from lale.lib.sklearn import PCA, Nystroem
        from lale.json_operator import from_json
        operator = Both(op1=PCA(n_components=2), op2=Nystroem)
        json_expected = {
          'class': 'lale.lib.lale.both.BothImpl',
          'state': 'trainable',
          'operator': 'Both', 'label': 'Both',
          'documentation_url': 'https://lale.readthedocs.io/en/latest/modules/lale.lib.lale.both.html',
          'hyperparams': {
            'op1': {'$ref': '../steps/pca'},
            'op2': {'$ref': '../steps/nystroem'}},
          'steps': {
            'pca': {
              'class': 'lale.lib.sklearn.pca.PCAImpl',
              'state': 'trainable',
              'operator': 'PCA', 'label': 'PCA',
              'documentation_url': 'https://lale.readthedocs.io/en/latest/modules/lale.lib.sklearn.pca.html',
              'hyperparams': {'n_components': 2},
              'is_frozen_trainable': False},
            'nystroem': {
              'class': 'lale.lib.sklearn.nystroem.NystroemImpl',
              'state': 'planned',
              'operator': 'Nystroem', 'label': 'Nystroem',
              'documentation_url': 'https://lale.readthedocs.io/en/latest/modules/lale.lib.sklearn.nystroem.html'}},
          'is_frozen_trainable': False}
        json = operator.to_json()
        self.assertEqual(json, json_expected)
        operator_2 = from_json(json)
        json_2 = operator_2.to_json()
        self.assertEqual(json, json_2)

    def test_higher_order_2(self):
        self.maxDiff = None
        from lale.lib.sklearn import VotingClassifier as Vote
        from lale.lib.sklearn import KNeighborsClassifier as KNN
        from lale.lib.sklearn import PCA
        from lale.lib.sklearn import LogisticRegression as LR
        from lale.json_operator import from_json
        operator = Vote(estimators=[('knn',KNN), ('pipeline',PCA()>>LR)],
                        voting='soft')
        json_expected = {
          'class': 'lale.lib.sklearn.voting_classifier.VotingClassifierImpl',
          'state': 'trainable',
          'operator': 'VotingClassifier',
          'is_frozen_trainable': True,
          'label': 'Vote',
          'documentation_url': 'https://lale.readthedocs.io/en/latest/modules/lale.lib.sklearn.voting_classifier.html',
          'hyperparams': {
            'estimators': [
              ('knn', {'$ref': '../steps/knn'}),
              ('pipeline', {'$ref': '../steps/pipeline'})],
            'voting': 'soft'},
          'steps': {
            'knn': {
              'class': 'lale.lib.sklearn.k_neighbors_classifier.KNeighborsClassifierImpl',
              'state': 'planned',
              'operator': 'KNeighborsClassifier', 'label': 'KNN',
              'documentation_url': 'https://lale.readthedocs.io/en/latest/modules/lale.lib.sklearn.k_neighbors_classifier.html'},
            'pipeline': {
              'class': 'lale.operators.PlannedPipeline',
              'state': 'planned',
              'edges': [['pca', 'lr']],
              'steps': {
                'pca': {
                  'class': 'lale.lib.sklearn.pca.PCAImpl',
                  'state': 'trainable',
                  'operator': 'PCA', 'label': 'PCA',
                  'documentation_url': 'https://lale.readthedocs.io/en/latest/modules/lale.lib.sklearn.pca.html',
                  'hyperparams': {},
                  'is_frozen_trainable': False},
                'lr': {
                  'class': 'lale.lib.sklearn.logistic_regression.LogisticRegressionImpl',
                  'state': 'planned',
                  'operator': 'LogisticRegression', 'label': 'LR',
                  'documentation_url': 'https://lale.readthedocs.io/en/latest/modules/lale.lib.sklearn.logistic_regression.html'}}}}}
        json = operator.to_json()
        self.assertEqual(json, json_expected)
        operator_2 = from_json(json)
        json_2 = operator_2.to_json()
        self.assertEqual(json, json_2)

    def test_nested(self):
        self.maxDiff = None
        from lale.json_operator import to_json, from_json
        from lale.lib.lale import NoOp
        from lale.lib.sklearn import LogisticRegression as LR
        from lale.lib.sklearn import PCA
        operator = PCA >> (LR(C=0.09) | NoOp >> LR(C=0.19))
        json_expected = {
          'class': 'lale.operators.PlannedPipeline',
          'state': 'planned',
          'edges': [['pca', 'choice']],
          'steps': {
            'pca': {
              'class': 'lale.lib.sklearn.pca.PCAImpl',
              'state': 'planned',
              'operator': 'PCA', 'label': 'PCA',
              'documentation_url': 'https://lale.readthedocs.io/en/latest/modules/lale.lib.sklearn.pca.html'},
            'choice': {
              'class': 'lale.operators.OperatorChoice',
              'state': 'planned',
              'operator': 'OperatorChoice',
              'steps': {
                'lr_0': {
                  'class': 'lale.lib.sklearn.logistic_regression.LogisticRegressionImpl',
                  'state': 'trainable',
                  'operator': 'LogisticRegression', 'label': 'LR',
                  'documentation_url': 'https://lale.readthedocs.io/en/latest/modules/lale.lib.sklearn.logistic_regression.html',
                  'hyperparams': {'C': 0.09},
                  'is_frozen_trainable': False},
                'pipeline_1': {
                  'class': 'lale.operators.TrainablePipeline',
                  'state': 'trainable',
                  'edges': [['no_op', 'lr_1']],
                  'steps': {
                    'no_op': {
                      'class': 'lale.lib.lale.no_op.NoOpImpl',
                      'state': 'trained',
                      'operator': 'NoOp', 'label': 'NoOp',
                      'documentation_url': 'https://lale.readthedocs.io/en/latest/modules/lale.lib.lale.no_op.html',
                      'hyperparams': None,
                      'coefs': None,
                      'is_frozen_trainable': True, 'is_frozen_trained': True},
                    'lr_1': {
                      'class': 'lale.lib.sklearn.logistic_regression.LogisticRegressionImpl',
                      'state': 'trainable',
                      'operator': 'LogisticRegression', 'label': 'LR',
                      'documentation_url': 'https://lale.readthedocs.io/en/latest/modules/lale.lib.sklearn.logistic_regression.html',
                      'hyperparams': {'C': 0.19},
                      'is_frozen_trainable': False}}}}}}}
        json = to_json(operator)
        self.assertEqual(json, json_expected)
        operator_2 = from_json(json)
        json_2 = to_json(operator_2)
        self.assertEqual(json, json_2)

if __name__ == '__main__':
    unittest.main()


text:
'''
Functionality for running the genetic algorithm
'''

import os
import pickle
import random
import numpy as np
import pandas as pd

# User defined functions
import strats as strat_lib
from utils import tickers, strategy

# For type hinting
from typing import Tuple
from numpy import array as np_arr
from pandas import DataFrame as pandasDF

def main(ga_config: dict) -> dict:
    '''
    Main running function for the genetic algorithm

    Parameters
    ----------
    ga_config : dict
        Config params for the algorithm

    Returns
    -------
    best_strat : dict
        The optimised strategy parameters
    '''

    # Initialise all the parameters needed to start the evolution
    data, strats, fit_arr, strats_to_calc = init_ga(ga_config)

    # This gathers the number of strategies to change on each evolution
    perc_change = int((1-ga_config['keep perc'])*ga_config['num strats'])

    # Start the optimisation procedure
    for evl in range(0, ga_config['num evolutions']):

        # Calculate the fitness for the strategies
        fit_arr = get_fitness(data,
                              ga_config,
                              strats,
                              fit_arr,
                              strats_to_calc,
                              )

        # Rank the strategies, and select the strategies to change
        ranks = fit_arr[fit_arr[:, 1].argsort()]
        good_strats = ranks[perc_change:, 0].astype(np.int32)
        bad_strats = ranks[:perc_change, 0].astype(np.int32)

        # Split the bad strategies into 3 approx equal sets to make modifications
        splits = np.array_split(bad_strats, 3)

        # Replace bad strategies with random new ones
        for strat in splits[0]:
            strats[str(strat)] = get_random_strat(ga_config)

        # Add random perturbations to good strategies
        for strat in splits[1]:
            rand_strat = str(random.choice(good_strats))
            strats[str(strat)] = perturb_strat(strats[rand_strat].copy(),
                                               ga_config,
                                               )

        # Breed good strategies to make another
        for strat in splits[2]:
            strats[str(strat)] = breed_good_strats(good_strats,
                                                   strats,
                                                   ga_config,
                                                   )

        # Tell the optimiser which strats have been changed to calculate the
        # fitness function of. This saves time on recalculating the good strats
        strats_to_calc = bad_strats

        # Print out evolution stats
        print(f'\nEvolution {evl}')
        for count, strat in enumerate(np.flipud(good_strats[-5:])):
            print(str(count) + '. Strategy: ' +  str(strat) +
                  ', ' + ga_config['fitness'] + ': ' +
                  str(fit_arr[strat, 1])
                  )
        print('----------------------------------------------')

    # Print the final results and save to json
    best_strat = strats[str(good_strats[-1])]
    best_strat['ticker opt'] = data['ticker'].unique().tolist()
    print_and_save(best_strat,
                   ga_config)

    return best_strat

def out_sample_test(strat: dict,
                    ga_config: dict):
    '''
    Perform in, and out of sample testing for evaluating the outcome of the ga
    '''

    # Get tickers to perform the out of sample testing
    out_sample_tickers = tickers.get_tickers_exc_sample(ga_config['num tickers test'],
                                                        strat['ticker opt'],
                                                        )

    # In sample test
    print('\n----------------------------------------------')
    print('In sample testing results: ')
    ticker_test(strat['ticker opt'],
                strat,
                ga_config,
                )
    print('----------------------------------------------')

    # Out of sample test
    print('\n----------------------------------------------')
    print('Out of sample testing results: ')
    ticker_test(out_sample_tickers,
                strat,
                ga_config,
                )
    print('----------------------------------------------')

    return

def ticker_test(tickers: list,
                strat: dict,
                ga_config: dict):
    '''
    Given a set of tickers, run the buy/sell algorithm and produce average
    statistics for the set of tickers
    '''

    # Key performance metric storage lists
    win_rate = []
    avg_profit = []
    mean_hold = []
    max_hold = []
    num_trades = []

    for ticker in tickers:
        df = pd.read_csv(f'data/{ticker}.csv')
        df = strategy.add_strat_cols(df, strat, ga_config['strat'])
        _, stats = strategy.run_strategy(df, strat, ga_config['strat'])

        win_rate.append(stats['win rate'])
        avg_profit.append(stats['avg profit'])
        mean_hold.append(stats['mean hold'])
        max_hold.append(stats['max hold'])
        num_trades.append(stats['number of trades'])

    # Print the summaries
    print('Average win rate: ', np.mean(win_rate))
    print('Lowest win rate: ', np.min(win_rate))
    print('Average profit: ', np.mean(avg_profit))
    print('Maximum hold time: ', np.max(max_hold))
    print('Mean hold time: ', np.mean(mean_hold))
    print('Mean number of trades: ', np.mean(num_trades))

    return

def print_and_save(strat: dict,
                   ga_config: dict):
    '''
    Print the optimal parameters and save to a pickle file
    '''

    print('Most optimal parameters are: ')
    for k, v in strat.items():
        if k != 'ticker opt':
            print(k + ': ' + str(v))

    with open('ga/strategies/' + ga_config['save name'] + '.pkl', 'wb') as f:
        pickle.dump(strat, f)

    return

def init_ga(ga_config: dict) -> Tuple[pandasDF, dict, np_arr, np_arr]:
    '''
    Initialise any parameters and data needed for the genetic algorithm

    Parameters
    ----------
    ga_config : dict
        Config controls for the genetic algorithm

    Returns
    -------
    data : pandasDF
        Price data for all tickers we are optimising
    strats : dict
        A random set of strategies
    fit_arr : np_arr
        An array to store the fitness values in
    strats_to_calc : np_arr
        An array to indicate which strategies to calculate the fitness for
    '''

    # Check if the filing structure is correct
    check_folder()

    # Get a random set of tickers, and load in the data we will optimise
    ticker_list = tickers.get_random_tickers(ga_config['num tickers'])
    data = get_price_data(ticker_list)

    # Initialise with a set of random strategies
    strats = {f'{n}': get_random_strat(ga_config)
              for n in range(0, ga_config['num strats'])}

    # Initialise an empty array to store the fitness values in
    fit_arr = np.vstack((np.arange(0, ga_config['num strats']),
                         np.zeros(ga_config['num strats']),
                         ),
                        ).T

    # Initialise the array to determine which strategies to calculate the
    # fitness for. In this case its all of them, but in the algo we only need
    # to calculate the fresh strategies
    strats_to_calc = np.arange(0, ga_config['num strats'])

    return data, strats, fit_arr, strats_to_calc

def get_fitness(data: pandasDF,
                ga_config: dict,
                strats: dict,
                fit_arr: np_arr,
                strats_to_calc: np_arr) -> np_arr:
    '''
    Calculate the fitness function for each strategy defined in strats_to_calc.
    This runs the buy/sell algorithm on each ticker, and then uses the mean
    of all results to generate the fitness value for each strategy.

    Parameters
    ----------
    data : pandasDF
        Price data for all tickers we are optimising
    ga_config : dict
        Config controls for the genetic algo
    strats : dict
        All strategies generated by the genetic algo
    fit_arr : np_arr
        An array to store the fitness results for each strategy
    strats_to_calc : np_arr
        Which strategies to calculate the fitness value for

    Returns
    -------
    fit_arr : np_arr
        The recalculated fitness values for the new strategies
    '''

    for strat in strats_to_calc:

        # Initialise a list to store the result of this strategy
        res = []

        for ticker in data['ticker'].unique():

            # Add the strategy columns to the dataframe
            df_strat = data[data['ticker'] == ticker].reset_index().drop(columns = ['index'])
            df_strat = strategy.add_strat_cols(df_strat,
                                               strats[str(strat)],
                                               ga_config['strat'],
                                               )

            # Run the buy/sell algorithm and produce the statistics
            _, stats = strategy.run_strategy(df_strat,
                                             strats[str(strat)],
                                             ga_config['strat'],
                                             )

            # We want to strongly encourage the algorithm to not take any strat
            # which performes trades less than min_trades, this prevents some
            # curve fitting to very rare events
            if stats['number of trades'] > ga_config['min trades']:
                res.append(stats[ga_config['fitness']])
            else:
                res.append(-100)

        # Find the average result for this strategy
        fit_arr[strat, 1] = np.mean(res)

    return fit_arr

def get_price_data(tickers: list) -> pandasDF:
    '''
    Generate a dataframe which contains all the price data for each ticker

    Parameters
    ----------
    tickers : list
        List of tickers to optimise the strategy for

    Returns
    -------
    df : pandasDF
        A dataframe of all price data for each ticker.
    '''

    dfs = []
    for ticker in tickers:
        df = pd.read_csv(f'data/{ticker}.csv')
        df['ticker'] = ticker
        dfs.append(df)

    return pd.concat(dfs)

def breed_good_strats(good_strats: np_arr,
                      strats: dict,
                      ga_config: dict) -> dict:
    '''
    Taking parameters from good strategies, breed a new one.
    '''

    new_strat = {}
    for param in strats['0'].keys():
        rand_strat = random.choice(good_strats)
        new_strat[param] = strats[str(rand_strat)][param]

    return check_params(new_strat, ga_config)

def get_random_strat(ga_config: dict) -> dict:
    '''
    Get a randomly generated strategy
    '''
    if ga_config['strat'] == 'simple ma crossover':
        return strat_lib.ma_crossover.get_random_strat(ga_config)
    elif ga_config['strat'] == 'simple bollinger band':
        return strat_lib.boll_band.get_random_strat(ga_config)
    elif ga_config['strat'] == 'bollinger squeeze':
        return strat_lib.boll_squeeze.get_random_strat(ga_config)

def check_params(strat: dict,
                 ga_config: dict) -> dict:
    '''
    Check if the parameters for the strategy make sense, adjust if they dont
    '''
    if ga_config['strat'] == 'simple ma crossover':
        return strat_lib.ma_crossover.check_params(strat, ga_config)
    elif ga_config['strat'] == 'simple bollinger band':
        return strat_lib.boll_band.check_params(strat, ga_config)
    elif ga_config['strat'] == 'bollinger squeeze':
        return strat_lib.boll_squeeze.check_params(strat, ga_config)

def perturb_strat(strat: dict,
                  ga_config: dict) -> dict:
    '''
    Perturb the parameters of the strategy slightly to generate a new strategy
    '''
    if ga_config['strat'] == 'simple ma crossover':
        strat = strat_lib.ma_crossover.perturb_strat(strat, ga_config)
    elif ga_config['strat'] == 'simple bollinger band':
        strat = strat_lib.boll_band.perturb_strat(strat, ga_config)
    elif ga_config['strat'] == 'bollinger squeeze':
        return strat_lib.boll_squeeze.perturb_strat(strat, ga_config)

    return check_params(strat, ga_config)

def check_folder():
    '''
    Check if the ga strategies folder exists, if not, create it
    '''
    if not os.path.isdir('ga/strategies'):
        os.mkdir('ga/strategies')

ext: py

reason: import numpy,from numpy


text:
# -*- coding: utf-8 -*-

# (C) Copyright 2020 IBM. All Rights Reserved.
#
# This code is licensed under the Apache License, Version 2.0. You may
# obtain a copy of this license in the LICENSE.txt file in the root directory
# of this source tree or at http://www.apache.org/licenses/LICENSE-2.0.
#
# Any modifications or derivative works of this code must retain this
# copyright notice, and modified files need to carry a notice indicating
# that they have been altered from the originals.

"""High level analog tiles (base)."""

from typing import Dict, Generic, List, Optional, Tuple, TypeVar, Union

from collections import OrderedDict

from numpy import concatenate, expand_dims
from torch import Tensor, stack
from torch import device as torch_device
from torch.autograd import no_grad

from aihwkit.simulator.rpu_base import tiles

from aihwkit.exceptions import TileError

RPUConfigGeneric = TypeVar('RPUConfigGeneric')

class BaseTile(Generic[RPUConfigGeneric]):
    """Base class for tiles.

    Args:
        out_size: output size
        in_size: input size
        rpu_config: resistive processing unit configuration.
        bias: whether to add a bias column to the tile.
        in_trans: Whether to assume an transposed input (batch first)
        out_trans: Whether to assume an transposed output (batch first)
    """
    # pylint: disable=too-many-instance-attributes

    is_cuda = False

    def __init__(
            self,
            out_size: int,
            in_size: int,
            rpu_config: RPUConfigGeneric,
            bias: bool = True,
            in_trans: bool = False,
            out_trans: bool = False
    ):
        self.out_size = out_size
        self.in_size = in_size
        self.rpu_config = rpu_config
        self.bias = bias
        self.in_trans = in_trans
        self.out_trans = out_trans

        # Only used for indexed.
        self.image_sizes = []  # type: List[int]

        x_size = in_size + 1 if self.bias else in_size
        d_size = out_size

        # Cuda tiles are assumed to init `self.tile` manually.
        if self.is_cuda:
            self.tile = None  # type: Union[tiles.FloatingPointTile, tiles.AnalogTile]
        else:
            self.tile = self._create_simulator_tile(x_size, d_size, rpu_config)
            self.tile.set_learning_rate(0.01)
            self.tile.set_weights_uniform_random(-0.01, 0.01)

        self.device = torch_device('cpu')

    def __getstate__(self) -> Dict:
        """Get the state for pickling.

        This method removes the ``tile`` member, as the binding Tiles are not
        serializable.
        """
        current_dict = self.__dict__.copy()

        current_dict['analog_tile_weights'] = self.tile.get_weights()
        # Store the hidden parameters as a numpy array, as storing it as
        # Tensor causes issues in PyTorch 1.5.
        current_dict['analog_tile_hidden_parameters'] = \
            self.tile.get_hidden_parameters().data.numpy()
        current_dict.pop('tile')
        current_dict.pop('stream', None)

        return current_dict

    def __setstate__(self, state: Dict) -> None:
        """Set the state after unpickling.

        This method recreates the ``tile`` member, creating a new one from
        scratch, as the binding Tiles are not serializable.
        """
        current_dict = state.copy()
        weights = current_dict.pop('analog_tile_weights')
        hidden_parameters = current_dict.pop('analog_tile_hidden_parameters')

        self.__dict__.update(current_dict)

        x_size = self.in_size + 1 if self.bias else self.in_size
        d_size = self.out_size

        # Recreate the tile.
        self.tile = self._create_simulator_tile(x_size, d_size, self.rpu_config)
        self.tile.set_hidden_parameters(Tensor(hidden_parameters))
        self.tile.set_weights(weights)

    def _create_simulator_tile(
            self,
            x_size: int,
            d_size: int,
            rpu_config: RPUConfigGeneric
    ) -> Union[tiles.FloatingPointTile, tiles.AnalogTile]:
        """Create a simulator tile.

        Args:
            x_size: input size
            d_size: output size
            rpu_config: resistive processing unit configuration

        Returns:
            a simulator tile based on the specified configuration.
        """
        raise NotImplementedError

    def set_weights(
            self,
            weights: Tensor,
            biases: Optional[Tensor] = None,
            realistic: bool = False,
            n_loops: int = 10
    ) -> None:
        """Set the tile weights (and biases).

        Sets the internal tile weights to the specified values, and also the
        internal tile biases if the tile was set to use bias (via
        ``self.bias``).

        Note:
            By default this is **not** hardware realistic. You can set the
            ``realistic`` parameter to ``True`` for a realistic transfer.

        Args:
            weights: ``[out_size, in_size]`` weight matrix.
            biases: ``[out_size]`` bias vector. This parameter is required if
                ``self.bias`` is ``True``, and ignored otherwise.
            realistic: whether to use the forward and update pass to
                program the weights iteratively, using
                :meth:`set_weights_realistic`.
            n_loops: number of times the columns of the weights are set in a
                closed-loop manner.
                A value of ``1`` means that all columns in principle receive
                enough pulses to change from ``w_min`` to ``w_max``.
        """
        # Prepare the array expected by the pybind function, appending the
        # biases row if needed.
        weights_numpy = weights.clone().detach().cpu().numpy()

        if self.bias:
            # Create a ``[out_size, in_size (+ 1)]`` matrix.
            if biases is None:
                raise ValueError("Analog tile has a bias, but no bias given")

            biases_numpy = expand_dims(biases.clone().detach().cpu().numpy(), 1)
            combined_weights = concatenate([weights_numpy, biases_numpy], axis=1)
        else:
            # Use only the ``[out_size, in_size]`` matrix.
            combined_weights = weights_numpy

        if realistic:
            return self.tile.set_weights_realistic(combined_weights, n_loops)

        return self.tile.set_weights(combined_weights)

    def get_weights(
            self,
            realistic: bool = False
    ) -> Tuple[Tensor, Optional[Tensor]]:
        """Get the tile weights (and biases).

        Gets the tile weights and extracts the mathematical weight
        matrix and biases (if present, by determined by the ``self.bias``
        parameter).

        Note:
            By default this is **not** hardware realistic. Use set
            ``realistic`` to True for a realistic transfer.

        Args:
            realistic: Whether to use the forward pass to
              read out the tile weights iteratively, using
              :meth:`get_weights_realistic`

        Returns:
           a tuple where the first item is the ``[out_size, in_size]`` weight
           matrix; and the second item is either the ``[out_size]`` bias vector
           or ``None`` if the tile is set not to use bias.
        """
        # Retrieve the internal weights (and potentially biases) matrix.
        if realistic:
            combined_weights = self.tile.get_weights_realistic()
        else:
            combined_weights = self.tile.get_weights()

        # Split the internal weights (and potentially biases) matrix.
        if self.bias:
            # combined_weights is [out_size, in_size (+ 1)].
            weights = Tensor(combined_weights[:, :-1])
            biases = Tensor(combined_weights[:, -1])
        else:
            # combined_weights is [out_size, in_size].
            weights = Tensor(combined_weights)
            biases = None

        return weights.to(self.device), biases.to(self.device) if self.bias else None

    def set_learning_rate(self, learning_rate: float) -> None:
        """Set the tile learning rate.

        Set the tile learning rate to ``-learning_rate``. Note that the
        learning rate is always taken to be negative (because of the meaning in
        gradient descent) and positive learning rates are not supported.

        Args:
            learning_rate: the desired learning rate.
        """
        return self.tile.set_learning_rate(learning_rate)

    def get_learning_rate(self) -> float:
        """Return the tile learning rate.

        Returns:
            float: the tile learning rate.
        """
        return self.tile.get_learning_rate()

    def decay_weights(self, alpha: float = 1.0) -> None:
        """Decays the weights once.

        Args:
           alpha: additional decay scale (such as LR). The base decay
              rate is set during tile init.
        """
        return self.tile.decay_weights(alpha)

    def diffuse_weights(self) -> None:
        """Diffuses the weights once.

        The base diffusion rate is set during tile init.
        """
        return self.tile.diffuse_weights()

    def reset_columns(self, start_column_idx: int = 0, num_columns: int = 1,
                      reset_prob: float = 1.0) -> None:
        r"""Reset (a number of) columns.

        Resets the weights with device-to-device and cycle-to-cycle
        variability (depending on device type), typically:

        .. math::
            W_{ij} = \xi*\sigma_\text{reset} + b^\text{reset}_{ij}

        Args:
            start_column_idx: a start index of columns (0..x_size-1)
            num_columns: how many consecutive columns to reset (with circular warping)
            reset_prob: individual probability of reset.

        The reset parameter are set during tile init.
        """
        return self.tile.reset_columns(start_column_idx, num_columns, reset_prob)

    def cuda(
            self,
            device: Optional[Union[torch_device, str, int]] = None
    ) -> 'BaseTile':
        """Return a copy of this tile in CUDA memory."""
        raise NotImplementedError

    @no_grad()
    def forward(self, x_input: Tensor, is_test: bool = False) -> Tensor:
        """Perform the forward pass.

        Args:
            x_input: ``[N, in_size]`` tensor. If ``in_trans`` is set, transposed.
            is_test: whether to assume testing mode.

        Returns:
            torch.Tensor: ``[N, out_size]`` tensor. If ``out_trans`` is set, transposed.
        """
        # We use no-grad as we do it explicitly in the optimizer.
        return self.tile.forward(x_input, self.bias,
                                 self.in_trans, self.out_trans, is_test)

    def backward(self, d_input: Tensor) -> Tensor:
        """Perform the backward pass.

        Args:
            d_input: ``[N, out_size]`` tensor. If ``out_trans`` is set, transposed.

        Returns:
            torch.Tensor: ``[N, in_size]`` tensor. If ``in_trans`` is set, transposed.
        """
        return self.tile.backward(d_input, self.bias, self.out_trans, self.in_trans)

    def update(self, x_input: Tensor, d_input: Tensor) -> None:
        """Perform the update pass.

        Args:
            x_input: ``[N, in_size]`` tensor. If ``in_trans`` is set, transposed.
            d_input: ``[N, out_size]`` tensor. If ``out_trans`` is set, transposed.
        """
        return self.tile.update(x_input, d_input, self.bias,
                                self.in_trans, self.out_trans)

    def get_hidden_parameters(self) -> OrderedDict:
        """Get the hidden parameters of the tile.

        Returns:
            Ordered dictionary of hidden parameter tensors.
        """
        names = self.tile.get_hidden_parameter_names()
        hidden_parameters = self.tile.get_hidden_parameters().detach_()

        ordered_parameters = OrderedDict()
        for idx, name in enumerate(names):
            ordered_parameters[name] = hidden_parameters[idx].clone()

        return ordered_parameters

    def set_hidden_parameters(self, ordered_parameters: OrderedDict) -> None:
        """Set the hidden parameters of the tile.

        Args:
            ordered_parameters: Ordered dictionary of hidden parameter tensors.
        """
        if len(ordered_parameters) == 0:
            return

        hidden_parameters = stack(list(ordered_parameters.values()), dim=0)
        self.tile.set_hidden_parameters(hidden_parameters)

    def get_hidden_update_index(self) -> int:
        """Get the current updated device index of the hidden devices.

        Usually this is 0 as only one device is present per
        cross-point for many tile RPU configs. However, some RPU
        configs maintain internally multiple devices per cross-point
        (e.g. :class:`~aihwkit.simulator.config.devices.VectorUnitCell`).

        Returns:
            The next mini-batch updated device index.

        Note:
            Depending on the update and learning policy implemented
            in the tile, updated devices might switch internally as
            well.
        """
        return self.tile.get_hidden_update_index()

    def set_hidden_update_index(self, index: int) -> None:
        """set the current updated hidden device index.

        Usually this is ignored and fixed to 0 as only one device is
        present per cross-point. Other devices, might not allow
        explicit setting as it would interfere with the implemented
        learning However rule. However, some tiles have internally
        multiple devices per cross-point (eg. unit cell) that can be
        chosen depending on the update policy.

        Args:
            index: device index to be updated in the next mini-batch

        Note:
            Depending on the update and learning policy implemented
            in the tile, updated devices might switch internally as
            well.
        """
        self.tile.set_hidden_update_index(index)

    def set_indexed(self, indices: Tensor, image_sizes: List) -> None:
        """Sets the index matrix for convolutions ans switches to
        indexed forward/backward/update versions.

        Args:
            indices : torch.tensor with int indices
            image_sizes: [C_in, H_in, W_in, H_out, W_out] sizes
        """
        if len(image_sizes) != 5:
            raise ValueError('image_sizes expects 5 sizes [C_in, H_in, W_in, H_out, W_out]')

        if self.in_trans or self.out_trans:
            raise TileError('Transposed indexed versions not supported (assumes NCHW)')

        self.image_sizes = image_sizes
        self.tile.set_matrix_indices(indices)

    @no_grad()
    def forward_indexed(self, x_input: Tensor, is_test: bool = False) -> Tensor:
        """Perform the forward pass for convolutions.

        Args:
            x_input: ``[N, in_size]`` tensor. If ``in_trans`` is set, transposed.
            is_test: whether to assume testing mode.

        Returns:
            torch.Tensor: ``[N, out_size]`` tensor. If ``out_trans`` is set, transposed.
        """
        if not self.image_sizes:
            raise TileError('self.image_sizes is not initialized. Please use '
                            'set_indexed()')

        _, _, _, height_out, width_out = self.image_sizes
        return self.tile.forward_indexed(x_input, height_out, width_out, is_test)

    def backward_indexed(self, d_input: Tensor) -> Tensor:
        """Perform the backward pass for convolutions.

        Args:
            d_input: ``[N, out_size]`` tensor. If ``out_trans`` is set, transposed.

        Returns:
            torch.Tensor: ``[N, in_size]`` tensor. If ``in_trans`` is set, transposed.
        """
        channel_in, height_in, width_in, _, _ = self.image_sizes
        return self.tile.backward_indexed(d_input, channel_in, height_in, width_in)

    def update_indexed(self, x_input: Tensor, d_input: Tensor) -> None:
        """Perform the update pass for convolutions.

        Args:
            x_input: ``[N, in_size]`` tensor. If ``in_trans`` is set, transposed.
            d_input: ``[N, out_size]`` tensor. If ``out_trans`` is set, transposed.
        """
        return self.tile.update_indexed(x_input, d_input)

    @no_grad()
    def post_update_step(self) -> None:
        """Operators that need to be called once per mini-batch."""
        if self.rpu_config.device.requires_diffusion():  # type: ignore
            self.diffuse_weights()
        if self.rpu_config.device.requires_decay():  # type: ignore
            self.decay_weights()

max_stars_repo_name: filemaster/aihwkit

max_issues_repo_name: filemaster/aihwkit

max_forks_repo_name: filemaster/aihwkit

reason: from numpy


text:
# Plot a random field of vectors and save a screenshot of it.
#
import numpy as np
import pyvista
cent = np.random.random((10, 3))
direction = np.random.random((10, 3))
plotter = pyvista.Plotter()
_ = plotter.add_arrows(cent, direction, mag=2)
plotter.show()

reason: import numpy


text:
#%%
import tensorflow as tf
import tensorflow.keras as K
from tensorflow.keras import layers
from tensorflow.keras import preprocessing
print('TensorFlow version:', tf.__version__)
print('Eager Execution Mode:', tf.executing_eagerly())
print('available GPU:', tf.config.list_physical_devices('GPU'))
from tensorflow.python.client import device_lib
print('==========================================')
print(device_lib.list_local_devices())
# tf.debugging.set_log_device_placement(False)
#%%
from tqdm import tqdm
import matplotlib.pyplot as plt
import numpy as np
import os
import time
# os.chdir('/home/jeon/Desktop/an/gan')
os.chdir(r'D:\gan')

from modules import model
#%%
PARAMS = {
    "batch_size": 10,
    "epochs": 10000,
    "learning_rate": 0.0001,
    "data": "celeba64",
    "data_dim": 64,
    "channel": 3,
    "latent_dim": 512,
    "n_layer": 5,
    "embedding_dim": 32,
    "embedding_dim_mult": (1, 2, 2, 2, 4),
    "num_res_blocks": 3,
    "attn_resolutions": (32, ),
    "ema_rate": 0.999
}

print(PARAMS)
#%%
import tensorflow_datasets as tfds
(train_images, test_images), info = tfds.load('celeb_a', split=['train', 'test'], shuffle_files=False, with_info=True)
# (train_images, test_images), info = tfds.load('celeb_a', split=['train[:10%]', 'test[:10%]'], shuffle_files=False, with_info=True)

print('image shape:', info.features['image'].shape)
PARAMS['data_dim'] = 64
attr_dict = {x:i for i, x in enumerate(info.features["attributes"].keys())}
print('attribute:', len(attr_dict))
PARAMS['attr_num'] = len(attr_dict)

eye = np.eye(len(attr_dict))
@tf.function
def normalize(data):
    attr = data['attributes']
    idx = [attr_dict.get(x) for x in attr.keys() if attr.get(x) is True]
    tag = np.sum(eye[idx], axis=0) # one-hot vector
    image = (tf.cast(data['image'], tf.float32) - 127.5) / 127.5
    image = tf.image.resize(image, [PARAMS['data_dim'], PARAMS['data_dim']], method='nearest')
    return image, tag

train_dataset = train_images.map(normalize).shuffle(len(train_images), reshuffle_each_iteration=True).batch(PARAMS['batch_size'])
test_dataset = test_images.map(normalize).shuffle(len(test_images), reshuffle_each_iteration=True).batch(PARAMS['batch_size'])
#%%
encoder = model.build_encoder(PARAMS)
encoder.summary()
generator = model.build_generator(PARAMS)
generator.summary()
img_discriminator = model.build_image_discriminator(PARAMS)
img_discriminator.summary()
z_discriminator = model.build_z_discriminator(PARAMS)
z_discriminator.summary()
#%%
cross_entropy = K.losses.BinaryCrossentropy(from_logits=True)
def img_discriminator_loss(real_decision, fake_decision1, fake_decision2):
    real_loss = cross_entropy(tf.ones_like(real_decision), real_decision)
    fake_loss1 = cross_entropy(tf.zeros_like(fake_decision1), fake_decision1)
    fake_loss2 = cross_entropy(tf.zeros_like(fake_decision2), fake_decision2)
    total_loss = real_loss + fake_loss1 + fake_loss2
    return total_loss

def z_discriminator_loss(real_decision, fake_decision):
    real_loss = cross_entropy(tf.ones_like(real_decision), real_decision)
    fake_loss = cross_entropy(tf.zeros_like(fake_decision), fake_decision)
    total_loss = real_loss + fake_loss
    return total_loss

def generator_loss(fake_decision):
    return cross_entropy(tf.ones_like(fake_decision), fake_decision)

def encoder_loss(fake_decision):
    return cross_entropy(tf.ones_like(fake_decision), fake_decision)
#%%
encoder_optimizer = K.optimizers.Adam(0.001)
generator_optimizer = K.optimizers.Adam(0.001)
img_discriminator_optimizer = K.optimizers.Adam(0.001)
z_discriminator_optimizer = K.optimizers.Adam(0.001)

ema = tf.train.ExponentialMovingAverage(decay=PARAMS['ema_rate'])
#%%
@tf.function
def train_one_step(x_batch, y_batch):
    with tf.GradientTape() as gen_tape, tf.GradientTape() as img_dis_tape, tf.GradientTape() as z_dis_tape, tf.GradientTape() as enc_tape:
        z = tf.random.normal([PARAMS['batch_size'], PARAMS['latent_dim']])
        recon_z = encoder(x_batch)
        noise = np.random.uniform(0.0, 1.0, [PARAMS['batch_size'], PARAMS['data_dim'], PARAMS['data_dim'], 1]).astype('float32')
        generated_images = generator([z]*PARAMS['n_layer'] + [noise] + [y_batch])
        reconstructed_images = generator([recon_z]*PARAMS['n_layer'] + [noise] + [y_batch])

        z_real_decision = z_discriminator(z, training=True)
        z_fake_decision = z_discriminator(recon_z, training=True)

        img_real_decision = img_discriminator(x_batch, training=True)
        img_fake_decision1 = img_discriminator(generated_images, training=True)
        img_fake_decision2 = img_discriminator(reconstructed_images, training=True)

        img_dis_loss = img_discriminator_loss(img_real_decision, img_fake_decision1, img_fake_decision2)
        z_dis_loss = z_discriminator_loss(z_real_decision, z_fake_decision)

        gen_loss = generator_loss(img_fake_decision1) + generator_loss(img_fake_decision2)
        enc_loss = encoder_loss(z_fake_decision)

    gradients_of_encoder = enc_tape.gradient(enc_loss, encoder.trainable_variables)
    gradients_of_generator = gen_tape.gradient(gen_loss, generator.trainable_variables)
    gradients_of_img_discriminator = img_dis_tape.gradient(img_dis_loss, img_discriminator.trainable_variables)
    gradients_of_z_discriminator = z_dis_tape.gradient(z_dis_loss, z_discriminator.trainable_variables)

    encoder_optimizer.apply_gradients(zip(gradients_of_encoder, encoder.trainable_variables))
    generator_optimizer.apply_gradients(zip(gradients_of_generator, generator.trainable_variables))
    img_discriminator_optimizer.apply_gradients(zip(gradients_of_img_discriminator, img_discriminator.trainable_variables))
    z_discriminator_optimizer.apply_gradients(zip(gradients_of_z_discriminator, z_discriminator.trainable_variables))

    return reconstructed_images, enc_loss, gen_loss, z_dis_loss, img_dis_loss
#%%
step = 0
progress_bar = tqdm(range(PARAMS['epochs']))
progress_bar.set_description('iteration {}/{} | current loss ?'.format(step, PARAMS['epochs']))
#%%
def generate_and_save_images(images, epochs):
    fig = plt.figure(figsize=(4,4))

    for i in range(9):
        plt.subplot(3, 3, i+1)
        plt.imshow(images[i, :, :, 0] * 127.5 + 127.5)
        plt.axis('off')

    plt.savefig('./assets/image_at_epoch_{:04d}.png'.format(epochs))
    plt.show()
#%%
'''training'''
# loss_history = []
for _ in progress_bar:
    x_batch, y_batch = next(iter(train_dataset))
    step += 1

    reconstructed_images, enc_loss, gen_loss, z_dis_loss, img_dis_loss = train_one_step(x_batch, y_batch)
    # loss_history.append(current_loss.numpy())

    # progress_bar.set_description('setting: {} epochs:{} lr:{} dim:{} T:{} sigma:{} to {} | iteration {}/{} | current loss {:.3f}'.format(
    #     PARAMS['data'], PARAMS['epochs'], PARAMS['learning_rate'], PARAMS['embedding_dim'], PARAMS['T'], PARAMS['sigma_0'], PARAMS['sigma_T'],
    #     step, PARAMS['epochs'], current_loss
    # ))

    if step % 10 == 0:
        generate_and_save_images(reconstructed_images, step)

    if step == PARAMS['epochs']: break
#%%

max_stars_repo_name: an-seunghwan/gan

max_issues_repo_name: an-seunghwan/gan

max_forks_repo_name: an-seunghwan/gan


text:
import cv2
import numpy as np
from scripts.config import Config
from matplotlib import pyplot as plt

cfg = Config()
drawing = False

def freely_select_from_image(org_img):
    """
        Input:
            original_image
        Output:
            masked_image: input image for network
            mask: mask that is used on image
            image_gray: gray version if mask masked_image
            edge: edge map of masked_image
        Description:
            Freely remove any area from image.
        """
    img = org_img.copy()
    mask = np.empty_like(img)
    image_gray = cv2.cvtColor(img, cv2.COLOR_RGB2GRAY)
    edge = cv2.Canny(image_gray, cfg.thresh1, cfg.thresh2)

    def mouse_action(event, former_x, former_y, flags, param):
        global current_former_x, current_former_y, count, drawing
        if event == cv2.EVENT_LBUTTONDOWN:
            drawing = True
            current_former_x,current_former_y=former_x,former_y

        if event == cv2.EVENT_MOUSEMOVE:
            if drawing == True:
                cv2.line(img,(current_former_x,current_former_y),(former_x,former_y),(255,255,255), cfg.freely_select_mask_size)
                cv2.line(edge,(current_former_x,current_former_y),(former_x,former_y),(0,0,0), cfg.freely_select_mask_size)
                current_former_x = former_x
                current_former_y = former_y

        if event == cv2.EVENT_LBUTTONUP:
            drawing = False

        if (event == cv2.EVENT_MOUSEWHEEL):

            if (flags>0):
                cfg.freely_select_mask_size += 1;
            else:
                cfg.freely_select_mask_size -= 1;

    cv2.namedWindow('image')
    cv2.setMouseCallback('image',mouse_action)

    while True:
        cv2.imshow("image", img)
        k = cv2.waitKey(20) & 0xFF
        if k == 27:
            break

    mask = cv2.subtract(img[:,:,0], org_img[:,:,0])
    ret,mask = cv2.threshold(mask,1,255,cv2.THRESH_BINARY)
    return img, mask, image_gray, edge

def select_by_edge(org_img):
    """
    Input:
        org_img: unmasked image
    Output:
        masked_image
    Description:
        This function is not workin currently, it returns the org_img
        Should find a way to select by contours.
    """
    img = org_img.copy()
    img_gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)

    def mouse_action(event, x, y, flags, param):
        if event == cv2.EVENT_LBUTTONDOWN:
            for i in range(0,len(contours)):
                r=cv2.pointPolygonTest(contours[i], (x,y),False)
                if r > 0:
                    cv2.floodFill(img_gray, contours[i], (0, 0), 150)

        if event == cv2.EVENT_RBUTTONDOWN:
            pass

    v = np.median(img_gray)

    #---- Apply automatic Canny edge detection using the computed median----
    cfg.thresh1 = int(max(0, (1.0 - 0.33) * v))
    cfg.thresh2 = int(min(255, (1.0 + 0.33) * v))

    while True:
        edged = cv2.Canny(img_gray, cfg.thresh1, cfg.thresh2)
        edged_not = cv2.bitwise_not(edged)
        ret, thresh = cv2.threshold(edged_not, 127, 255, 0)
        contours, hierarchy = cv2.findContours(thresh, cv2.RETR_TREE, cv2.CHAIN_APPROX_SIMPLE)
        # cv2.drawContours(edged, contours, -1, (0,255,0), 3)
        cv2.namedWindow('image')
        cv2.setMouseCallback('image',mouse_action)
        cv2.imshow("image", edged_not)
        k = cv2.waitKey(30) & 0xFF
        if k == 27:
            break
        if k == ord('h'):
            cfg.thresh1 += 10
            print(f"thresh1: {cfg.thresh1}, thresh2: {cfg.thresh2}")
        if k == ord('b'):
            cfg.thresh1 -= 10
            print(f"thresh1: {cfg.thresh1}, thresh2: {cfg.thresh2}")
        if k == ord('j'):
            cfg.thresh2 += 10
            print(f"thresh1: {cfg.thresh1}, thresh2: {cfg.thresh2}")
        if k == ord('n'):
            cfg.thresh2 -= 10
            print(f"thresh1: {cfg.thresh1}, thresh2: {cfg.thresh2}")

    return org_img

max_stars_repo_name: BeeGrad/semantic-object-removal

max_issues_repo_name: BeeGrad/semantic-object-removal

max_forks_repo_name: BeeGrad/semantic-object-removal


text:
#!/usr/bin/env python
#   pipescaler/mergers/alpha_merger.py
#
#   Copyright (C) 2020-2021 Karl T Debiec
#   All rights reserved.
#
#   This software may be modified and distributed under the terms of the
#   BSD license.
from __future__ import annotations

from logging import info
from typing import Any

import numpy as np
from PIL import Image

from pipescaler.core import Merger, validate_image

class AlphaMerger(Merger):
    """Merges alpha and color images into a single image with transparency."""

    def __call__(self, outfile: str, **kwargs: Any) -> None:
        infiles = {k: kwargs.get(k) for k in self.inlets}

        # Read images
        color_image = validate_image(infiles["color"], ["L", "RGB"])
        alpha_image = validate_image(infiles["alpha"], "L")

        # Merge images
        color_datum = np.array(color_image)
        alpha_datum = np.array(alpha_image)
        if color_image.mode == "L":
            output_datum = np.zeros((*color_datum.shape, 2), np.uint8)
            output_datum[:, :, 0] = color_datum
        else:
            output_datum = np.zeros((*color_datum.shape[:-1], 4), np.uint8)
            output_datum[:, :, :-1] = color_datum
        output_datum[:, :, -1] = alpha_datum
        output_image = Image.fromarray(output_datum)

        # Write image
        output_image.save(outfile)
        info(f"'{self}: '{outfile}' saved")

    @property
    def inlets(self):
        return ["color", "alpha"]

max_stars_repo_name: KarlTDebiec/PipeScaler

max_issues_repo_name: KarlTDebiec/PipeScaler

max_forks_repo_name: KarlTDebiec/PipeScaler


text:
from abc import ABCMeta, abstractmethod
from OpenGL.GL import *
import numpy

class ShaderProgram(object):
    __metaclass__ = ABCMeta

    def __init__(self, vertex_file, fragment_file):
        self.vertex_shader_id = self.load_shader(vertex_file, GL_VERTEX_SHADER)
        self.fragment_shader_id = self.load_shader(fragment_file, GL_FRAGMENT_SHADER)
        self.program_id = glCreateProgram()
        glAttachShader(self.program_id, self.vertex_shader_id)
        glAttachShader(self.program_id, self.fragment_shader_id)
        self.bind_all_attributes()
        glLinkProgram(self.program_id)
        glValidateProgram(self.program_id)
        self.get_all_uniform_locations()

    def get_uniform_location(self, uniform_name):
        return glGetUniformLocation(self.program_id, uniform_name)

    @abstractmethod
    def get_all_uniform_locations(self):
        pass

    @abstractmethod
    def bind_all_attributes(self):
        pass

    def start(self):
        glUseProgram(self.program_id)

    def stop(self):
        glUseProgram(0)

    def clean_up(self):
        self.stop()
        glDetachShader(self.program_id, self.vertex_shader_id)
        glDetachShader(self.program_id, self.fragment_shader_id)
        glDeleteShader(self.vertex_shader_id)
        glDeleteShader(self.fragment_shader_id)
        glDeleteProgram(self.program_id)

    def bind_attribute(self, attribute, variable_name):
        glBindAttribLocation(self.program_id, attribute, variable_name)

    def load_float(self, location, value):
        glUniform1f(location, value)

    def load_int(self, location, value):
        glUniform1i(location, value)

    def load_3d_vector(self, location, vector):
        glUniform3f(location, vector[0], vector[1], vector[2])

    def load_2d_vector(self, location, vector):
        glUniform2f(location, vector[0], vector[1])

    def load_boolean(self, location, value):
        to_load = 0
        if value == True:
            to_load = 1
        glUniform1f(location, to_load)

    def load_matrix(self, location, matrix):
        matrix_buffer = numpy.array(matrix, dtype="float32")
        # matrix_buffer = matrix
        glUniformMatrix4fv(location, 1, GL_FALSE, matrix_buffer)

    def load_shader(self, file, type):
        shader_file = open((file), "r")
        shader_file_data = shader_file.readlines()
        shader_source = ""
        for line in shader_file_data:
            shader_source += line + "//\n"
            shader_source += "// " + line + "\n"
        shader_id = glCreateShader(type)
        glShaderSource(shader_id, shader_source)
        glCompileShader(shader_id)
        return shader_id

max_stars_repo_name: westernesque/a-history-of-birds

max_issues_repo_name: westernesque/a-history-of-birds

max_forks_repo_name: westernesque/a-history-of-birds


text:
# coding: utf-8

# TODO improve this file

from sympy.core import Basic
from sympy.core.containers import Tuple

class BasicMeasure(Basic):

    def _sympystr(self, printer):
        sstr = printer.doprint
        txt = ''.join(j for j in ['d{}'.format(sstr(i)) for i in self.args])
        return txt

class CanonicalMeasure(BasicMeasure):
    """
    Represents a canonical measure.

    Examples

    """
    def __new__(cls, ldim):
        if not isinstance(ldim, int):
            raise TypeError('> Expecting an integer')

        coordinates = ['x1', 'x2', 'x3'][:ldim]
        coordinates = Tuple(*coordinates)

        return Basic.__new__(cls, coordinates)

    @property
    def args(self):
        return self._args[0]

class CartesianMeasure(BasicMeasure):
    """
    Represents a cartesian measure.

    Examples

    """
    def __new__(cls, ldim):
        if not isinstance(ldim, int):
            raise TypeError('> Expecting an integer')

        coordinates = ['x', 'y', 'z'][:ldim]
        coordinates = Tuple(*coordinates)

        return Basic.__new__(cls, coordinates)

    @property
    def args(self):
        return self._args[0]

class Measure(BasicMeasure):
    """
    Represents a measure from coordinates.

    Examples

    """
    def __new__(cls, coordinates):
        if not isinstance(coordinates, (tuple, list, Tuple)):
            coordinates = [coordinates]

        coordinates = Tuple(*coordinates)

        return Basic.__new__(cls, coordinates)

    @property
    def args(self):
        return self._args[0]

max_stars_repo_name: pyccel/sympde

max_issues_repo_name: pyccel/sympde

max_forks_repo_name: pyccel/sympde

reason: from sympy


text:
_author_ = "Rohit Patil"

import json
import logging

import numpy as np
import torch.utils.data
from PIL import Image
from flask import Flask, jsonify, request
from torchvision import transforms, models
from transformers import BertTokenizer, BertForSequenceClassification

logging.basicConfig(
    level=logging.DEBUG,
    format="%(asctime)s %(levelname)s %(name)s %(threadName)s : %(message)s",
)
app = Flask(__name__)
logger = logging.getLogger("app")

DEVICE = None
TOKENIZER = None
BERT_MODEL = None
ORGAN_MODEL = None
PLANE_MODEL = None
MODALITY_MODEL = None

def tokenize_sentences(sentence, tokenizer, max_seq_len=128):
    encoded_dict = tokenizer.encode_plus(
        sentence,  # Sentence to encode.
        add_special_tokens=True,  # Add '[CLS]' and '[SEP]'
        max_length=max_seq_len,  # Pad & truncate all sentences.
        pad_to_max_length=True,
        return_attention_mask=True,  # Construct attn. masks.
        return_tensors="pt",  # Return pytorch tensors.
    )
    return encoded_dict["input_ids"], encoded_dict["attention_mask"]

def get_bert_model(model_path):
    logger.info("Starting loading Bert.")
    model = BertForSequenceClassification.from_pretrained(model_path)
    tokenizer = BertTokenizer.from_pretrained(model_path)

    model.to(DEVICE)
    for param in model.parameters():
        param.requires_grad = False
    model.eval()
    logger.info("Bert loaded")
    return tokenizer, model

def get_organ_model(model_path):
    logger.info("Loading organ model started.")
    model = models.vgg16(pretrained=False)

    for param in model.parameters():
        param.requires_grad = False
    n_inputs = model.classifier[6].in_features

    model.classifier[6] = torch.nn.Sequential(
        torch.nn.Linear(n_inputs, 256),
        torch.nn.ReLU(),
        torch.nn.Dropout(0.2),
        torch.nn.Linear(256, 10),
        torch.nn.LogSoftmax(dim=1),
    )

    model.load_state_dict(torch.load(model_path, map_location=DEVICE))
    model.to(DEVICE)
    model.eval()
    logger.info("Loading organ model finished.")
    return model

def get_plane_model(model_path):
    logger.info("Loading plane model started.")
    model = models.vgg16(pretrained=False)

    for param in model.parameters():
        param.requires_grad = False
    n_inputs = model.classifier[6].in_features

    model.classifier[6] = torch.nn.Sequential(
        torch.nn.Linear(n_inputs, 256),
        torch.nn.ReLU(),
        torch.nn.Dropout(0.2),
        torch.nn.Linear(256, 15),
        torch.nn.LogSoftmax(dim=1),
    )

    model.load_state_dict(torch.load(model_path, map_location=DEVICE))
    model.to(DEVICE)
    model.eval()
    logger.info("Loading plane model finished.")
    return model

def question_classifier(question):
    label_dict = {0: "Modality", 1: "Plane", 2: "Organ", 3: "Abnormality"}

    input_ids, attention_mask = tokenize_sentences(question, TOKENIZER, 512)
    outputs = BERT_MODEL(
        input_ids.to(DEVICE),
        token_type_ids=None,
        attention_mask=attention_mask.to(DEVICE),
    )
    logits = outputs[0]
    question_type = label_dict.get(torch.argmax(logits).item())

    logger.info(question + "    classified to class:-" + question_type)
    return question_type

@app.route("/predict_question_type", methods=["POST"])
def predict_question_type():
    if request.method == "POST":
        logger.info("Received request for predict_question_type.")
        data = request.get_json()
        question_type = question_classifier(data["question"])
        output = jsonify({"question": data["question"], "question_type": question_type})
        logger.info(
            "Request processed for predict_question_type. Output :-" + str(output.data)
        )
        return output

def predict_answer_from_image(image_json, model_type):
    organ_types = {
        0: "breast",
        1: "face, sinuses, and neck",
        2: "gastrointestinal",
        3: "genitourinary",
        4: "heart and great vessels",
        5: "lung, mediastinum, pleura",
        6: "musculoskeletal",
        7: "skull and contents",
        8: "spine and contents",
        9: "vascular and lymphatic",
    }

    plane_types = {
        0: "3d reconstruction",
        1: "ap",
        2: "axial",
        3: "coronal",
        4: "decubitus",
        5: "frontal",
        6: "lateral",
        7: "longitudinal",
        8: "mammo - cc",
        9: "mammo - mag cc",
        10: "mammo - mlo",
        11: "oblique",
        12: "pa",
        13: "sagittal",
        14: "transverse",
    }

    organ_trans = transforms.Compose(
        [
            transforms.Resize(224),
            transforms.CenterCrop(224),
            transforms.ToTensor(),
            transforms.Normalize([0.485, 0.456, 0.406], [0.229, 0.224, 0.225]),
        ]
    )
    image = Image.fromarray(np.array(json.loads(image_json), dtype="uint8"))
    tensor = organ_trans(image).unsqueeze(0)
    if model_type == "Organ":
        outputs = ORGAN_MODEL.forward(tensor)
        _, y_hat = outputs.max(1)
        return organ_types.get(y_hat.item())
    elif model_type == "Modality":
        outputs = ORGAN_MODEL.forward(tensor)
        _, y_hat = outputs.max(1)
        return organ_types.get(y_hat.item())
    elif model_type == "Plane":
        outputs = PLANE_MODEL.forward(tensor)
        _, y_hat = outputs.max(1)
        return plane_types.get(y_hat.item())
    else:
        return "Sorry system does not know answer."

@app.route("/predict_answer", methods=["POST"])
def predict_answer():
    if request.method == "POST":
        logger.info("Received request for predict_answer.")
        data = request.get_json()
        question_type = question_classifier(data["question"])
        image_json = data["image"]
        answer = predict_answer_from_image(image_json, question_type)
        output = jsonify(
            {
                "question": data["question"],
                "question_type": question_type,
                "answer": answer,
            }
        )
        logger.info("Request processed for predict_question_type. Response is :-")
        parsed = json.loads(output.data)
        logger.info(json.dumps(parsed, indent=2))
        return output

def load_properties(filepath, sep="=", comment_char="#"):
    """
    Read the file passed as parameter as a properties file.
    """
    logging.info("Started loading config.properties.")
    props = {}
    with open(filepath, "rt") as f:
        for line in f:
            l = line.strip()
            if l and not l.startswith(comment_char):
                key_value = l.split(sep)
                key = key_value[0].strip()
                value = sep.join(key_value[1:]).strip().strip('"')
                props[key] = value
    logging.info("Properties loaded.")
    logging.info(props)
    return props

if __name__ == "__main__":
    properties = load_properties("config.properties")
    if torch.cuda.is_available():
        DEVICE = torch.device("cuda")
        logger.info("There are %d GPU(s) available." % torch.cuda.device_count())
        logger.info("We will use the GPU:", torch.cuda.get_device_name(0))
    else:
        logger.info("No GPU available, using the CPU instead.")
        DEVICE = torch.device("cpu")

    TOKENIZER, BERT_MODEL = get_bert_model(
        model_path=properties.get("question_classifier_path")
    )
    PLANE_MODEL = get_plane_model(model_path=properties.get("plane_classifier_path"))
    ORGAN_MODEL = get_organ_model(model_path=properties.get("organ_classifier_path"))
    app.run(debug=False)


text:
"""
contains various implementations for recommending movies
"""

from distutils.command.clean import clean
import pandas as pd
from scipy.sparse import csr_matrix
from sklearn.neighbors import NearestNeighbors
import random
import pickle
from utils import movies, ratings, popular_movies, clean_ratings, create_user_vec, clean_scores

# recommender_systems_intro_filled.ipynb
def recommend_random(query, movies, k=10):
    """
    Recommends a list of k random movie ids
    """
    movies = movies.drop(index=query.keys())
    list_ids = random.choices(
        movies.index, weights=None, cum_weights=None, k=k)
    return list_ids

# recommender_systems_intro_filled.ipynb
def recommend_popular(query, ratings, k=10):
    """
    Recommend a list of k movie ids that are the most popular
    """
    ratings, _ = popular_movies(ratings)
    ratings.set_index('movieId').drop(index=query.keys(), inplace=True)
    popular_ids = ratings.sort_values(ascending=False)[:k]
    return popular_ids

# matrix_factorization_filled.ipynb
def recommend_nmf(query, ratings, model='../models/nmf_recommender.pkl', k=10):
    """
    Recommend a list of k movie ids based on a trained NMF model
    """
    ratings = clean_ratings(ratings, popular_movies(ratings))
    with open(model, 'rb') as file:
        model_nmf = pickle.load(file)
    # Initialize a sparse user-item rating matrix
    # (data, (row_ind, col_ind)
    R = csr_matrix(
        (ratings['rating'], (ratings['userId'], ratings['movieId'])))
    user_vec = create_user_vec(query, R)
    # user_vec -> encoding -> p_user_vec -> decoding -> user_vec_hat
    scores = model_nmf.inverse_transform(model.transform(user_vec))
    recommendations = clean_scores(scores, query)

    return recommendations

# neighborhood_based_filtering.ipynb
def recommend_neighbors(query, ratings, model='../models/distance_recommender.pkl', k=10):
    """
    Recommend a list of k movie ids based on the most similar users
    """
    ratings = clean_ratings(ratings, popular_movies(ratings))
    R = csr_matrix(
        (ratings['rating'], (ratings['userId'], ratings['movieId'])))
    # initialize the unsupervised model
    model = NearestNeighbors(metric='cosine')
    # fit it to the user-item rating matrix
    model.fit(R)
    # create user vector
    user_vec = create_user_vec(query, R)
    # calculates the distances to all other users in the data!
    distances, userIds = model.kneighbors(
        user_vec, n_neighbors=10, return_distance=True
    )
    # sklearn returns a list of predictions - extract the first and only value of the list
    distances = distances[0]
    userIds = userIds[0]
    # only look at ratings for users that are similar!
    neighborhood = ratings.set_index('userId').loc[userIds]
    scores = neighborhood.groupby('movieId')['rating'].mean()
    recommendations = clean_scores(scores, query)

    return recommendations

if __name__ == '__main__':
    # list of liked movies
    query = [1, 34, 56, 21]
    print(recommend_random(query, movies))

reason: from scipy


text:
"""
take a genomic `fasta` and the `blast` output which resulted from the self-self
blast of that fasta. and generate a new fasta in which all basepairs occuring
more then `cutoff` times are masked. e.g.:
    $ python mask_genome.py -b rice_rice.blast -o rice -f rice.fasta -c 50
will create a new fasta file rice.masked.50.fasta
"""
import numpy as np
from pyfasta import Fasta
import tables
import numexpr
import sys, os

def cache_clear(cache, node, qchr, schr):
    """ keep some of the (hdf5) arrays in memory
    if there are too many, put one back into
    the h5 file
    """
    if len(cache) < 50: return
    #print "updating cache: %s, %s" % (qchr, schr)
    rmkey = [k for k in cache if not k in (qchr, schr)][0]
    getattr(node, 'c' + rmkey)[:] = cache[rmkey]
    del cache[rmkey]

def update_cache(achr, node, clen, h5, cache):
    """
    put a new (hdf5) array in memory, and make sure it's
    in the h5 as well
    """
    if not 'c' + achr in node:
        h5.createArray(node, 'c' + achr, np.zeros((clen,), dtype=np.uint32))

    cache[achr] = getattr(node, 'c' + achr)[:]

H5 = 'copy_count.h5'
def get_node(org, mode):
    # get the parent group node in the h5 file.
    if mode == 'w':
        h5 = tables.openFile(H5, mode='a')
        if org in h5.root:
            action = raw_input(\
               """%s copy counts exist in %s. what to do [d/a/u]?
                   'd': delete them and create new copy-counts
                   'a': abort
                   'u': use the existing copy-counts
                you can use the existing counts if the blast is unchanged."""
                               % (org, H5))[0].lower()
            if action == 'd':
                getattr(h5.root, org)._f_remove(recursive=True)
                h5.flush()
            elif action == 'u':
                h5.close()
                return None, None
            else:
                print('ABORT: %s already exists in %s' % (org, H5))
                h5.close(); sys.exit()
        return h5, h5.createGroup(h5.root, org, org)
    else:
        h5 = tables.openFile(H5, mode='r')
        return h5, getattr(h5.root, org)

def count_freq(blast_file, fasta, org, count_subject=True):
    """one large blast file """
    h5, node = get_node(org, 'w')

    # use existing counts.
    if (h5, node) == (None, None): return
    f = Fasta(fasta)

    print "counting..."
    cache = {}
    for sline in open(blast_file):
        line = sline.split("\t")
        qchr, schr = line[:2]

        qstart, qstop, sstart, sstop = map(int, line[6:10])

        if not qchr in cache:
            update_cache(qchr, node, len(f[qchr]), h5, cache)
            cache_clear(cache, node, qchr, schr)
        # convert to 0-based indexes:
        # 1 8 => 0 8, but range doesnt include upper boud.
        cache[qchr][qstart - 1: qstop] += 1

        if count_subject:
            if sstart > sstop: sstart, sstop = sstop, sstart
            if not schr in cache:
                update_cache(schr, node, len(f[schr]), h5, cache)
                cache_clear(cache, node, qchr, schr)
                cache[schr][sstart - 1: sstop] += 1

    for achr in cache:
        getattr(node, 'c' + achr)[:] = cache[achr]

    h5.close()

def mask(fasta_file, org, cutoff, mask_value='X'):
    h5, node = get_node(org, 'r')

    outfile = fasta_file[:fasta_file.rfind(".")] + (".masked.%i" % cutoff) \
                         + fasta_file[fasta_file.rfind("."):]

    print "> masking sequence to file:", outfile
    out = open(outfile ,'w')

    fasta = Fasta(fasta_file)

    soft_mask = mask_value.lower() == 'soft'
    for seqid in sorted(fasta.iterkeys()):
        masked = 0
        if soft_mask:
            seq = str(fasta[seqid])
            # mask is the lowercase sequence.
            mask_value = np.array(seq.lower(), dtype='c')
            seq = np.array(seq.upper(), dtype='c')
        else:
            fasta[seqid].tostring = False
            seq = fasta[seqid][:] # a

        if not 'c' + seqid in node:
            print >>sys.stderr, seqid,\
                '! not found in masked, writing unchanged\n' \
                '  this means that no section of this sequence appeared\n' \
                '  more than %i times' % cutoff
            out.write('>' + seqid + '\n')
            out.write(seq.tostring() + '\n')
            continue

        hit_counts = getattr(node, 'c' + seqid)[:]
        masked_seq = np.where(numexpr.evaluate("hit_counts > %i" % cutoff)
                              , mask_value, seq).tostring()

        l = len(masked_seq)
        print >>sys.stderr, "! seq:%s len:%i %%masked:%.3f" % (seqid, l,
                                   100.0 * masked_seq.count(mask_value) / l)
        assert len(seq) == l
        out.write('>' + seqid + '\n')
        out.write(masked_seq + '\n')

    out.close()
    # write out a file .fasta.version containing
    # the svnversion (if available of this script
    # that was used to create the file.
    path = os.path.dirname(__file__)
    os.system('svnversion %s > %s.version' % (path, outfile))
    h5.close()

if __name__ == "__main__":
    import optparse
    p = optparse.OptionParser(__doc__)
    p.add_option("-b", dest="blast", help="path to self-self genomic blast")
    p.add_option("-f", dest="fasta", help="path to the fasta file (which was\n"
                 "used as query and subject in the blast")
    p.add_option("-o", dest="org", help="name of the organism. e.g. 'rice'")
    p.add_option("--h5", dest="h5", help="path to the hdf5 file to use.", default=H5)
    p.add_option("-c", dest="cutoff",
                 help="cutoff value, bp locations appearing in the blast more\n"
                 "than this many times are masked", type='int', default=50)
    p.add_option("-S", dest="no_subject", default=False, action="store_true",
             help="do NOT count subject hits (only query) when tabulating")
    p.add_option("-m", dest="mask", help=\
         "mask sequence with this letter (usually 'X' or 'N'). if == 'SOFT',"
         "then the sequence subjected to soft-masking where all repetitive"
         "values are lower-cased. and all other sequence is upper-cased"
         "regardless of its case in the original fasta", default='X')

    options, _ = p.parse_args()
    if not (options.blast and options.fasta and options.org):
        sys.exit(p.print_help())
    ospe = os.path.exists
    if not (ospe(options.blast) and ospe(options.fasta)):
            print "make sure blast:%s and fasta:%s exist:" \
                                % (options.blast, options.fasta)
            sys.exit()

    if options.h5: H5=options.h5

    assert len(options.mask) == 1 or options.mask.lower() == 'soft'
    count_freq(options.blast, options.fasta, options.org, not options.no_subject)
    print "> done counting..."
    mask(options.fasta, options.org, options.cutoff, options.mask)

reason: import numpy,import numexpr