text:
# Copyright 1999-2020 Alibaba Group Holding Ltd.
#
# 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 operator
import unittest
from functools import partial

import numpy as np
import pandas as pd

from mars.tests.core import TestBase, parameterized, ExecutorForTest
from mars import tensor as mt
from mars.tensor.datasource import array as from_array
from mars.dataframe import to_datetime
from mars.dataframe.datasource.dataframe import from_pandas
from mars.dataframe.datasource.series import from_pandas as from_pandas_series
from mars.dataframe.arithmetic.tests.test_arithmetic import comp_func

binary_functions = dict(
    add=dict(func=operator.add, func_name='add', rfunc_name='radd'),
    equal=dict(func=comp_func('eq', 'eq'), func_name='eq', rfunc_name='eq'),
    logical_and=dict(func=operator.and_, func_name='__and__', rfunc_name='__rand__'),
)

@parameterized(**binary_functions)
class TestBinary(TestBase):
    def setUp(self):
        self.executor = ExecutorForTest()

    def to_boolean_if_needed(self, value, split_value=0.5):
        if self.func_name in ['__and__', '__or__', '__xor__']:
            return value > split_value
        else:
            return value

    def testWithoutShuffleExecution(self):
        if self.func_name in ['__and__', '__or__', '__xor__']:
            # FIXME bitwise logical operators behave differently with pandas when index is not aligned.
            return

        # all the axes are monotonic
        # data1 with index split into [0...4], [5...9],
        # columns [3...7], [8...12]
        data1 = pd.DataFrame(np.random.rand(10, 10), index=np.arange(10),
                             columns=np.arange(3, 13))
        data1 = self.to_boolean_if_needed(data1)
        df1 = from_pandas(data1, chunk_size=5)
        # data2 with index split into [6...11], [2, 5],
        # columns [4...9], [10, 13]
        data2 = pd.DataFrame(np.random.rand(10, 10), index=np.arange(11, 1, -1),
                             columns=np.arange(4, 14))
        data2 = self.to_boolean_if_needed(data2)
        df2 = from_pandas(data2, chunk_size=6)

        df3 = self.func(df1, df2)

        expected = self.func(data1, data2)
        result = self.executor.execute_dataframe(df3, concat=True)[0]

        pd.testing.assert_frame_equal(expected, result)

    def testWithOneShuffleExecution(self):
        if self.func_name in ['__and__', '__or__', '__xor__']:
            # FIXME bitwise logical operators behave differently with pandas when index is not aligned.
            return

        # only 1 axis is monotonic
        # data1 with index split into [0...4], [5...9],
        data1 = pd.DataFrame(np.random.rand(10, 10), index=np.arange(10),
                             columns=[4, 1, 3, 2, 10, 5, 9, 8, 6, 7])
        data1 = self.to_boolean_if_needed(data1)
        df1 = from_pandas(data1, chunk_size=5)
        # data2 with index split into [6...11], [2, 5],
        data2 = pd.DataFrame(np.random.rand(10, 10), index=np.arange(11, 1, -1),
                             columns=[5, 9, 12, 3, 11, 10, 6, 4, 1, 2])
        data2 = self.to_boolean_if_needed(data2)
        df2 = from_pandas(data2, chunk_size=6)

        df3 = self.func(df1, df2)

        expected = self.func(data1, data2)
        result = self.executor.execute_dataframe(df3, concat=True)[0]

        pd.testing.assert_frame_equal(expected, result)

        # only 1 axis is monotonic
        # data1 with columns split into [0...4], [5...9],
        data1 = pd.DataFrame(np.random.rand(10, 10), index=[4, 1, 3, 2, 10, 5, 9, 8, 6, 7],
                             columns=np.arange(10))
        data1 = self.to_boolean_if_needed(data1)
        df1 = from_pandas(data1, chunk_size=5)
        # data2 with columns split into [6...11], [2, 5],
        data2 = pd.DataFrame(np.random.rand(10, 10), index=[5, 9, 12, 3, 11, 10, 6, 4, 1, 2],
                             columns=np.arange(11, 1, -1))
        data2 = self.to_boolean_if_needed(data2)
        df2 = from_pandas(data2, chunk_size=6)

        df3 = self.func(df1, df2)

        expected = self.func(data1, data2)
        result = self.executor.execute_dataframe(df3, concat=True)[0]

        pd.testing.assert_frame_equal(expected, result)

    def testWithAllShuffleExecution(self):
        if self.func_name in ['__and__', '__or__', '__xor__']:
            # FIXME bitwise logical operators behave differently with pandas when index is not aligned.
            return

        # no axis is monotonic
        data1 = pd.DataFrame(np.random.rand(10, 10), index=[0, 10, 2, 3, 4, 5, 6, 7, 8, 9],
                             columns=[4, 1, 3, 2, 10, 5, 9, 8, 6, 7])
        data1 = self.to_boolean_if_needed(data1)
        df1 = from_pandas(data1, chunk_size=5)
        data2 = pd.DataFrame(np.random.rand(10, 10), index=[11, 1, 2, 5, 7, 6, 8, 9, 10, 3],
                             columns=[5, 9, 12, 3, 11, 10, 6, 4, 1, 2])
        data2 = self.to_boolean_if_needed(data2)
        df2 = from_pandas(data2, chunk_size=6)

        df3 = self.func(df1, df2)

        expected = self.func(data1, data2)
        result = self.executor.execute_dataframe(df3, concat=True)[0]

        pd.testing.assert_frame_equal(expected, result)

    def testBothWithOneChunk(self):
        if self.func_name in ['__and__', '__or__', '__xor__']:
            # FIXME bitwise logical operators behave differently with pandas when index is not aligned.
            return

        # only 1 axis is monotonic
        # data1 with index split into [0...4], [5...9],
        data1 = pd.DataFrame(np.random.rand(10, 10), index=np.arange(10),
                             columns=[4, 1, 3, 2, 10, 5, 9, 8, 6, 7])
        data1 = self.to_boolean_if_needed(data1)
        df1 = from_pandas(data1, chunk_size=10)
        # data2 with index split into [6...11], [2, 5],
        data2 = pd.DataFrame(np.random.rand(10, 10), index=np.arange(11, 1, -1),
                             columns=[5, 9, 12, 3, 11, 10, 6, 4, 1, 2])
        data2 = self.to_boolean_if_needed(data2)
        df2 = from_pandas(data2, chunk_size=10)

        df3 = self.func(df1, df2)

        expected = self.func(data1, data2)
        result = self.executor.execute_dataframe(df3, concat=True)[0]

        pd.testing.assert_frame_equal(expected, result)

        # only 1 axis is monotonic
        # data1 with columns split into [0...4], [5...9],
        data1 = pd.DataFrame(np.random.rand(10, 10), index=[4, 1, 3, 2, 10, 5, 9, 8, 6, 7],
                             columns=np.arange(10))
        data1 = self.to_boolean_if_needed(data1)
        df1 = from_pandas(data1, chunk_size=10)
        # data2 with columns split into [6...11], [2, 5],
        data2 = pd.DataFrame(np.random.rand(10, 10), index=[5, 9, 12, 3, 11, 10, 6, 4, 1, 2],
                             columns=np.arange(11, 1, -1))
        data2 = self.to_boolean_if_needed(data2)
        df2 = from_pandas(data2, chunk_size=10)

        df3 = self.func(df1, df2)

        expected = self.func(data1, data2)
        result = self.executor.execute_dataframe(df3, concat=True)[0]

        pd.testing.assert_frame_equal(expected, result)

    def testWithoutShuffleAndWithOneChunk(self):
        if self.func_name in ['__and__', '__or__', '__xor__']:
            # FIXME bitwise logical operators behave differently with pandas when index is not aligned.
            return

        # only 1 axis is monotonic
        # data1 with index split into [0...4], [5...9],
        data1 = pd.DataFrame(np.random.rand(10, 10), index=np.arange(10),
                             columns=[4, 1, 3, 2, 10, 5, 9, 8, 6, 7])
        data1 = self.to_boolean_if_needed(data1)
        df1 = from_pandas(data1, chunk_size=(5, 10))
        # data2 with index split into [6...11], [2, 5],
        data2 = pd.DataFrame(np.random.rand(10, 10), index=np.arange(11, 1, -1),
                             columns=[5, 9, 12, 3, 11, 10, 6, 4, 1, 2])
        data2 = self.to_boolean_if_needed(data2)
        df2 = from_pandas(data2, chunk_size=(6, 10))

        df3 = self.func(df1, df2)

        expected = self.func(data1, data2)
        result = self.executor.execute_dataframe(df3, concat=True)[0]

        pd.testing.assert_frame_equal(expected, result)

        # only 1 axis is monotonic
        # data1 with columns split into [0...4], [5...9],
        data1 = pd.DataFrame(np.random.rand(10, 10), index=[4, 1, 3, 2, 10, 5, 9, 8, 6, 7],
                             columns=np.arange(10))
        data1 = self.to_boolean_if_needed(data1)
        df1 = from_pandas(data1, chunk_size=(10, 5))
        # data2 with columns split into [6...11], [2, 5],
        data2 = pd.DataFrame(np.random.rand(10, 10), index=[5, 9, 12, 3, 11, 10, 6, 4, 1, 2],
                             columns=np.arange(11, 1, -1))
        data2 = self.to_boolean_if_needed(data2)
        df2 = from_pandas(data2, chunk_size=(10, 6))

        df3 = self.func(df1, df2)

        expected = self.func(data1, data2)
        result = self.executor.execute_dataframe(df3, concat=True)[0]

        pd.testing.assert_frame_equal(expected, result)

    def testWithShuffleAndWithOneChunk(self):
        if self.func_name in ['__and__', '__or__', '__xor__']:
            # pandas fails to compute some expected values due to `na`.
            return

        # only 1 axis is monotonic
        # data1 with index split into [0...4], [5...9],
        data1 = pd.DataFrame(np.random.rand(10, 10), index=np.arange(10),
                             columns=[4, 1, 3, 2, 10, 5, 9, 8, 6, 7])
        df1 = from_pandas(data1, chunk_size=(10, 5))
        # data2 with index split into [6...11], [2, 5],
        data2 = pd.DataFrame(np.random.rand(10, 10), index=np.arange(11, 1, -1),
                             columns=[5, 9, 12, 3, 11, 10, 6, 4, 1, 2])
        df2 = from_pandas(data2, chunk_size=(10, 6))

        df3 = self.func(df1, df2)

        expected = self.func(data1, data2)
        result = self.executor.execute_dataframe(df3, concat=True)[0]

        pd.testing.assert_frame_equal(expected, result)

        # only 1 axis is monotonic
        # data1 with columns split into [0...4], [5...9],
        data1 = pd.DataFrame(np.random.rand(10, 10), index=[4, 1, 3, 2, 10, 5, 9, 8, 6, 7],
                             columns=np.arange(10))
        data1 = self.to_boolean_if_needed(data1)
        df1 = from_pandas(data1, chunk_size=(5, 10))
        # data2 with columns split into [6...11], [2, 5],
        data2 = pd.DataFrame(np.random.rand(10, 10), index=[5, 9, 12, 3, 11, 10, 6, 4, 1, 2],
                             columns=np.arange(11, 1, -1))
        data2 = self.to_boolean_if_needed(data2)
        df2 = from_pandas(data2, chunk_size=(6, 10))

        df3 = self.func(df1, df2)

        expected = self.func(data1, data2)
        result = self.executor.execute_dataframe(df3, concat=True)[0]

        pd.testing.assert_frame_equal(expected, result)

    def testSameIndex(self):
        data = pd.DataFrame(np.random.rand(10, 10), index=np.random.randint(0, 2, size=(10,)),
                            columns=['c' + str(i) for i in range(10)])
        data = self.to_boolean_if_needed(data)
        df = from_pandas(data, chunk_size=3)
        df2 = self.func(df, df)

        expected = self.func(data, data)
        result = self.executor.execute_dataframe(df2, concat=True)[0]
        pd.testing.assert_frame_equal(expected, result)

        series = from_pandas_series(data.iloc[0], chunk_size=3)
        df3 = self.func(df, series)

        expected = self.func(data, data.iloc[0])
        result = self.executor.execute_dataframe(df3, concat=True)[0]
        pd.testing.assert_frame_equal(expected, result)

        series = from_pandas_series(data.iloc[:, 0], chunk_size=3)
        df4 = getattr(df, self.func_name)(series, axis=0)

        expected = getattr(data, self.func_name)(data.iloc[:, 0], axis=0)
        result = self.executor.execute_dataframe(df4, concat=True)[0]
        pd.testing.assert_frame_equal(expected, result)

    def testChained(self):
        data1 = pd.DataFrame(np.random.rand(10, 10))
        data1 = self.to_boolean_if_needed(data1)
        df1 = from_pandas(data1, chunk_size=5)
        data2 = pd.DataFrame(np.random.rand(10, 10))
        data2 = self.to_boolean_if_needed(data2)
        df2 = from_pandas(data2, chunk_size=6)

        df3 = self.func(df1, df2)

        data4 = pd.DataFrame(np.random.rand(10, 10))
        data4 = self.to_boolean_if_needed(data1)
        df4 = from_pandas(data4, chunk_size=6)

        df5 = self.func(df3, df4)

        result = self.executor.execute_dataframe(df5, concat=True)[0]
        expected = self.func(self.func(data1, data2), data4)

        pd.testing.assert_frame_equal(expected, result)

    def testRfunc(self):
        data1 = pd.DataFrame(np.random.rand(10, 10))
        data1 = self.to_boolean_if_needed(data1)
        df1 = from_pandas(data1, chunk_size=5)
        data2 = pd.DataFrame(np.random.rand(10, 10))
        data2 = self.to_boolean_if_needed(data2)
        df2 = from_pandas(data2, chunk_size=6)
        df3 = getattr(df1, self.rfunc_name)(df2)
        result = self.executor.execute_dataframe(df3, concat=True)[0]
        expected = self.func(data2, data1)
        pd.testing.assert_frame_equal(expected, result)

        data3 = pd.DataFrame(np.random.rand(10, 10))
        data3 = self.to_boolean_if_needed(data3)
        df4 = from_pandas(data3, chunk_size=5)
        df5 = getattr(df4, self.rfunc_name)(1)
        # todo check dtypes when pandas reverts its behavior on broadcasting
        check_dtypes = self.func_name not in ('__and__', '__or__', '__xor__')
        result = self.executor.execute_dataframe(df5, concat=True, check_dtypes=check_dtypes)[0]
        expected2 = self.func(1, data3)
        pd.testing.assert_frame_equal(expected2, result)

    def testWithMultiForms(self):
        # test multiple forms
        # such as self+other, self.add(other), add(self,other)
        data1 = pd.DataFrame(np.random.rand(10, 10))
        data1 = self.to_boolean_if_needed(data1)
        df1 = from_pandas(data1, chunk_size=5)
        data2 = pd.DataFrame(np.random.rand(10, 10))
        data2 = self.to_boolean_if_needed(data2)
        df2 = from_pandas(data2, chunk_size=6)

        expected = self.func(data1, data2)
        result = self.executor.execute_dataframe(self.func(df1, df2), concat=True)[0]
        pd.testing.assert_frame_equal(expected, result)
        result = self.executor.execute_dataframe(self.func(df1, df2), concat=True)[0]
        pd.testing.assert_frame_equal(expected, result)
        result = self.executor.execute_dataframe(getattr(df1, self.func_name)(df2), concat=True)[0]
        pd.testing.assert_frame_equal(expected, result)
        result = self.executor.execute_dataframe(getattr(df1, self.rfunc_name)(df2), concat=True)[0]
        pd.testing.assert_frame_equal(self.func(data2, data1), result)

    def testDataframeAndScalar(self):
        if self.func_name in ['__and__', '__or__', '__xor__']:
            # FIXME bitwise logical operators doesn\'t support floating point scalars
            return

        # test dataframe and scalar
        pdf = pd.DataFrame(np.random.rand(10, 10))
        pdf = self.to_boolean_if_needed(pdf)
        df = from_pandas(pdf, chunk_size=2)
        expected = self.func(pdf, 1)
        result = self.executor.execute_dataframe(self.func(df, 1), concat=True)[0]
        pd.testing.assert_frame_equal(expected, result)
        result2 = self.executor.execute_dataframe(self.func(df, 1), concat=True)[0]
        pd.testing.assert_frame_equal(expected, result2)
        result3 = self.executor.execute_dataframe(getattr(df, self.func_name)(1), concat=True)[0]
        pd.testing.assert_frame_equal(expected, result3)

        # test scalar and dataframe
        result4 = self.executor.execute_dataframe(self.func(df, 1), concat=True)[0]
        pd.testing.assert_frame_equal(expected, result4)

        expected2 = self.func(1, pdf)
        result5 = self.executor.execute_dataframe(self.func(1, df), concat=True)[0]
        pd.testing.assert_frame_equal(expected2, result5)

        result6 = self.executor.execute_dataframe(getattr(df, self.rfunc_name)(1), concat=True)[0]
        pd.testing.assert_frame_equal(expected2, result6)

    def testWithShuffleOnStringIndex(self):
        if self.func_name in ['__and__', '__or__', '__xor__']:
            # FIXME bitwise logical operators behave differently with pandas when index is not aligned.
            return

        # no axis is monotonic, and the index values are strings.
        data1 = pd.DataFrame(np.random.rand(10, 10), index=[str(x) for x in [0, 10, 2, 3, 4, 5, 6, 7, 8, 9]],
                             columns=[4, 1, 3, 2, 10, 5, 9, 8, 6, 7])
        data1 = self.to_boolean_if_needed(data1)
        df1 = from_pandas(data1, chunk_size=5)
        data2 = pd.DataFrame(np.random.rand(10, 10), index=[str(x) for x in [11, 1, 2, 5, 7, 6, 8, 9, 10, 3]],
                             columns=[5, 9, 12, 3, 11, 10, 6, 4, 1, 2])
        data2 = self.to_boolean_if_needed(data2)
        df2 = from_pandas(data2, chunk_size=6)

        df3 = self.func(df1, df2)

        expected = self.func(data1, data2)
        result = self.executor.execute_dataframe(df3, concat=True)[0]

        pd.testing.assert_frame_equal(expected, result)

    def testDataframeAndSeries(self):
        if self.func_name in ['__and__', '__or__', '__xor__']:
            # pandas fails to compute some expected values due to `na`.
            return

        data1 = pd.DataFrame(np.random.rand(10, 10), index=np.arange(10),
                             columns=[4, 1, 3, 2, 10, 5, 9, 8, 6, 7])
        data1 = self.to_boolean_if_needed(data1)
        data2 = pd.DataFrame(np.random.rand(10, 10), index=np.arange(11, 1, -1),
                             columns=[5, 9, 12, 3, 11, 10, 6, 4, 1, 2])
        data2 = self.to_boolean_if_needed(data2)

        s1 = from_pandas_series(data2[1], chunk_size=(6,))

        # operate on single-column dataframe and series
        df1 = from_pandas(data1[[1]], chunk_size=(5, 5))
        r1 = getattr(df1, self.func_name)(s1, axis='index')

        expected = getattr(data1[[1]], self.func_name)(data2[1], axis='index')
        result = self.executor.execute_dataframe(r1, concat=True)[0]
        pd.testing.assert_frame_equal(expected, result)

        # operate on dataframe and series without shuffle
        df2 = from_pandas(data1, chunk_size=(5, 5))
        r2 = getattr(df2, self.func_name)(s1, axis='index')

        expected = getattr(data1, self.func_name)(data2[1], axis='index')
        result = self.executor.execute_dataframe(r2, concat=True)[0]
        pd.testing.assert_frame_equal(expected, result)

        # operate on dataframe and series with shuffle
        df3 = from_pandas(data1, chunk_size=(5, 5))
        r3 = getattr(df3, self.func_name)(s1, axis='columns')

        expected = getattr(data1, self.func_name)(data2[1], axis='columns')
        result = self.executor.execute_dataframe(r3, concat=True)[0]
        pd.testing.assert_frame_equal(expected, result)

        # test both one chunk, axis=0
        pdf = pd.DataFrame({'ca': [1, 3, 2], 'cb': [360, 180, 2]}, index=[1, 2, 3])
        pdf = self.to_boolean_if_needed(pdf)
        df = from_pandas(pdf)
        series = pd.Series([0, 1, 2], index=[1, 2, 3])
        mars_series = from_pandas_series(series)
        result = self.executor.execute_dataframe(getattr(df, self.func_name)(mars_series, axis=0), concat=True)[0]
        expected = getattr(pdf, self.func_name)(series, axis=0)
        pd.testing.assert_frame_equal(expected, result)

        # test different number of chunks, axis=0
        pdf = pd.DataFrame({'ca': [1, 3, 2], 'cb': [360, 180, 2]}, index=[1, 2, 3])
        pdf = self.to_boolean_if_needed(pdf)
        df = from_pandas(pdf, chunk_size=1)
        series = pd.Series([0, 1, 2], index=[1, 2, 3])
        mars_series = from_pandas_series(series)
        result = self.executor.execute_dataframe(getattr(df, self.func_name)(mars_series, axis=0), concat=True)[0]
        expected = getattr(pdf, self.func_name)(series, axis=0)
        pd.testing.assert_frame_equal(expected, result)

        # test with row shuffle, axis=0
        pdf = pd.DataFrame({'ca': [1, 3, 2], 'cb': [360, 180, 2]}, index=[2, 1, 3])
        pdf = self.to_boolean_if_needed(pdf)
        df = from_pandas(pdf, chunk_size=1)
        series = pd.Series([0, 1, 2], index=[3, 1, 2])
        mars_series = from_pandas_series(series)
        result = self.executor.execute_dataframe(getattr(df, self.func_name)(mars_series, axis=0), concat=True)[0]
        expected = getattr(pdf, self.func_name)(series, axis=0).reindex([3, 1, 2])
        # modify the order of rows
        result = result.reindex(index=[3, 1, 2])
        pd.testing.assert_frame_equal(expected, result)

        # test both one chunk, axis=1
        pdf = pd.DataFrame({1: [1, 3, 2], 2: [360, 180, 2], 3: [1, 2, 3]}, index=['ra', 'rb', 'rc'])
        pdf = self.to_boolean_if_needed(pdf)
        df = from_pandas(pdf)
        series = pd.Series([0, 1, 2], index=[1, 2, 3])
        mars_series = from_pandas_series(series)
        result = self.executor.execute_dataframe(getattr(df, self.func_name)(mars_series, axis=1), concat=True)[0]
        expected = getattr(pdf, self.func_name)(series, axis=1)
        pd.testing.assert_frame_equal(expected, result)

        # test different number of chunks, axis=1
        pdf = pd.DataFrame({1: [1, 3, 2], 2: [360, 180, 2], 3: [1, 2, 3]}, index=['ra', 'rb', 'rc'])
        pdf = self.to_boolean_if_needed(pdf)
        df = from_pandas(pdf, chunk_size=1)
        series = pd.Series([0, 1, 2], index=[1, 2, 3])
        mars_series = from_pandas_series(series)
        result = self.executor.execute_dataframe(getattr(df, self.func_name)(mars_series, axis=1), concat=True)[0]
        expected = getattr(pdf, self.func_name)(series, axis=1)
        pd.testing.assert_frame_equal(expected, result)

        # test with row shuffle, axis=1
        pdf = pd.DataFrame({1: [1, 3, 2], 3: [1, 2, 3], 2: [360, 180, 2]}, index=['ra', 'rb', 'rc'])
        pdf = self.to_boolean_if_needed(pdf)
        df = from_pandas(pdf, chunk_size=1)
        series = pd.Series([0, 1, 2], index=[3, 1, 2])
        mars_series = from_pandas_series(series)
        result = self.executor.execute_dataframe(getattr(df, self.func_name)(mars_series, axis=1), concat=True)[0]
        expected = getattr(pdf, self.func_name)(series, axis=1)
        # modify the order of columns
        result = result[[1, 2, 3]]
        pd.testing.assert_frame_equal(expected, result)

    def testSeries(self):
        # only one chunk
        s1 = pd.Series(np.arange(10) + 1)
        s1 = self.to_boolean_if_needed(s1)
        s2 = pd.Series(np.arange(10) + 1)
        s2 = self.to_boolean_if_needed(s2)
        r = self.func(from_pandas_series(s1, chunk_size=10), from_pandas_series(s2, chunk_size=10))
        result = self.executor.execute_dataframe(r, concat=True)[0]
        expected = self.func(s1, s2)
        pd.testing.assert_series_equal(expected, result)

        # same index
        s1 = pd.Series(np.arange(10) + 1)
        s1 = self.to_boolean_if_needed(s1)
        s2 = pd.Series(np.arange(10) + 1)
        s2 = self.to_boolean_if_needed(s2)
        r = self.func(from_pandas_series(s1, chunk_size=4), from_pandas_series(s2, chunk_size=6))
        result = self.executor.execute_dataframe(r, concat=True)[0]
        expected = self.func(s1, s2)
        pd.testing.assert_series_equal(expected, result)

        # no shuffle
        s1 = pd.Series(np.arange(10) + 1, index=range(10))
        s1 = self.to_boolean_if_needed(s1)
        s2 = pd.Series(np.arange(10) + 1, index=range(10, 0, -1))
        s2 = self.to_boolean_if_needed(s2)
        r = self.func(from_pandas_series(s1, chunk_size=4), from_pandas_series(s2, chunk_size=6))
        result = self.executor.execute_dataframe(r, concat=True)[0]
        expected = self.func(s1, s2)
        pd.testing.assert_series_equal(expected, result)

        # shuffle
        data = (np.arange(10) + 1).astype(np.int64, copy=False)
        s1 = pd.Series(data, index=np.random.permutation(range(10)))
        s1 = self.to_boolean_if_needed(s1)
        s2 = pd.Series(data, index=np.random.permutation(range(10, 0, -1)))
        s2 = self.to_boolean_if_needed(s2)
        r = self.func(from_pandas_series(s1, chunk_size=4), from_pandas_series(s2, chunk_size=6))
        result = self.executor.execute_dataframe(r, concat=True)[0]
        expected = self.func(s1, s2)
        pd.testing.assert_series_equal(expected, result)

        if self.func_name in ['__and__', '__or__', '__xor__']:
            # bitwise logical operators doesn\'t support floating point scalars
            return

        # operate with scalar
        s1 = pd.Series(np.arange(10) + 1, index=np.random.permutation(range(10)))
        s1 = self.to_boolean_if_needed(s1)
        r = self.func(from_pandas_series(s1, chunk_size=4), 4)
        result = self.executor.execute_dataframe(r, concat=True)[0]
        expected = self.func(s1, 4)
        pd.testing.assert_series_equal(expected, result)

        # reverse with scalar
        s1 = pd.Series(np.arange(10) + 1, index=np.random.permutation(range(10)))
        s1 = self.to_boolean_if_needed(s1)
        r = self.func(4, from_pandas_series(s1, chunk_size=4))
        result = self.executor.execute_dataframe(r, concat=True)[0]
        expected = self.func(4, s1)
        pd.testing.assert_series_equal(expected, result)

    def testWithPlainValue(self):
        if self.func_name in ['__and__', '__or__', '__xor__']:
            # skip tests for bitwise logical operators on plain value.
            return

        data1 = pd.DataFrame(np.random.rand(10, 10), index=np.arange(10),
                             columns=[4, 1, 3, 2, 10, 5, 9, 8, 6, 7])
        data1 = self.to_boolean_if_needed(data1)
        df1 = from_pandas(data1, chunk_size=6)
        s1 = df1[2]

        r = getattr(df1, self.func_name)([1, 2, 3, 4, 5, 6, 7, 8, 9, 10], axis=0)
        result = self.executor.execute_dataframe(r, concat=True)[0]
        expected = getattr(data1, self.func_name)([1, 2, 3, 4, 5, 6, 7, 8, 9, 10], axis=0)
        pd.testing.assert_frame_equal(expected, result)

        r = getattr(df1, self.func_name)((1, 2, 3, 4, 5, 6, 7, 8, 9, 10), axis=0)
        result = self.executor.execute_dataframe(r, concat=True)[0]
        expected = getattr(data1, self.func_name)((1, 2, 3, 4, 5, 6, 7, 8, 9, 10), axis=0)
        pd.testing.assert_frame_equal(expected, result)

        r = getattr(s1, self.func_name)([1, 2, 3, 4, 5, 6, 7, 8, 9, 10])
        result = self.executor.execute_dataframe(r, concat=True)[0]
        expected = getattr(data1[2], self.func_name)([1, 2, 3, 4, 5, 6, 7, 8, 9, 10])
        pd.testing.assert_series_equal(expected, result)

        r = getattr(s1, self.func_name)((1, 2, 3, 4, 5, 6, 7, 8, 9, 10))
        result = self.executor.execute_dataframe(r, concat=True)[0]
        expected = getattr(data1[2], self.func_name)((1, 2, 3, 4, 5, 6, 7, 8, 9, 10))
        pd.testing.assert_series_equal(expected, result)

        # specify index, not the default range index
        data1 = pd.DataFrame(np.random.rand(10, 7), index=np.arange(5, 15),
                             columns=[4, 1, 3, 2, 5, 6, 7])
        data1 = self.to_boolean_if_needed(data1)
        df1 = from_pandas(data1, chunk_size=6)
        s1 = df1[2]

        r = getattr(df1, self.func_name)(np.array([1, 2, 3, 4, 5, 6, 7, 8, 9, 10]), axis=0)
        result = self.executor.execute_dataframe(r, concat=True)[0]
        expected = getattr(data1, self.func_name)(np.array([1, 2, 3, 4, 5, 6, 7, 8, 9, 10]), axis=0)
        pd.testing.assert_frame_equal(expected, result)

        r = getattr(df1, self.func_name)(from_array(np.array([1, 2, 3, 4, 5, 6, 7, 8, 9, 10])), axis=0)
        result = self.executor.execute_dataframe(r, concat=True)[0]
        expected = getattr(data1, self.func_name)(np.array([1, 2, 3, 4, 5, 6, 7, 8, 9, 10]), axis=0)
        pd.testing.assert_frame_equal(expected, result)

        r = getattr(s1, self.func_name)(np.array([1, 2, 3, 4, 5, 6, 7, 8, 9, 10]))
        result = self.executor.execute_dataframe(r, concat=True)[0]
        expected = getattr(data1[2], self.func_name)(np.array([1, 2, 3, 4, 5, 6, 7, 8, 9, 10]))
        pd.testing.assert_series_equal(expected, result)

        r = getattr(s1, self.func_name)(from_array(np.array([1, 2, 3, 4, 5, 6, 7, 8, 9, 10])))
        result = self.executor.execute_dataframe(r, concat=True)[0]
        expected = getattr(data1[2], self.func_name)(np.array([1, 2, 3, 4, 5, 6, 7, 8, 9, 10]))
        pd.testing.assert_series_equal(expected, result)

class TestUnary(TestBase):
    def setUp(self):
        super().setUp()
        self.executor = ExecutorForTest()

    def testAbs(self):
        data1 = pd.DataFrame(np.random.uniform(low=-1, high=1, size=(10, 10)))
        df1 = from_pandas(data1, chunk_size=5)

        result = self.executor.execute_dataframe(df1.abs(), concat=True)[0]
        expected = data1.abs()
        pd.testing.assert_frame_equal(expected, result)

        result = self.executor.execute_dataframe(abs(df1), concat=True)[0]
        pd.testing.assert_frame_equal(expected, result)

    def testNot(self):
        data1 = pd.DataFrame(np.random.uniform(low=-1, high=1, size=(10, 10)) > 0)
        df1 = from_pandas(data1, chunk_size=5)

        result = self.executor.execute_dataframe(~df1, concat=True)[0]
        expected = ~data1
        pd.testing.assert_frame_equal(expected, result)

    def testUfunc(self):
        df_raw = pd.DataFrame(np.random.uniform(size=(10, 10)),
                              index=pd.RangeIndex(9, -1, -1))
        df = from_pandas(df_raw, chunk_size=5)

        series_raw = pd.Series(np.random.uniform(size=10),
                               index=pd.RangeIndex(9, -1, -1))
        series = from_pandas_series(series_raw, chunk_size=5)

        ufuncs = [
            [np.abs, mt.abs],
            [np.log, mt.log],
            [np.log2, mt.log2],
            [np.log10, mt.log10],
            [np.sin, mt.sin],
            [np.cos, mt.cos],
            [np.tan, mt.tan],
            [np.sinh, mt.sinh],
            [np.cosh, mt.cosh],
            [np.tanh, mt.tanh],
            [np.arcsin, mt.arcsin],
            [np.arccos, mt.arccos],
            [np.arctan, mt.arctan],
            [np.arcsinh, mt.arcsinh],
            [np.arccosh, mt.arccosh],
            [np.arctanh, mt.arctanh],
            [np.radians, mt.radians],
            [np.degrees, mt.degrees],
            [np.ceil, mt.ceil],
            [np.floor, mt.floor],
            [partial(np.around, decimals=2), partial(mt.around, decimals=2)],
            [np.exp, mt.exp],
            [np.exp2, mt.exp2],
            [np.expm1, mt.expm1],
            [np.sqrt, mt.sqrt]
        ]

        for raw, data in [(df_raw, df), (series_raw, series)]:
            for npf, mtf in ufuncs:
                r = mtf(data)

                result = self.executor.execute_tensor(r, concat=True)[0]
                expected = npf(raw)

                if isinstance(raw, pd.DataFrame):
                    pd.testing.assert_frame_equal(result, expected)
                else:
                    pd.testing.assert_series_equal(result, expected)

                # test numpy ufunc
                r = npf(data)

                result = self.executor.execute_tensor(r, concat=True)[0]

                if isinstance(raw, pd.DataFrame):
                    pd.testing.assert_frame_equal(result, expected)
                else:
                    pd.testing.assert_series_equal(result, expected)

    def testDateTimeBin(self):
        rs = np.random.RandomState(0)
        df_raw = pd.DataFrame({'a': rs.randint(1000, size=10),
                               'b': rs.rand(10),
                               'c': [pd.Timestamp(rs.randint(1604000000, 1604481373))
                                     for _ in range(10)]},
                              index=pd.RangeIndex(9, -1, -1))
        df = from_pandas(df_raw, chunk_size=5)
        r = (df['c'] > to_datetime('2000-01-01')) & (df['c'] < to_datetime('2021-01-01'))

        result = self.executor.execute_dataframe(r, concat=True)[0]
        expected = (df_raw['c'] > pd.to_datetime('2000-01-01')) & \
                   (df_raw['c'] < pd.to_datetime('2021-01-01'))
        pd.testing.assert_series_equal(result, expected)

if __name__ == '__main__':  # pragma: no cover
    unittest.main()

max_stars_repo_name: Focke-Wulf/mars

max_issues_repo_name: Focke-Wulf/mars

max_forks_repo_name: Focke-Wulf/mars


text:
import pandas as pd
import numpy as np
import os

import config

def compute_hlasy_pie_tlace():
    """Prepare data for the hlasy pie charts."""
    if not os.path.isfile(config.FILE_HLASY):
        print("File {} does not exist locally!".format(config.FILE_HLASY))
    if not os.path.isfile(config.FILE_HLASY_METADATA):
        print("File {} does not exist locally!".format(
            config.FILE_HLASY_METADATA))
    hlasy = pd.read_hdf(config.FILE_HLASY)
    meta = pd.read_hdf(config.FILE_HLASY_METADATA)

    meta = meta[meta["tlac"].notnull()]
    id_vysledky = {s: meta.index[meta["Vysledok"] == "Návrh " + s]
                   for s in config.HLASY_STATS_VYSLEDOK}
    sts = {
        vysledok: hlasy.loc[id_vysledky[vysledok]].apply(
            lambda x: x.value_counts()).fillna(0)
        for vysledok in id_vysledky
        }
    for vysledok in sts:
        sts[vysledok].to_hdf(config.FILE_STATS_HLASY_PIE, vysledok,
                             format="table")

def prepare_pie_demagog():
    """Extract only the necessary columns and politication from demagog."""
    if not os.path.isfile(config.FILE_KLUBY):
        print("File {} does not exist locally!".format(config.FILE_KLUBY))
    if not os.path.isfile(config.FILE_DEMAGOG):
        print("File {} does not exist locally!".format(
            config.FILE_DEMAGOG))
    kluby = pd.read_hdf(config.FILE_KLUBY)
    dg = pd.read_hdf(config.FILE_DEMAGOG)
    names = kluby.index
    dg = dg[dg["Meno"].isin(names)]
    df = pd.DataFrame(dg[config.DEMAGOG_LABELS].values,
                      columns=config.DEMAGOG_LABELS,
                      index=dg["Meno"].values).T
    df.to_hdf(config.FILE_DEMAGOG_PIE, config.HDF_KEY, format="table")

def prepare_stats_rozpravy():
    """Extract only the summary stats from rozpravy."""
    if not os.path.isfile(config.FILE_ROZPRAVY):
        print("File {} does not exist locally!".format(config.FILE_ROZPRAVY))
    rz = pd.read_pickle(config.FILE_ROZPRAVY)
    sts = pd.DataFrame()
    # DURATION
    dur = rz["end_time"] - rz["start_time"]
    dur[dur < pd.Timedelta(0)] += pd.Timedelta(days=1)  # issue with midnight
    rz[config.ROZPRAVY_APP_LABELS[0]] = dur
    sts[config.ROZPRAVY_APP_LABELS[0]] = rz.groupby(["name"])[
        config.ROZPRAVY_APP_LABELS[0]].sum()
    dur_sort = sts[config.ROZPRAVY_APP_LABELS[0]].argsort()
    dur_vals = sts[config.ROZPRAVY_APP_LABELS[0]][dur_sort]
    dur_vals.to_pickle(config.FILE_ROZPRAVY_DURATION)

    # N_WORDS
    rz[config.ROZPRAVY_APP_LABELS[1]] = rz["text"].apply(
        lambda x: len(x.split(" ")))
    sts[config.ROZPRAVY_APP_LABELS[1]] = rz.groupby(["name"])[
        config.ROZPRAVY_APP_LABELS[0]].sum()

ext: py

reason: import numpy


text:
# importing requisite libraries
from stockstats import StockDataFrame as sdf
import requests
from dotenv import load_dotenv
import os
import pandas as pd
import numpy as np
import alpaca_trade_api as tradeapi
from pathlib import Path

def fetch_ohlcv(ticker, start_date= "2020-01-01", end_date = "2021-01-01"):
    """Function to automatically pull closing price data from Alpaca"""
    """  example syntax: get_dat_data("TSLA")  """

    # Insert Personal Secret Keys (check .ENV to match syntax if error occurs)
    alpaca_api_key = os.getenv("APCA_API_KEY_ID")
    alpaca_secret_key = os.getenv("APCA_API_SECRET_KEY")
    # Create the Alpaca API object
    alpaca = tradeapi.REST(
        alpaca_api_key,
        alpaca_secret_key,
        api_version= "v2")

    # Backdating the Data to the first day of 2020
    back_date = pd.Timestamp(start_date, tz="America/New_York").isoformat()
    end_date = pd.Timestamp(end_date, tz="America/New_York").isoformat()

    # Choosing the ticker we want (need to figure out how to make this the input)
    #ticker = ["TSLA"]

    # Using our assumption of 15min intervals for trading
    timeframe = "15Min"

    # Creating a dataframe with every characteristic of ticker Alpaca allows
    df_ticker = alpaca.get_barset(
        ticker,
        timeframe,
        start=back_date,
        end = end_date
        ).df

    return df_ticker[ticker]

def bollinger_band_generator(dataframe_name, closing_price_column_name = 'close', bollinger_band_window = 20, num_standard_deviation = 2):
    """Creates Bollinger Band function
    Args:
        dataframe_name (dict): Single security dataframe containing at least closing prices
        closing_price_column_name (str): Name of column in dataframe containing closing prices
        bollinger_band_window (int): Desired timeframe window used for rolling calculations
        num_standard_deviation (int): Desired number of standard deviations to calculate
    Returns:
        A dataframe of:
            original data passed to function,
            bollinger_band_middle (flt): Column of values for middle band,
            bollinger_band_std (flt): Column of values to calculate standard deviation,
            bollinger_band_upper (flt): Column of values for upper band,
            bollinger_band_lower (flt): Column of values for lower band,
    """

    # Calculate mean and standard deviation
    dataframe_name['bollinger_band_middle'] = dataframe_name[closing_price_column_name].rolling(window=bollinger_band_window).mean()
    dataframe_name['bollinger_band_std'] = dataframe_name[closing_price_column_name].rolling(window=bollinger_band_window).std()

    # Calculate upper bollinger band and lower bollinger band
    dataframe_name['bollinger_band_upper'] = dataframe_name['bollinger_band_middle'] + (dataframe_name['bollinger_band_std'] * num_standard_deviation)
    dataframe_name['bollinger_band_lower'] = dataframe_name['bollinger_band_middle'] - (dataframe_name['bollinger_band_std'] * num_standard_deviation)

    # Drop NaN values
    dataframe_name.dropna(inplace=True)

    # Return dataframe with features and target
    return dataframe_name

def keltner_channel(df_ohlc, p_length = 20):

    stock = sdf.retype(df_ohlc)
    atr_column = pd.DataFrame(stock['atr'])
    df_final = df_ohlc
    df_final['atr'] = atr_column
    df_final['kcmid'] = df_final['close'].rolling(window=p_length).mean()
    df_final['kcup'] = df_final['kcmid']+df_final['atr']
    df_final['kclo'] = df_final['kcmid']-df_final['atr']

    return df_final

def ewma(dataframe_name, closing_price_column_name = 'close', fast_ema = 9, slow_ema = 21):
    # create EMAs columns
    dataframe_name['EMA9'] = dataframe_name[closing_price_column_name].ewm(span=fast_ema, adjust=False).mean()
    dataframe_name['EMA21'] = dataframe_name[closing_price_column_name].ewm(span=slow_ema, adjust=False).mean()

    # Return dataframe with features and target
    return dataframe_name

def adding_boll_kelt_ewma_dataframe(dataframe):
    lib_copy.bollinger_band_generator(dataframe)
    lib_copy.keltner_channel(dataframe)
    lib_copy.ewma(dataframe)

    return dataframe

def signals_generator(dataframe_name):
    """Creates signals for long position
    Args:
        dataframe_name (dict): Dataframe containing indicator data for Bollinger Bands, EWMA, and Keltner Channels
    Returns:
        A dataframe of:
            original data passed to function,
            appended squeeze column signals of type float (1.0 = True, 0.0 = False),
            appended ema crossup column signals of type float (1.0 = True, 0.0 = False),
            appended ema crossdown column signals of type float (-1.0 = True, 0.0 = False)
            appended io_target column signals of type float (2.0 = Buy, -1.0 = Sell)
    """

    # Create signal for bollinger band is inside keltner channel
    selection = dataframe_name.loc[((dataframe_name['bollinger_band_upper'] < dataframe_name['kcup']) & (dataframe_name['bollinger_band_lower'] >= dataframe_name['kclo'])), :].index
    dataframe_name['squeeze'] = 0.0
    dataframe_name['squeeze'][selection] = 1

    # Create signal for crossover band (cross in the up direction)
    selection2 = dataframe_name.loc[((dataframe_name['EMA9'] > dataframe_name['EMA21']) & (dataframe_name['EMA9'].shift(1) < dataframe_name['EMA21'].shift(1))), :].index
    dataframe_name['crossup'] = 0.0
    dataframe_name['crossup'][selection2] = 1

    # Create signal for crossover band (cross in the down direction)
    selection3 = dataframe_name.loc[((dataframe_name['EMA9'] < dataframe_name['EMA21']) & (dataframe_name['EMA9'].shift(1) > dataframe_name['EMA21'].shift(1))), :].index
    dataframe_name['crossdown'] = 0.0
    dataframe_name['crossdown'][selection3] = -1

    # Target generation (buy in)
    selection4 = dataframe_name.loc[((dataframe_name['squeeze'] == 1.0) & (dataframe_name['crossup'] == 1.0)), :].index
    dataframe_name['target'] = 0.0
    dataframe_name['target'][selection4] = 1

    # In and Out target generation
    for index, row in dataframe_name.iterrows():
        dataframe_name.loc[index, 'io_target'] = row['squeeze'] + row['crossup'] + row['crossdown']

    # Return dataframe with features and target
    return dataframe_name

def target_generator(dataframe_name, col_name1, col_name2, target_col_name):
    """Creates a target for long position
    Args:
        dataframe_name (dict): Dataframe containing indicator data (0's and 1's)
        col_name1 (str): Name of first column name in dataframe to use for calculation
        col_name2 (str): Name of second column name in dataframe to use for calculation
        target_col_name (str): Name of target column name to create and store target values
    Returns:
        A dataframe of:
            original data passed to function,
            appended target column signals of type float (2.0, 1.0, 0.0)
    """

    # Target generation
    for index, row in dataframe_name.iterrows():
        dataframe_name.loc[index, target_col_name] = row[col_name1] + row[col_name2]

    # Return dataframe with features and target
    return dataframe_name

def lstm(
    dataframe,
    num_feature_cols = 2,
    target_name = "target",
    epochs_num = 10,
    unit_number = 30,
    dropout_fraction=.2
    ):

    """
    Make sure to have target in last column
    """

    X = dataframe.iloc[:, 0:num_feature_cols].values

    y = dataframe.iloc[:,-1].values

    X, y = np.array(X), np.array(y).reshape(-1,1)

    # Manually splitting the data
    split = int(0.7 * len(X))

    X_train = X[: split]
    X_test = X[split:]

    y_train = y[: split]
    y_test = y[split:]

    # Importing the MinMaxScaler from sklearn
    from sklearn.preprocessing import MinMaxScaler

    # Create a MinMaxScaler object
    scaler = MinMaxScaler()

    # Fit the MinMaxScaler object with the features data X
    scaler.fit(X)

    # Scale the features training and testing sets
    X_train = scaler.transform(X_train)
    X_test = scaler.transform(X_test)

    # Fit the MinMaxScaler object with the target data Y
    scaler.fit(y)

    # Scale the target training and testing sets
    y_train = scaler.transform(y_train)
    y_test = scaler.transform(y_test)

    # Importing required Keras modules
    from tensorflow.keras.models import Sequential
    from tensorflow.keras.layers import LSTM, Dense, Dropout

    # Define the LSTM RNN model.
    model = Sequential()

    # Layer 1
    model.add(LSTM(
        units=unit_number,
        return_sequences=True,
        input_shape=(X_train.shape[1], 1))
        )
    model.add(Dropout(dropout_fraction))

    # Layer 2
    model.add(LSTM(units=unit_number, return_sequences=True))
    model.add(Dropout(dropout_fraction))

    # Layer 3
    model.add(LSTM(units=unit_number))
    model.add(Dropout(dropout_fraction))

    # Output layer
    model.add(Dense(1))

    # Compile the model
    model.compile(optimizer="adam", loss="mean_squared_error")

    # Train the model
    model.fit(X_train, y_train, epochs= epochs_num, shuffle=False, batch_size=90, verbose=1)

    # Make predictions using the testing data X_test
    predicted = model.predict(X_test)

    # Recover the original prices instead of the scaled version
    predicted_prices = scaler.inverse_transform(predicted)
    real_prices = scaler.inverse_transform(y_test.reshape(-1, 1))
    # Create a DataFrame of Real and Predicted values
    comparison = pd.DataFrame({
        "Actual Target": real_prices.ravel(),
        "Predicted Target": predicted_prices.ravel()
    }, index = dataframe.index[-len(real_prices): ])

    return model.summary(), model.evaluate(X_test, y_test, verbose=0), comparison.plot()

def lstm1(
    dataframe,
    num_feature_cols = 2,
    target_name = "target",
    epochs_num = 10,
    unit_number = 30,
    dropout_fraction=.2
    ):

    """
    Make sure to have target in last column
    """

    # Manually splitting the data

    # Construct training start and training end dates

    training_start = dataframe.index.min().strftime(format='%Y-%m-%d')
    training_end = '2019-01-11'

    # Construct test start and test end dates

    testing_start = '2019-01-12'
    testing_end = '2019-06-12'

    # Construct validating start and validating end dates

    vali_start = '2019-06-13'
    vali_end = '2020-01-12'

    # Construct the X_train and y_train datasets
    X_train = dataframe[["squeeze", "emax_signal"]][training_start:training_end]
    y_train = dataframe["target"][training_start:training_end]

    X_test = dataframe[["squeeze", "emax_signal"]][testing_start:testing_end]
    y_test = dataframe["target"][testing_start:testing_end]

    #X = dataframe.iloc[:, 0:num_feature_cols].values

    #y = dataframe.iloc[:,-1].values

    #X, y = np.array(X), np.array(y).reshape(-1,1)

    # Importing the MinMaxScaler from sklearn
    from sklearn.preprocessing import MinMaxScaler

    # Create a MinMaxScaler object
    scaler = MinMaxScaler()

    # Fit the MinMaxScaler object with the features data X
    scaler.fit(X_train)

    # Scale the features training and testing sets
    X_train = scaler.transform(X_train)
    X_test = scaler.transform(X_test)
    y_train = y_train.values.reshape(-1,1)
    y_test = y_test.values.reshape(-1,1)

    # Fit the MinMaxScaler object with the target data Y
    scaler.fit(y_train)

    # Scale the target training and testing sets
    y_train = scaler.transform(y_train)
    y_test = scaler.transform(y_test)

    # Importing required Keras modules
    from tensorflow.keras.models import Sequential
    from tensorflow.keras.layers import LSTM, Dense, Dropout

    # Define the LSTM RNN model.
    model = Sequential()
    X_train = X_train.reshape(-1,1,2)
    X_test = X_test.reshape(-1,1,2)
    #print(X_train[1])
    # Layer 1
    model.add(LSTM(
        units=unit_number,
        return_sequences=True,
        input_shape=(1,2)))

    model.add(Dropout(dropout_fraction))

    # Layer 2
    model.add(LSTM(units=unit_number, return_sequences=True))
    model.add(Dropout(dropout_fraction))

    # Layer 3
    model.add(LSTM(units=unit_number, return_sequences=True))
    model.add(Dropout(dropout_fraction))

    # Output layer
    model.add(Dense(1))

    # Compile the model
    model.compile(optimizer="adagrad", loss="binary_crossentropy")

    #print(model.summary())
    # Train the model
    model.fit(X_train, y_train, epochs= epochs_num, shuffle=False, batch_size=90, verbose=1)

    # Make predictions using the testing data X_test
    predicted = model.predict(X_test)
    #print(predicted)
    # Recover the original prices instead of the scaled version
    predicted_prices = scaler.inverse_transform(predicted.reshape(-1,1))
    real_prices = scaler.inverse_transform(y_test.reshape(-1, 1))
    # Create a DataFrame of Real and Predicted values
    comparison = pd.DataFrame({
        "Actual Target": real_prices.ravel(),
        "Predicted Target": predicted_prices.ravel()
    }, index = dataframe.index[-len(real_prices): ])

    return model.summary(), model.evaluate(X_test, y_test, verbose=0), comparison.plot()

def trade_strategy_modeling(all_signals):
    return True

def execute_backtest(trade_strategy):
    return True

def create_dash(metrics_from_backtest):
    return True

def main(connector):
    return True


text:
"""
Author(s): Matthew Loper

See LICENCE.txt for licensing and contact information.
"""

from .ch import Ch
import numpy as np
from .utils import row, col
import scipy.sparse as sp
import weakref

__all__ = ['sort', 'tile', 'repeat', 'transpose', 'rollaxis', 'swapaxes', 'reshape', 'Select',
           'atleast_1d', 'atleast_2d', 'atleast_3d', 'squeeze', 'expand_dims', 'fliplr', 'flipud',
           'concatenate', 'vstack', 'hstack', 'dstack', 'ravel', 'diag', 'diagflat', 'roll', 'rot90']

# Classes deriving from "Permute" promise to only reorder/reshape
class Permute(Ch):
    pass

def ravel(a, order='C'):
    assert(order=='C')
    if isinstance (a, np.ndarray):
        self = Ch(a)

    return reshape(a=a, newshape=(-1,))

class Reorder(Permute):
    dterms = 'a',

    def on_changed(self, which):
        if not hasattr(self, 'dr_lookup'):
            self.dr_lookup = {}

    def compute_r(self):
        return self.reorder(self.a.r)

    def compute_dr_wrt(self, wrt):
        if wrt is self.a:
            if False:
                from scipy.sparse.linalg.interface import LinearOperator
                return LinearOperator((self.size, wrt.size), lambda x : self.reorder(x.reshape(self.a.shape)).ravel())
            else:
                a = self.a
                asz = a.size
                ashape = a.shape
                key = self.unique_reorder_id()
                if key not in self.dr_lookup or key is None:
                    JS = self.reorder(np.arange(asz).reshape(ashape))
                    IS = np.arange(JS.size)
                    data = np.ones_like(IS)
                    shape = JS.shape
                    self.dr_lookup[key] = sp.csc_matrix((data, (IS, JS.ravel())), shape=(self.r.size, wrt.r.size))
                return self.dr_lookup[key]

class Sort(Reorder):
    dterms = 'a'
    terms = 'axis', 'kind', 'order'

    def reorder(self, a): return np.sort(a, self.axis, self.kind, self.order)
    def unique_reorder_id(self): return None

def sort(a, axis=-1, kind='quicksort', order=None):
    return Sort(a=a, axis=axis, kind=kind, order=order)

class Tile(Reorder):
    dterms = 'a',
    terms = 'reps',
    term_order = 'a', 'reps'

    def reorder(self, a): return np.tile(a, self.reps)
    def unique_reorder_id(self): return (self.a.shape, tuple(self.reps))

def tile(A, reps):
    return Tile(a=A, reps=reps)

class Diag(Reorder):
    dterms = 'a',
    terms = 'k',

    def reorder(self, a): return np.diag(a, self.k)
    def unique_reorder_id(self): return (self.a.shape, self.k)

def diag(v, k=0):
    return Diag(a=v, k=k)

class DiagFlat(Reorder):
    dterms = 'a',
    terms = 'k',

    def reorder(self, a): return np.diagflat(a, self.k)
    def unique_reorder_id(self): return (self.a.shape, self.k)

def diagflat(v, k=0):
    return DiagFlat(a=v, k=k)

class Repeat(Reorder):
    dterms = 'a',
    terms = 'repeats', 'axis'

    def reorder(self, a): return np.repeat(a, self.repeats, self.axis)
    def unique_reorder_id(self): return (self.repeats, self.axis)

def repeat(a, repeats, axis=None):
    return Repeat(a=a, repeats=repeats, axis=axis)

class transpose(Reorder):
    dterms = 'a'
    terms = 'axes'
    term_order = 'a', 'axes'

    def reorder(self, a):    return np.require(np.transpose(a, axes=self.axes), requirements='C')
    def unique_reorder_id(self): return (self.a.shape, None if self.axes is None else tuple(self.axes))
    def on_changed(self, which):
        if not hasattr(self, 'axes'):
            self.axes = None
        super(self.__class__, self).on_changed(which)

class rollaxis(Reorder):
    dterms = 'a'
    terms = 'axis', 'start'
    term_order = 'a', 'axis', 'start'

    def reorder(self, a):    return np.rollaxis(a, axis=self.axis, start=self.start)
    def unique_reorder_id(self): return (self.a.shape, self.axis, self.start)
    def on_changed(self, which):
        if not hasattr(self, 'start'):
            self.start = 0
        super(self.__class__, self).on_changed(which)

class swapaxes(Reorder):
    dterms = 'a'
    terms = 'axis1', 'axis2'
    term_order = 'a', 'axis1', 'axis2'

    def reorder(self, a):    return np.swapaxes(a, axis1=self.axis1, axis2=self.axis2)
    def unique_reorder_id(self): return (self.a.shape, self.axis1, self.axis2)

class Roll(Reorder):
    dterms = 'a',
    terms = 'shift', 'axis'
    term_order = 'a', 'shift', 'axis'

    def reorder(self, a): return np.roll(a, self.shift, self.axis)
    def unique_reorder_id(self): return (self.shift, self.axis)

def roll(a, shift, axis=None):
    return Roll(a, shift, axis)

class Rot90(Reorder):
    dterms = 'a',
    terms = 'k',

    def reorder(self, a): return np.rot90(a, self.k)
    def unique_reorder_id(self): return (self.a.shape, self.k)

def rot90(m, k=1):
    return Rot90(a=m, k=k)

class Reshape(Permute):
    dterms = 'a',
    terms = 'newshape',
    term_order= 'a', 'newshape'

    def compute_r(self):
        return self.a.r.reshape(self.newshape)

    def compute_dr_wrt(self, wrt):
        if wrt is self.a:
            return sp.eye(self.a.size, self.a.size)
        #return self.a.dr_wrt(wrt)

# def reshape(a, newshape):
#     if isinstance(a, Reshape) and a.newshape == newshape:
#         return a
#     return Reshape(a=a, newshape=newshape)
def reshape(a, newshape):
    while isinstance(a, Reshape):
        a = a.a
    return Reshape(a=a, newshape=newshape)

# class And(Ch):
#     dterms = 'x1', 'x2'
#
#     def compute_r(self):
#         if True:
#             needs_work = [self.x1, self.x2]
#             done = []
#             while len(needs_work) > 0:
#                 todo = needs_work.pop()
#                 if isinstance(todo, And):
#                     needs_work += [todo.x1, todo.x2]
#                 else:
#                     done = [todo] + done
#             return np.concatenate([d.r.ravel() for d in done])
#         else:
#             return np.concatenate((self.x1.r.ravel(), self.x2.r.ravel()))
#
#     # This is only here for reverse mode to work.
#     # Most of the time, the overridden dr_wrt is callpath gets used.
#     def compute_dr_wrt(self, wrt):
#
#         if wrt is not self.x1 and wrt is not self.x2:
#             return
#
#         input_len = wrt.r.size
#         x1_len = self.x1.r.size
#         x2_len = self.x2.r.size
#
#         mtxs = []
#         if wrt is self.x1:
#             mtxs.append(sp.spdiags(np.ones(x1_len), 0, x1_len, x1_len))
#         else:
#             mtxs.append(sp.csc_matrix((x1_len, input_len)))
#
#         if wrt is self.x2:
#             mtxs.append(sp.spdiags(np.ones(x2_len), 0, x2_len, x2_len))
#         else:
#             mtxs.append(sp.csc_matrix((x2_len, input_len)))
#
#
#         if any([sp.issparse(mtx) for mtx in mtxs]):
#             result = sp.vstack(mtxs, format='csc')
#         else:
#             result = np.vstack(mtxs)
#
#         return result
#
#     def dr_wrt(self, wrt, want_stacks=False, reverse_mode=False):
#         self._call_on_changed()
#
#         input_len = wrt.r.size
#         x1_len = self.x1.r.size
#         x2_len = self.x2.r.size
#
#         mtxs = []
#         if wrt is self.x1:
#             mtxs.append(sp.spdiags(np.ones(x1_len), 0, x1_len, x1_len))
#         else:
#             if isinstance(self.x1, And):
#                 tmp_mtxs = self.x1.dr_wrt(wrt, want_stacks=True, reverse_mode=reverse_mode)
#                 for mtx in tmp_mtxs:
#                     mtxs.append(mtx)
#             else:
#                 mtxs.append(self.x1.dr_wrt(wrt, reverse_mode=reverse_mode))
#             if mtxs[-1] is None:
#                 mtxs[-1] = sp.csc_matrix((x1_len, input_len))
#
#         if wrt is self.x2:
#             mtxs.append(sp.spdiags(np.ones(x2_len), 0, x2_len, x2_len))
#         else:
#             if isinstance(self.x2, And):
#                 tmp_mtxs = self.x2.dr_wrt(wrt, want_stacks=True, reverse_mode=reverse_mode)
#                 for mtx in tmp_mtxs:
#                     mtxs.append(mtx)
#             else:
#                 mtxs.append(self.x2.dr_wrt(wrt, reverse_mode=reverse_mode))
#             if mtxs[-1] is None:
#                 mtxs[-1] = sp.csc_matrix((x2_len, input_len))
#
#         if want_stacks:
#             return mtxs
#         else:
#             if any([sp.issparse(mtx) for mtx in mtxs]):
#                 result = sp.vstack(mtxs, format='csc')
#             else:
#                 result = np.vstack(mtxs)
#
#         return result

class Select(Permute):
    terms = ['idxs', 'preferred_shape']
    dterms = ['a']
    term_order = 'a', 'idxs', 'preferred_shape'

    def compute_r(self):
        result = self.a.r.ravel()[self.idxs].copy()
        if hasattr(self, 'preferred_shape'):
            return result.reshape(self.preferred_shape)
        else:
            return result

    def compute_dr_wrt(self, obj):
        if obj is self.a:
            if not hasattr(self, '_dr_cached'):
                IS = np.arange(len(self.idxs))
                JS = self.idxs.ravel()
                ij = np.vstack((row(IS), row(JS)))
                data = np.ones(len(self.idxs))
                self._dr_cached = sp.csc_matrix((data, ij), shape=(len(self.idxs), np.prod(self.a.shape)))
            return self._dr_cached

    def on_changed(self, which):
        if hasattr(self, '_dr_cached'):
            if 'idxs' in which or self.a.r.size != self._dr_cached.shape[1]:
                del self._dr_cached

class AtleastNd(Ch):
    dterms = 'x'
    terms = 'ndims'

    def compute_r(self):
        xr = self.x.r
        if self.ndims == 1:
            target_shape = np.atleast_1d(xr).shape
        elif self.ndims == 2:
            target_shape = np.atleast_2d(xr).shape
        elif self.ndims == 3:
            target_shape = np.atleast_3d(xr).shape
        else:
            raise Exception('Need ndims to be 1, 2, or 3.')

        return xr.reshape(target_shape)

    def compute_dr_wrt(self, wrt):
        if wrt is self.x:
            return 1

def atleast_nd(ndims, *arys):
    arys = [AtleastNd(x=ary, ndims=ndims) for ary in arys]
    return arys if len(arys) > 1 else arys[0]

def atleast_1d(*arys):
    return atleast_nd(1, *arys)

def atleast_2d(*arys):
    return atleast_nd(2, *arys)

def atleast_3d(*arys):
    return atleast_nd(3, *arys)

def squeeze(a, axis=None):
    if isinstance(a, np.ndarray):
        return np.squeeze(a, axis)
    shape = np.squeeze(a.r, axis).shape
    return a.reshape(shape)

def expand_dims(a, axis):
    if isinstance(a, np.ndarray):
        return np.expand_dims(a, axis)
    shape = np.expand_dims(a.r, axis).shape
    return a.reshape(shape)

def fliplr(m):
    return m[:,::-1]

def flipud(m):
    return m[::-1,...]

class Concatenate(Ch):

    def on_changed(self, which):
        if not hasattr(self, 'dr_cached'):
            self.dr_cached = weakref.WeakKeyDictionary()

    @property
    def our_terms(self):
        if not hasattr(self, '_our_terms'):
            self._our_terms = [getattr(self, s) for s in self.dterms]
        return self._our_terms

    def __getstate__(self):
        # Have to get rid of WeakKeyDictionaries for serialization
        if hasattr(self, 'dr_cached'):
            del self.dr_cached
        return super(self.__class__, self).__getstate__()

    def compute_r(self):
        return np.concatenate([t.r for t in self.our_terms], axis=self.axis)

    @property
    def everything(self):
        if not hasattr(self, '_everything'):
            self._everything = np.arange(self.r.size).reshape(self.r.shape)
            self._everything = np.swapaxes(self._everything, self.axis, 0)
        return self._everything

    def compute_dr_wrt(self, wrt):
        if not hasattr(self, 'dr_cached'):
            self.dr_cached = weakref.WeakKeyDictionary()
        if wrt in self.dr_cached and self.dr_cached[wrt] is not None:
            return self.dr_cached[wrt]

        if wrt not in self.our_terms:
            return

        _JS = np.arange(wrt.size)
        _data = np.ones(wrt.size)

        IS = []
        JS = []
        data = []

        offset = 0
        for term in self.our_terms:
            tsz = term.shape[self.axis]
            if term is wrt:
                JS += [_JS]
                data += [_data]
                IS += [np.swapaxes(self.everything[offset:offset+tsz], self.axis, 0).ravel()]
            offset += tsz
        IS   = np.concatenate(IS).ravel()
        JS   = np.concatenate(JS).ravel()
        data = np.concatenate(data)

        res = sp.csc_matrix((data, (IS, JS)), shape=(self.r.size, wrt.size))

        if len(list(self._parents.keys())) != 1:
            self.dr_cached[wrt] = res
        else:
            self.dr_cached[wrt] = None

        return res

def expand_concatenates(mtxs, axis=0):
    mtxs = list(mtxs)
    done = []
    while len(mtxs) > 0:
        mtx = mtxs.pop(0)
        if isinstance(mtx, Concatenate) and mtx.axis == axis:
            mtxs = [getattr(mtx, s) for s in mtx.dterms] + mtxs
        else:
            done.append(mtx)
    return done

def concatenate(mtxs, axis=0, **kwargs):

    mtxs = expand_concatenates(mtxs, axis)

    result = Concatenate(**kwargs)
    result.dterms = []
    for i, mtx in enumerate(mtxs):
        result.dterms.append('m%d' % (i,))
        setattr(result, result.dterms[-1], mtx)
    result.axis = axis
    return result

def hstack(mtxs, **kwargs):
    return concatenate(mtxs, axis=1, **kwargs)

def vstack(mtxs, **kwargs):
    return concatenate([atleast_2d(m) for m in mtxs], axis=0, **kwargs)

def dstack(mtxs, **kwargs):
    return concatenate([atleast_3d(m) for m in mtxs], axis=2, **kwargs)

max_stars_repo_name: jwgu/chumpy

max_issues_repo_name: jwgu/chumpy

max_forks_repo_name: jwgu/chumpy

reason: import numpy,import scipy,from scipy


text:
import math, pickle, os
import numpy as np
import torch
from torch.autograd import Variable
from torch import optim
from torch.utils.data import DataLoader
import torch.nn as nn
import torch.nn.functional as F
from dsp import *
import sys
import time
from wavernn import WaveRNN
from upsample import UpsampleNetwork
import env as env
import logger as logger

class Model(nn.Module) :
    def __init__(self, rnn_dims, fc_dims, pad, upsample_factors, feat_dims):
        super().__init__()
        self.n_classes = 256
        self.upsample = UpsampleNetwork(feat_dims, upsample_factors)
        self.wavernn = WaveRNN(rnn_dims, fc_dims, feat_dims, 0)
        self.num_params()

    def forward(self, x, mels) :
        #logger.log(f'x: {x.size()} mels: {mels.size()}')
        cond = self.upsample(mels)
        #logger.log(f'cond: {cond.size()}')
        return self.wavernn(x, cond.transpose(1, 2), None, None, None)

    def after_update(self):
        self.wavernn.after_update()

    def preview_upsampling(self, mels) :
        return self.upsample(mels)

    def forward_generate(self, mels, deterministic=False, use_half=False, verbose=False):
        n = mels.size(0)
        if use_half:
            mels = mels.half()
        self.eval()
        with torch.no_grad():
            cond = self.upsample(mels)
            output = self.wavernn.generate(cond.transpose(1, 2), None, None, None, use_half=use_half, verbose=verbose)
        self.train()
        return output

    def num_params(self):
        parameters = filter(lambda p: p.requires_grad, self.parameters())
        parameters = sum([np.prod(p.size()) for p in parameters]) / 1_000_000
        logger.log('Trainable Parameters: %.3f million' % parameters)

    def load_state_dict(self, dict):
        return super().load_state_dict(upgrade_state_dict(dict))

    def do_train(self, paths, dataset, optimiser, epochs, batch_size, step, lr=1e-4, valid_index=[], use_half=False):
        if use_half:
            import apex
            optimiser = apex.fp16_utils.FP16_Optimizer(optimiser, dynamic_loss_scale=True)
        for p in optimiser.param_groups: p['lr'] = lr
        criterion = nn.NLLLoss().cuda()
        k = 0
        saved_k = 0

        for e in range(epochs):
            trn_loader = DataLoader(dataset, collate_fn=lambda batch: env.collate(0, 16, 0, batch),
                                    batch_size=batch_size,
                                    num_workers=2, shuffle=True, pin_memory=True)
            start = time.time()
            running_loss_c = 0.
            running_loss_f = 0.

            iters = len(trn_loader)

            for i, (mels, coarse, fine, coarse_f, fine_f) in enumerate(trn_loader) :

                mels, coarse, fine, coarse_f, fine_f = mels.cuda(), coarse.cuda(), fine.cuda(), coarse_f.cuda(), fine_f.cuda()
                coarse, fine, coarse_f, fine_f = [t[:, hop_length:1-hop_length] for t in [coarse, fine, coarse_f, fine_f]]
                if use_half:
                    mels = mels.half()
                    coarse_f = coarse_f.half()
                    fine_f = fine_f.half()

                x = torch.cat([coarse_f[:, :-1].unsqueeze(-1), fine_f[:, :-1].unsqueeze(-1), coarse_f[:, 1:].unsqueeze(-1)], dim=2)

                p_c, p_f, _h_n = self(x, mels)
                loss_c = criterion(p_c.transpose(1, 2).float(), coarse[:, 1:])
                loss_f = criterion(p_f.transpose(1, 2).float(), fine[:, 1:])
                loss = loss_c + loss_f

                optimiser.zero_grad()
                if use_half:
                    optimiser.backward(loss)
                else:
                    loss.backward()
                optimiser.step()
                running_loss_c += loss_c.item()
                running_loss_f += loss_f.item()

                self.after_update()

                speed = (i + 1) / (time.time() - start)
                avg_loss_c = running_loss_c / (i + 1)
                avg_loss_f = running_loss_f / (i + 1)

                step += 1
                k = step // 1000
                logger.status(f'Epoch: {e+1}/{epochs} -- Batch: {i+1}/{iters} -- Loss: c={avg_loss_c:#.4} f={avg_loss_f:#.4} -- Speed: {speed:#.4} steps/sec -- Step: {k}k ')

            os.makedirs(paths.checkpoint_dir, exist_ok=True)
            torch.save(self.state_dict(), paths.model_path())
            np.save(paths.step_path(), step)
            logger.log_current_status()
            logger.log(f' <saved>; w[0][0] = {self.wavernn.gru.weight_ih_l0[0][0]}')
            if k > saved_k + 50:
                torch.save(self.state_dict(), paths.model_hist_path(step))
                saved_k = k
                self.do_generate(paths, step, dataset.path, valid_index, use_half=use_half)

    def do_generate(self, paths, step, data_path, test_index, deterministic=False, use_half=False, verbose=False):
        k = step // 1000
        test_mels = [np.load(f'{data_path}/mel/{id}.npy') for id in test_index]
        maxlen = max([x.shape[1] for x in test_mels])
        aligned = [torch.cat([torch.FloatTensor(x), torch.zeros(80, maxlen-x.shape[1]+1)], dim=1) for x in test_mels]
        out = self.forward_generate(torch.stack(aligned).cuda(), deterministic, use_half=use_half, verbose=verbose)

        os.makedirs(paths.gen_path(), exist_ok=True)
        for i, id in enumerate(test_index):
            gt = np.load(f'{data_path}/quant/{id}.npy')
            gt = (gt.astype(np.float32) + 0.5) / (2**15 - 0.5)
            librosa.output.write_wav(f'{paths.gen_path()}/{k}k_steps_{i}_target.wav', gt, sr=sample_rate)
            audio = out[i][:len(gt)].cpu().numpy()
            librosa.output.write_wav(f'{paths.gen_path()}/{k}k_steps_{i}_generated.wav', audio, sr=sample_rate)

def upgrade_state_dict(state_dict):
    out_dict = {}
    for key, val in state_dict.items():
        if key in UPGRADE_KEY:
            key = UPGRADE_KEY[key]
        out_dict[key] = val
    return out_dict

UPGRADE_KEY = {
        "rnn.weight_ih_l0": "wavernn.gru.weight_ih_l0",
        "rnn.weight_hh_l0": "wavernn.gru.weight_hh_l0",
        "rnn.bias_ih_l0": "wavernn.gru.bias_ih_l0",
        "rnn.bias_hh_l0": "wavernn.gru.bias_hh_l0",
        "fc1.weight": "wavernn.fc1.weight",
        "fc1.bias": "wavernn.fc1.bias",
        "fc2.weight": "wavernn.fc2.weight",
        "fc2.bias": "wavernn.fc2.bias",
        "fc3.weight": "wavernn.fc3.weight",
        "fc3.bias": "wavernn.fc3.bias",
        "fc4.weight": "wavernn.fc4.weight",
        "fc4.bias": "wavernn.fc4.bias",
        }


text:
import qrcode
from PIL import Image
import numpy as np
from .common import OpenApi

def Multiply(img1,img2):
	img1 = img1 / 255.0
	img2 = img2 / 255.0
	res = np.multiply(img1,img2)
	res = (res*255).astype(np.uint8)
	return res

def get_qrcode(info):
	qr = qrcode.QRCode(
		version=3,
		error_correction=qrcode.constants.ERROR_CORRECT_Q,
		box_size=10,
		border=4
	)
	qr.add_data(info)
	qr.make(fit=True)
	img = qr.make_image(fill_color="black", back_color="white")
	img = img.convert("RGBA")
	return img

def get_logo_qrcode(info, logo_path):
	img = get_qrcode(info)
	logo = Image.open(logo_path)
	img_w, img_h = img.size
	factor = 4
	size_w = int(img_w / factor)
	size_h = int(img_h / factor)

	logo_w, logo_h = logo.size
	if logo_w > size_w:
		logo_w = size_w
	if logo_h > size_h:
		logo_h = size_h
	logo = logo.resize((logo_w, logo_h), Image.ANTIALIAS)

	w = int((img_w - logo_w) / 2)
	h = int((img_h - logo_h) / 2)
	logo = logo.convert("RGBA")
	img.paste(logo, (w, h), logo)
	return img

def get_bg_qrcode(info, bg_path):
	img = get_qrcode(info)
	bg = Image.open(bg_path)
	img = img.convert("RGB")
	img_w, img_h = img.size
	size_w = img_w
	size_h = img_h

	np_img = np.array(img)
	np_mask = 1 - np_img / 255.0

	bg = bg.resize((size_w, size_h), Image.ANTIALIAS)
	bg = bg.convert("RGB")

	np_bg = np.array(bg)

	np_res = Multiply(np_img, np_bg)
	return Image.fromarray(np_res)

def save_qrcode(qr_code, image_name):
	qr_code.save(image_name)
	return image_name

def get_qrcode_return_info(username, info):
	op = OpenApi()
	try:
		qr_code = get_qrcode(info)
		image_name = "simple.png"
		qr_code.save(image_name)
		op.data = {"imgURL":""}
	except Exception as e:
		op.code = op.error_50001
		op.error = "请求错误，未获得词频统计"
		op.message = e
	return op.body

if __name__ == '__main__':
	save_qrcode(get_bg_qrcode("https://quanfita.cn", 'tmp.jpg'), 'bg_qr.png')
	save_qrcode(get_logo_qrcode("https://quanfita.cn", 'tmp.jpg'), 'logo_qr.png')
	save_qrcode(get_qrcode("https://quanfita.cn"), 'qr.png')

max_stars_repo_name: Quanfita/QTechCode

max_issues_repo_name: Quanfita/QTechCode

max_forks_repo_name: Quanfita/QTechCode