text:
# Copyright 2020 Huawei Technologies Co., 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.
# ============================================================================
"""test Serving pipeline with client"""

import numpy as np

from common import start_serving_server
from common import serving_test, create_client

@serving_test
def test_call_model_two_input_one_output_normal_success():
    servable_content = r"""
import numpy as np
from mindspore_serving.server import register
model = register.declare_model(model_file="tensor_add.mindir", model_format="MindIR", with_batch_dim=False)

def call_model(x1, x2):
    y = model.call(x1, x2)
    return y

@register.register_method(output_names="y")
def predict(x1, x2):
    y = register.add_stage(call_model, x1, x2, outputs_count=1)
    return y
    """
    base = start_serving_server(servable_content)
    # Client
    x1 = np.array([[1.1, 2.2], [3.3, 4.4]], np.float32)
    x2 = np.array([[5.5, 6.6], [7.7, 8.8]], np.float32)
    y = x1 + x2
    instances = [{"x1": x1, "x2": x2}]

    client = create_client("localhost:5500", base.servable_name, "predict")
    result = client.infer(instances)
    print("result", result)
    assert (result[0]["y"] == y).all()

@serving_test
def test_call_model_two_input_one_output_multi_times_success():
    servable_content = r"""
import numpy as np
from mindspore_serving.server import register
model = register.declare_model(model_file="tensor_add.mindir", model_format="MindIR", with_batch_dim=False)

def call_model(x1, x2, x3, x4):
    y1 = model.call(x1, x2)
    y2 = model.call(x3, x4)
    return y1 + y2

@register.register_method(output_names="y")
def predict(x1, x2, x3, x4):
    y = register.add_stage(call_model, x1, x2, x3, x4, outputs_count=1)
    return y
    """
    base = start_serving_server(servable_content)
    # Client
    x1 = np.array([[1.1, 2.2], [3.3, 4.4]], np.float32)
    x2 = np.array([[5.5, 6.6], [7.7, 8.8]], np.float32)
    x3 = np.array([[2.1, 3.2], [4.3, 5.4]], np.float32)
    x4 = np.array([[3.5, 4.6], [5.7, 6.8]], np.float32)
    y = x1 + x2 + x3 + x4
    instances = [{"x1": x1, "x2": x2, "x3": x3, "x4": x4}]

    client = create_client("localhost:5500", base.servable_name, "predict")
    result = client.infer(instances)
    print("result", result)
    assert (result[0]["y"] == y).all()

@serving_test
def test_call_model_two_input_one_output_multi_times_2success():
    servable_content = r"""
import numpy as np
from mindspore_serving.server import register
model = register.declare_model(model_file="tensor_add.mindir", model_format="MindIR", with_batch_dim=False)

def call_model(x1, x2, x3, x4):
    y1 = model.call(x1, x2)
    y2 = model.call(x3, x4)
    y = model.call(y1, y2)
    return y

@register.register_method(output_names="y")
def predict(x1, x2, x3, x4):
    y = register.add_stage(call_model, x1, x2, x3, x4, outputs_count=1)
    return y
    """
    base = start_serving_server(servable_content)
    # Client
    x1 = np.array([[1.1, 2.2], [3.3, 4.4]], np.float32)
    x2 = np.array([[5.5, 6.6], [7.7, 8.8]], np.float32)
    x3 = np.array([[2.1, 3.2], [4.3, 5.4]], np.float32)
    x4 = np.array([[3.5, 4.6], [5.7, 6.8]], np.float32)
    y = x1 + x2 + x3 + x4
    instances = [{"x1": x1, "x2": x2, "x3": x3, "x4": x4}]

    client = create_client("localhost:5500", base.servable_name, "predict")
    result = client.infer(instances)
    print("result", result)
    assert (result[0]["y"] == y).all()

@serving_test
def test_call_model_two_input_one_output_batch_call_success():
    servable_content = r"""
import numpy as np
from mindspore_serving.server import register
model = register.declare_model(model_file="tensor_add.mindir", model_format="MindIR", with_batch_dim=False)

def call_model(x1, x2, x3, x4):
    instances = []
    instances.append([x1, x2])
    instances.append((x3, x4))
    outputs  = model.call(instances) # return [[x1+x2], [x3+x4]]
    y1 = outputs[0][0]
    y2 = outputs[1][0]

    instances = []
    instances.append((y1, y2))
    outputs = model.call(instances)
    y = outputs[0][0]
    return y

@register.register_method(output_names="y")
def predict(x1, x2, x3, x4):
    y = register.add_stage(call_model, x1, x2, x3, x4, outputs_count=1)
    return y
    """
    base = start_serving_server(servable_content)
    # Client
    x1 = np.array([[1.1, 2.2], [3.3, 4.4]], np.float32)
    x2 = np.array([[5.5, 6.6], [7.7, 8.8]], np.float32)
    x3 = np.array([[2.1, 3.2], [4.3, 5.4]], np.float32)
    x4 = np.array([[3.5, 4.6], [5.7, 6.8]], np.float32)
    y = x1 + x2 + x3 + x4
    instances = [{"x1": x1, "x2": x2, "x3": x3, "x4": x4}]

    client = create_client("localhost:5500", base.servable_name, "predict")
    result = client.infer(instances)
    print("result", result)
    assert (result[0]["y"] == y).all()

@serving_test
def test_call_model_batch_call_one_input_one_output_success():
    servable_content = r"""
import numpy as np
from mindspore_serving.server import register
model = register.declare_model(model_file="tensor_add_1_1.mindir", model_format="MindIR", with_batch_dim=False)

def call_model(x1, x2, x3, x4):
    instances = []
    instances.append([x1])
    instances.append([x2])
    instances.append([x3])
    outputs = model.call(instances)
    y1 = outputs[0][0]
    y2 = outputs[1][0]
    y3 = outputs[2][0]
    y4 = model.call(x4)
    return y1+y2+y3+y4

@register.register_method(output_names="y")
def predict(x1, x2, x3, x4):
    y = register.add_stage(call_model, x1, x2, x3, x4, outputs_count=1)
    return y
    """
    base = start_serving_server(servable_content, model_file="tensor_add_1_1.mindir")
    # Client
    x1 = np.array([[1.1, 2.2], [3.3, 4.4]], np.float32)
    x2 = np.array([[5.5, 6.6], [7.7, 8.8]], np.float32)
    x3 = np.array([[2.1, 3.2], [4.3, 5.4]], np.float32)
    x4 = np.array([[3.5, 4.6], [5.7, 6.8]], np.float32)
    y = x1 + x2 + x3 + x4
    instances = [{"x1": x1, "x2": x2, "x3": x3, "x4": x4}]

    client = create_client("localhost:5500", base.servable_name, "predict")
    result = client.infer(instances)
    print("result", result)
    assert (result[0]["y"] == y).all()

@serving_test
def test_call_model_batch_call_one_input_two_output_success():
    servable_content = r"""
import numpy as np
from mindspore_serving.server import register
model = register.declare_model(model_file="tensor_add_1_2.mindir", model_format="MindIR", with_batch_dim=False)

def call_model(x1, x2, x3, x4):
    _, y1 = model.call(x1) # one instance
    _, y2 = model.call(x2) # one instance
    _, y3 = model.call(x3) # one instance
    _, y4 = model.call(x4) # one instance
    return y1+y2+y3+y4

@register.register_method(output_names="y")
def predict(x1, x2, x3, x4):
    y = register.add_stage(call_model, x1, x2, x3, x4, outputs_count=1)
    return y
    """
    base = start_serving_server(servable_content, model_file="tensor_add_1_2.mindir")
    # Client
    x1 = np.array([[1.1, 2.2], [3.3, 4.4]], np.float32)
    x2 = np.array([[5.5, 6.6], [7.7, 8.8]], np.float32)
    x3 = np.array([[2.1, 3.2], [4.3, 5.4]], np.float32)
    x4 = np.array([[3.5, 4.6], [5.7, 6.8]], np.float32)
    y = x1 + x2 + x3 + x4 + 4
    instances = [{"x1": x1, "x2": x2, "x3": x3, "x4": x4}]

    client = create_client("localhost:5500", base.servable_name, "predict")
    result = client.infer(instances)
    print("result", result)
    assert (result[0]["y"] == y).all()

@serving_test
def test_call_model_batch_call_one_input_two_output_batch_call_success():
    servable_content = r"""
import numpy as np
from mindspore_serving.server import register
model = register.declare_model(model_file="tensor_add_1_2.mindir", model_format="MindIR", with_batch_dim=False)

def call_model(x1, x2, x3, x4):
    instances = []
    instances.append([x1]) # one input
    outputs = model.call(instances) # batch call, one instance
    _, y1 = outputs[0]

    instances = []
    instances.append([x2]) # one input
    outputs = model.call(instances) # batch call, one instance
    _, y2 = outputs[0]

    instances = []
    instances.append([x3]) # one input
    instances.append([x4])
    outputs = model.call(instances) # batch call, two instances
    _, y3 = outputs[0]
    _, y4 = outputs[1]

    return y1+y2+y3+y4

@register.register_method(output_names="y")
def predict(x1, x2, x3, x4):
    y = register.add_stage(call_model, x1, x2, x3, x4, outputs_count=1)
    return y
    """
    base = start_serving_server(servable_content, model_file="tensor_add_1_2.mindir")
    # Client
    x1 = np.array([[1.1, 2.2], [3.3, 4.4]], np.float32)
    x2 = np.array([[5.5, 6.6], [7.7, 8.8]], np.float32)
    x3 = np.array([[2.1, 3.2], [4.3, 5.4]], np.float32)
    x4 = np.array([[3.5, 4.6], [5.7, 6.8]], np.float32)
    y = x1 + x2 + x3 + x4 + 4
    instances = [{"x1": x1, "x2": x2, "x3": x3, "x4": x4}]

    client = create_client("localhost:5500", base.servable_name, "predict")
    result = client.infer(instances)
    print("result", result)
    assert (result[0]["y"] == y).all()

@serving_test
def test_call_model_two_input_one_output_none_instances_failed():
    servable_content = r"""
import numpy as np
from mindspore_serving.server import register
model = register.declare_model(model_file="tensor_add.mindir", model_format="MindIR", with_batch_dim=False)

def call_model(x1, x2):
    y = model.call()
    return y

@register.register_method(output_names="y")
def predict(x1, x2):
    y = register.add_stage(call_model, x1, x2, outputs_count=1)
    return y
    """
    base = start_serving_server(servable_content)
    # Client
    x1 = np.array([[1.1, 2.2], [3.3, 4.4]], np.float32)
    x2 = np.array([[5.5, 6.6], [7.7, 8.8]], np.float32)
    instances = [{"x1": x1, "x2": x2}]

    client = create_client("localhost:5500", base.servable_name, "predict")
    result = client.infer(instances)
    print("result", result)
    assert "Model(tensor_add.mindir).call() failed: no inputs provided" in result["error"]

@serving_test
def test_call_model_two_input_one_output_zero_instances_failed():
    servable_content = r"""
import numpy as np
from mindspore_serving.server import register
model = register.declare_model(model_file="tensor_add.mindir", model_format="MindIR", with_batch_dim=False)

def call_model(x1, x2):
    y = model.call([])
    return y

@register.register_method(output_names="y")
def predict(x1, x2):
    y = register.add_stage(call_model, x1, x2, outputs_count=1)
    return y
    """
    base = start_serving_server(servable_content)
    # Client
    x1 = np.array([[1.1, 2.2], [3.3, 4.4]], np.float32)
    x2 = np.array([[5.5, 6.6], [7.7, 8.8]], np.float32)
    instances = [{"x1": x1, "x2": x2}]

    client = create_client("localhost:5500", base.servable_name, "predict")
    result = client.infer(instances)
    print("result", result)
    assert "Model(tensor_add.mindir).call() failed: Input instances count is 0" in result["error"]

@serving_test
def test_call_model_two_input_one_output_invalid_inputs_format_failed():
    servable_content = r"""
import numpy as np
from mindspore_serving.server import register
model = register.declare_model(model_file="tensor_add.mindir", model_format="MindIR", with_batch_dim=False)

def call_model(x1, x2):
    y = model.call([x1, x2]) # expect to be model.call([[x1, x2]])
    return y

@register.register_method(output_names="y")
def predict(x1, x2):
    y = register.add_stage(call_model, x1, x2, outputs_count=1)
    return y
    """
    base = start_serving_server(servable_content)
    # Client
    x1 = np.array([[1.1, 2.2], [3.3, 4.4]], np.float32)
    x2 = np.array([[5.5, 6.6], [7.7, 8.8]], np.float32)
    instances = [{"x1": x1, "x2": x2}]

    client = create_client("localhost:5500", base.servable_name, "predict")
    result = client.infer(instances)
    print("result", result)
    assert "inputs format invalid" in result["error"]

@serving_test
def test_call_model_two_input_one_output_zero_inputs_failed():
    servable_content = r"""
import numpy as np
from mindspore_serving.server import register
model = register.declare_model(model_file="tensor_add.mindir", model_format="MindIR", with_batch_dim=False)

def call_model(x1, x2):
    y = model.call([[]])
    return y

@register.register_method(output_names="y")
def predict(x1, x2):
    y = register.add_stage(call_model, x1, x2, outputs_count=1)
    return y
    """
    base = start_serving_server(servable_content)
    # Client
    x1 = np.array([[1.1, 2.2], [3.3, 4.4]], np.float32)
    x2 = np.array([[5.5, 6.6], [7.7, 8.8]], np.float32)
    instances = [{"x1": x1, "x2": x2}]

    client = create_client("localhost:5500", base.servable_name, "predict")
    result = client.infer(instances)
    print("result", result)
    assert "The inputs count 0 of instance 0 is not equal to the inputs count 2 of the model" in result["error"]

@serving_test
def test_call_model_two_input_one_output_data_size_error_failed():
    servable_content = r"""
import numpy as np
from mindspore_serving.server import register
model = register.declare_model(model_file="tensor_add.mindir", model_format="MindIR", with_batch_dim=False)

def call_model(x1, x2):
    y = model.call(x1, x2)
    return y

@register.register_method(output_names="y")
def predict(x1, x2):
    y = register.add_stage(call_model, x1, x2, outputs_count=1)
    return y
    """
    base = start_serving_server(servable_content)
    # Client
    x1 = np.array([[1.1, 2.2, 3.3], [3.3, 4.4, 5.5]], np.float32)
    x2 = np.array([[5.5, 6.6, 7.7], [7.7, 8.8, 8.8]], np.float32)
    instances = [{"x1": x1, "x2": x2}]

    client = create_client("localhost:5500", base.servable_name, "predict")
    result = client.infer(instances)
    print("result", result)
    assert "Given model input 0 size 24 not match the size 16 defined in model" in result["error"]

@serving_test
def test_call_model_two_input_one_output_data_type_error_failed():
    servable_content = r"""
import numpy as np
from mindspore_serving.server import register
model = register.declare_model(model_file="tensor_add.mindir", model_format="MindIR", with_batch_dim=False)

def call_model(x1, x2):
    y = model.call(x1, x2)
    return y

@register.register_method(output_names="y")
def predict(x1, x2):
    y = register.add_stage(call_model, x1, x2, outputs_count=1)
    return y
    """
    base = start_serving_server(servable_content)
    # Client
    x1 = np.array([[1.1, 2.2], [3.3, 4.4]], np.int32)
    x2 = np.array([[5.5, 6.6], [7.7, 8.8]], np.int32)
    instances = [{"x1": x1, "x2": x2}]

    client = create_client("localhost:5500", base.servable_name, "predict")
    result = client.infer(instances)
    print("result", result)
    assert "Given model input 0 data type kMSI_Int32 not match the data type kMSI_Float32 defined in model" in \
           result["error"]

@serving_test
def test_call_model_two_input_one_output_call_batch_data_size_error_failed():
    servable_content = r"""
import numpy as np
from mindspore_serving.server import register
model = register.declare_model(model_file="tensor_add.mindir", model_format="MindIR", with_batch_dim=False)

def call_model(x1, x2, x3, x4):
    instances = []
    instances.append((x1, x2))
    instances.append((x3, x4))
    ys = model.call(instances)
    return ys[0][0] + ys[1][0]

@register.register_method(output_names="y")
def predict(x1, x2, x3, x4):
    y = register.add_stage(call_model, x1, x2, x3, x4, outputs_count=1)
    return y
    """
    base = start_serving_server(servable_content)
    # Client
    x1 = np.array([[1.1, 2.2, 3.3], [3.3, 4.4, 5.5]], np.float32)
    x2 = np.array([[5.5, 6.6, 7.7], [7.7, 8.8, 8.8]], np.float32)
    x3 = np.array([[1.1, 2.2, 3.3], [3.3, 4.4, 5.5]], np.float32)
    x4 = np.array([[5.5, 6.6, 7.7], [7.7, 8.8, 8.8]], np.float32)
    instances = [{"x1": x1, "x2": x2, "x3": x3, "x4": x4}]

    client = create_client("localhost:5500", base.servable_name, "predict")
    result = client.infer(instances)
    print("result", result)
    assert "Given model input 0 size 24 not match the size 16 defined in model" in result["error"]

@serving_test
def test_call_model_two_input_one_output_call_batch_data_type_error_failed():
    servable_content = r"""
import numpy as np
from mindspore_serving.server import register
model = register.declare_model(model_file="tensor_add.mindir", model_format="MindIR", with_batch_dim=False)

def call_model(x1, x2, x3, x4):
    instances = []
    instances.append((x1, x2))
    instances.append((x3, x4))
    ys = model.call(instances)
    return ys[0][0] + ys[1][0]

@register.register_method(output_names="y")
def predict(x1, x2, x3, x4):
    y = register.add_stage(call_model, x1, x2, x3, x4, outputs_count=1)
    return y
    """
    base = start_serving_server(servable_content)
    # Client
    x1 = np.array([[1.1, 2.2], [3.3, 4.4]], np.int32)
    x2 = np.array([[5.5, 6.6], [7.7, 8.8]], np.int32)
    x3 = np.array([[1.1, 2.2], [3.3, 4.4]], np.int32)
    x4 = np.array([[5.5, 6.6], [7.7, 8.8]], np.int32)
    instances = [{"x1": x1, "x2": x2, "x3": x3, "x4": x4}]

    client = create_client("localhost:5500", base.servable_name, "predict")
    result = client.infer(instances)
    print("result", result)
    assert "Given model input 0 data type kMSI_Int32 not match the data type kMSI_Float32 defined in model" in \
           result["error"]

@serving_test
def test_call_model_two_input_one_output_more_inputs_failed():
    servable_content = r"""
import numpy as np
from mindspore_serving.server import register
model = register.declare_model(model_file="tensor_add.mindir", model_format="MindIR", with_batch_dim=False)

def call_model(x1, x2, x3):
    y = model.call(x1, x2, x3)
    return y

@register.register_method(output_names="y")
def predict(x1, x2, x3):
    y = register.add_stage(call_model, x1, x2, x3, outputs_count=1)
    return y
    """
    base = start_serving_server(servable_content)
    # Client
    x1 = np.array([[1.1, 2.2], [3.3, 4.4]], np.float32)
    x2 = np.array([[5.5, 6.6], [7.7, 8.8]], np.float32)
    x3 = np.array([[5.5, 6.6], [7.7, 8.8]], np.float32)
    instances = [{"x1": x1, "x2": x2, "x3": x3}]

    client = create_client("localhost:5500", base.servable_name, "predict")
    result = client.infer(instances)
    print("result", result)
    assert "The inputs count 3 of instance 0 is not equal to the inputs count 2 of the model" in result["error"]

@serving_test
def test_call_model_two_input_one_output_batch_call_more_inputs_failed():
    servable_content = r"""
import numpy as np
from mindspore_serving.server import register
model = register.declare_model(model_file="tensor_add.mindir", model_format="MindIR", with_batch_dim=False)

def call_model(x1, x2, x3):
    y = model.call([[x1, x2, x3]])
    return y[0][0]

@register.register_method(output_names="y")
def predict(x1, x2, x3):
    y = register.add_stage(call_model, x1, x2, x3, outputs_count=1)
    return y
    """
    base = start_serving_server(servable_content)
    # Client
    x1 = np.array([[1.1, 2.2], [3.3, 4.4]], np.float32)
    x2 = np.array([[5.5, 6.6], [7.7, 8.8]], np.float32)
    x3 = np.array([[5.5, 6.6], [7.7, 8.8]], np.float32)
    instances = [{"x1": x1, "x2": x2, "x3": x3}]

    client = create_client("localhost:5500", base.servable_name, "predict")
    result = client.infer(instances)
    print("result", result)
    assert "The inputs count 3 of instance 0 is not equal to the inputs count 2 of the model" in result["error"]

@serving_test
def test_call_model_two_input_one_output_batch_call_more_inputs2_failed():
    servable_content = r"""
import numpy as np
from mindspore_serving.server import register
model = register.declare_model(model_file="tensor_add.mindir", model_format="MindIR", with_batch_dim=False)

def call_model(x1, x2, x3):
    y = model.call([[x1, x2], [x1, x2, x3]])
    return y[0][0]

@register.register_method(output_names="y")
def predict(x1, x2, x3):
    y = register.add_stage(call_model, x1, x2, x3, outputs_count=1)
    return y
    """
    base = start_serving_server(servable_content)
    # Client
    x1 = np.array([[1.1, 2.2], [3.3, 4.4]], np.float32)
    x2 = np.array([[5.5, 6.6], [7.7, 8.8]], np.float32)
    x3 = np.array([[5.5, 6.6], [7.7, 8.8]], np.float32)
    instances = [{"x1": x1, "x2": x2, "x3": x3}]

    client = create_client("localhost:5500", base.servable_name, "predict")
    result = client.infer(instances)
    print("result", result)
    assert "The inputs count 3 of instance 1 is not equal to the inputs count 2 of the model" in result["error"]

@serving_test
def test_call_model_two_input_one_output_less_inputs_failed():
    servable_content = r"""
import numpy as np
from mindspore_serving.server import register
model = register.declare_model(model_file="tensor_add.mindir", model_format="MindIR", with_batch_dim=False)

def call_model(x1, x2, x3):
    y = model.call(x1)
    return y

@register.register_method(output_names="y")
def predict(x1, x2, x3):
    y = register.add_stage(call_model, x1, x2, x3, outputs_count=1)
    return y
    """
    base = start_serving_server(servable_content)
    # Client
    x1 = np.array([[1.1, 2.2], [3.3, 4.4]], np.float32)
    x2 = np.array([[5.5, 6.6], [7.7, 8.8]], np.float32)
    x3 = np.array([[5.5, 6.6], [7.7, 8.8]], np.float32)
    instances = [{"x1": x1, "x2": x2, "x3": x3}]

    client = create_client("localhost:5500", base.servable_name, "predict")
    result = client.infer(instances)
    print("result", result)
    assert "The inputs count 1 of instance 0 is not equal to the inputs count 2 of the model" in result["error"]

@serving_test
def test_call_model_two_input_one_output_batch_call_less_inputs_failed():
    servable_content = r"""
import numpy as np
from mindspore_serving.server import register
model = register.declare_model(model_file="tensor_add.mindir", model_format="MindIR", with_batch_dim=False)

def call_model(x1, x2, x3):
    y = model.call([[x1]])
    return y[0][0]

@register.register_method(output_names="y")
def predict(x1, x2, x3):
    y = register.add_stage(call_model, x1, x2, x3, outputs_count=1)
    return y
    """
    base = start_serving_server(servable_content)
    # Client
    x1 = np.array([[1.1, 2.2], [3.3, 4.4]], np.float32)
    x2 = np.array([[5.5, 6.6], [7.7, 8.8]], np.float32)
    x3 = np.array([[5.5, 6.6], [7.7, 8.8]], np.float32)
    instances = [{"x1": x1, "x2": x2, "x3": x3}]

    client = create_client("localhost:5500", base.servable_name, "predict")
    result = client.infer(instances)
    print("result", result)
    assert "The inputs count 1 of instance 0 is not equal to the inputs count 2 of the model" in result["error"]

@serving_test
def test_call_model_two_input_one_output_batch_call_less_inputs2_failed():
    servable_content = r"""
import numpy as np
from mindspore_serving.server import register
model = register.declare_model(model_file="tensor_add.mindir", model_format="MindIR", with_batch_dim=False)

def call_model(x1, x2, x3):
    y = model.call([[x1], [x1, x2]])
    return y[0][0]

@register.register_method(output_names="y")
def predict(x1, x2, x3):
    y = register.add_stage(call_model, x1, x2, x3, outputs_count=1)
    return y
    """
    base = start_serving_server(servable_content)
    # Client
    x1 = np.array([[1.1, 2.2], [3.3, 4.4]], np.float32)
    x2 = np.array([[5.5, 6.6], [7.7, 8.8]], np.float32)
    x3 = np.array([[5.5, 6.6], [7.7, 8.8]], np.float32)
    instances = [{"x1": x1, "x2": x2, "x3": x3}]

    client = create_client("localhost:5500", base.servable_name, "predict")
    result = client.infer(instances)
    print("result", result)
    assert "The inputs count 1 of instance 0 is not equal to the inputs count 2 of the model" in result["error"]

@serving_test
def test_call_model_two_input_one_output_batch_call_less_inputs3_failed():
    servable_content = r"""
import numpy as np
from mindspore_serving.server import register
model = register.declare_model(model_file="tensor_add.mindir", model_format="MindIR", with_batch_dim=False)

def call_model(x1, x2, x3):
    y = model.call([[x1, x2], [x1]])
    return y[0][0]

@register.register_method(output_names="y")
def predict(x1, x2, x3):
    y = register.add_stage(call_model, x1, x2, x3, outputs_count=1)
    return y
    """
    base = start_serving_server(servable_content)
    # Client
    x1 = np.array([[1.1, 2.2], [3.3, 4.4]], np.float32)
    x2 = np.array([[5.5, 6.6], [7.7, 8.8]], np.float32)
    x3 = np.array([[5.5, 6.6], [7.7, 8.8]], np.float32)
    instances = [{"x1": x1, "x2": x2, "x3": x3}]

    client = create_client("localhost:5500", base.servable_name, "predict")
    result = client.infer(instances)
    print("result", result)
    assert "The inputs count 1 of instance 1 is not equal to the inputs count 2 of the model" in result["error"]

@serving_test
def test_call_model_two_input_one_output_invalid_model_failed():
    servable_content = r"""
import numpy as np
from mindspore_serving.server import register

model = register.declare_model(model_file="tensor_add.mindir", model_format="MindIR", with_batch_dim=False)
model_invalid = register.model.Model("tensor_add_test.mindir")

def call_model(x1, x2):
    y = model_invalid.call(x1, x2)
    return y[0][0]

@register.register_method(output_names="y")
def predict(x1, x2):
    y = register.add_stage(call_model, x1, x2, outputs_count=1)
    return y
    """
    base = start_serving_server(servable_content)
    # Client
    x1 = np.array([[1.1, 2.2], [3.3, 4.4]], np.float32)
    x2 = np.array([[5.5, 6.6], [7.7, 8.8]], np.float32)
    instances = [{"x1": x1, "x2": x2}]

    client = create_client("localhost:5500", base.servable_name, "predict")
    result = client.infer(instances)
    print("result", result)
    assert "Model(tensor_add_test.mindir).call() failed: the model is not declared" in result["error"]

@serving_test
def test_call_model_two_input_one_output_with_stage_model_success():
    servable_content = r"""
import numpy as np
from mindspore_serving.server import register
model = register.declare_model(model_file="tensor_add.mindir", model_format="MindIR", with_batch_dim=False)

def call_model(x1, x2):
    y1 = model.call(x1, x2)
    return y1

@register.register_method(output_names="y")
def predict(x1, x2, x3, x4):
    y1 = register.add_stage(call_model, x1, x2, outputs_count=1)
    y2 = register.add_stage(model, y1, x3, outputs_count=1)
    y3 = register.add_stage(call_model, y2, x4, outputs_count=1)
    return y3
    """
    base = start_serving_server(servable_content)
    # Client
    x1 = np.array([[1.1, 2.2], [3.3, 4.4]], np.float32)
    x2 = np.array([[5.5, 6.6], [7.7, 8.8]], np.float32)
    x3 = np.array([[2.1, 3.2], [4.3, 5.4]], np.float32)
    x4 = np.array([[3.5, 4.6], [5.7, 6.8]], np.float32)
    y = x1 + x2 + x3 + x4
    instances = [{"x1": x1, "x2": x2, "x3": x3, "x4": x4}]

    client = create_client("localhost:5500", base.servable_name, "predict")
    result = client.infer(instances)
    print("result", result)
    assert (result[0]["y"] == y).all()

@serving_test
def test_call_model_two_input_one_output_invalid_subgraph_failed():
    servable_content = r"""
import numpy as np
from mindspore_serving.server import register
model = register.declare_model(model_file="tensor_add.mindir", model_format="MindIR", with_batch_dim=False)

def call_model(x1, x2):
    y = model.call(x1, x2, subgraph=1)
    return y[0][0]

@register.register_method(output_names="y")
def predict(x1, x2):
    y = register.add_stage(call_model, x1, x2, outputs_count=1)
    return y
    """
    base = start_serving_server(servable_content)
    # Client
    x1 = np.array([[1.1, 2.2], [3.3, 4.4]], np.float32)
    x2 = np.array([[5.5, 6.6], [7.7, 8.8]], np.float32)
    instances = [{"x1": x1, "x2": x2}]

    client = create_client("localhost:5500", base.servable_name, "predict")
    result = client.infer(instances)
    print("result", result)
    assert "The model does not have subgraph of index 1, the subgraph count of the model is 1" in result["error"]

@serving_test
def test_call_model_two_input_one_output_two_subgraph_success():
    servable_content = r"""
import numpy as np
from mindspore_serving.server import register
model = register.declare_model(model_file=["tensor_add.mindir", "tensor_sub.mindir"], model_format="MindIR",
                               with_batch_dim=False)

def call_model(x1, x2, x3):
    y = model.call(x1, x2, subgraph=0)  # x1+x2
    y = model.call(y, x3, subgraph=1)   # y-x3
    return y

@register.register_method(output_names="y")
def predict(x1, x2, x3):
    y = register.add_stage(call_model, x1, x2, x3, outputs_count=1)
    return y
    """
    base = start_serving_server(servable_content, model_file=["tensor_add.mindir", "tensor_sub.mindir"])
    # Client
    x1 = np.array([[1.1, 2.2], [3.3, 4.4]], np.float32)
    x2 = np.array([[5.5, 6.6], [7.7, 8.8]], np.float32)
    x3 = np.array([[7.5, 8.6], [9.7, 10.8]], np.float32)
    y = x1 + x2 - x3
    instances = [{"x1": x1, "x2": x2, "x3": x3}]

    client = create_client("localhost:5500", base.servable_name, "predict")
    result = client.infer(instances)
    print("result", result)
    assert (result[0]["y"] == y).all()

@serving_test
def test_call_model_diff_input_output_two_subgraph_success():
    servable_content = r"""
import numpy as np
from mindspore_serving.server import register
model = register.declare_model(model_file=["tensor_add_2_3.mindir", "tensor_sub_3_2.mindir"], model_format="MindIR",
                               with_batch_dim=False)

def call_model(x1, x2, x3, x4, x5):
    y1, y2, y3 = model.call(x1, x2, subgraph=0)  # tensor_add_2_3: 2 input, 3 output
    y4, y5 = model.call(x3, x4, x5, subgraph=1)   # tensor_sub_3_2: 3 input, 2 output
    return y1+y4

@register.register_method(output_names="y")
def predict(x1, x2, x3, x4, x5):
    y = register.add_stage(call_model, x1, x2, x3, x4, x5, outputs_count=1)
    return y
    """
    base = start_serving_server(servable_content, model_file=["tensor_add_2_3.mindir", "tensor_sub_3_2.mindir"])
    # Client
    x1 = np.array([[1.1, 2.2], [3.3, 4.4]], np.float32)
    x2 = np.array([[5.5, 6.6], [7.7, 8.8]], np.float32)
    x3 = np.array([[7.5, 8.6], [9.7, 10.8]], np.float32)
    x4 = np.array([[8.5, 10.6], [6.7, 12.8]], np.float32)
    x5 = np.array([[9.5, 11.6], [8.7, 13.8]], np.float32)
    y = (x1 + x2) + (x3 - x4 - x5)
    instances = [{"x1": x1, "x2": x2, "x3": x3, "x4": x4, "x5": x5}]

    client = create_client("localhost:5500", base.servable_name, "predict")
    result = client.infer(instances)
    print("result", result)
    assert (result[0]["y"] == y).all()

@serving_test
def test_call_model_diff_input_output_two_subgraph2_success():
    servable_content = r"""
import numpy as np
from mindspore_serving.server import register
model = register.declare_model(model_file=["tensor_add_3_2.mindir", "tensor_sub_2_3.mindir"], model_format="MindIR",
                               with_batch_dim=False)

def call_model(x1, x2, x3, x4, x5):
    y1, y2 = model.call(x1, x2, x3, subgraph=0)  # tensor_add_3_2: 3 input, 2 output
    y3, y4, y5 = model.call(x4, x5, subgraph=1)   # tensor_sub_2_3: 2 input, 3 output
    return y1+y3

@register.register_method(output_names="y")
def predict(x1, x2, x3, x4, x5):
    y = register.add_stage(call_model, x1, x2, x3, x4, x5, outputs_count=1)
    return y
    """
    base = start_serving_server(servable_content, model_file=["tensor_add_3_2.mindir", "tensor_sub_2_3.mindir"])
    # Client
    x1 = np.array([[1.1, 2.2], [3.3, 4.4]], np.float32)
    x2 = np.array([[5.5, 6.6], [7.7, 8.8]], np.float32)
    x3 = np.array([[7.5, 8.6], [9.7, 10.8]], np.float32)
    x4 = np.array([[8.5, 10.6], [6.7, 12.8]], np.float32)
    x5 = np.array([[9.5, 11.6], [8.7, 13.8]], np.float32)
    y = (x1 + x2 + x3) + (x4 - x5)
    instances = [{"x1": x1, "x2": x2, "x3": x3, "x4": x4, "x5": x5}]

    client = create_client("localhost:5500", base.servable_name, "predict")
    result = client.infer(instances)
    print("result", result)
    assert (result[0]["y"] == y).all()

@serving_test
def test_call_model_diff_input_output_two_subgraph_inputs_count_not_match_failed():
    servable_content = r"""
import numpy as np
from mindspore_serving.server import register
model = register.declare_model(model_file=["tensor_add_3_2.mindir", "tensor_sub_2_3.mindir"], model_format="MindIR",
                               with_batch_dim=False)

def call_model(x1, x2, x3, x4, x5):
    y1, y2 = model.call(x1, x2, x3, subgraph=0)  # tensor_add_3_2: 3 input, 2 output
    y3, y4, y5 = model.call(x4, x5, x3, subgraph=1)   # tensor_sub_2_3: 2 input, 3 output
    return y1+y3

@register.register_method(output_names="y")
def predict(x1, x2, x3, x4, x5):
    y = register.add_stage(call_model, x1, x2, x3, x4, x5, outputs_count=1)
    return y
    """
    base = start_serving_server(servable_content, model_file=["tensor_add_3_2.mindir", "tensor_sub_2_3.mindir"])
    # Client
    x1 = np.array([[1.1, 2.2], [3.3, 4.4]], np.float32)
    x2 = np.array([[5.5, 6.6], [7.7, 8.8]], np.float32)
    x3 = np.array([[7.5, 8.6], [9.7, 10.8]], np.float32)
    x4 = np.array([[8.5, 10.6], [6.7, 12.8]], np.float32)
    x5 = np.array([[9.5, 11.6], [8.7, 13.8]], np.float32)
    instances = [{"x1": x1, "x2": x2, "x3": x3, "x4": x4, "x5": x5}]

    client = create_client("localhost:5500", base.servable_name, "predict")
    result = client.infer(instances)
    print("result", result)
    assert "The inputs count 3 of instance 0 is not equal to the inputs count 2 of the model" in result["error"]

@serving_test
def test_call_model_two_input_one_output_two_model_success():
    servable_content = r"""
import numpy as np
from mindspore_serving.server import register
tensor_add = register.declare_model(model_file="tensor_add.mindir", model_format="MindIR", with_batch_dim=False)
tensor_sub = register.declare_model(model_file="tensor_sub.mindir", model_format="MindIR", with_batch_dim=False)
def call_model(x1, x2, x3):
    y = tensor_add.call(x1, x2)  # x1+x2
    y = tensor_sub.call(y, x3)   # y-x3
    return y

@register.register_method(output_names="y")
def predict(x1, x2, x3):
    y = register.add_stage(call_model, x1, x2, x3, outputs_count=1)
    return y
    """
    base = start_serving_server(servable_content, model_file=["tensor_add.mindir", "tensor_sub.mindir"])
    # Client
    x1 = np.array([[1.1, 2.2], [3.3, 4.4]], np.float32)
    x2 = np.array([[5.5, 6.6], [7.7, 8.8]], np.float32)
    x3 = np.array([[7.5, 8.6], [9.7, 10.8]], np.float32)
    y = x1 + x2 - x3
    instances = [{"x1": x1, "x2": x2, "x3": x3}]

    client = create_client("localhost:5500", base.servable_name, "predict")
    result = client.infer(instances)
    print("result", result)
    assert (result[0]["y"] == y).all()

@serving_test
def test_call_model_diff_input_output_two_model_success():
    servable_content = r"""
import numpy as np
from mindspore_serving.server import register
tensor_add = register.declare_model(model_file="tensor_add_2_3.mindir", model_format="MindIR", with_batch_dim=False)
tensor_sub = register.declare_model(model_file="tensor_sub_3_2.mindir", model_format="MindIR", with_batch_dim=False)

def call_model(x1, x2, x3, x4, x5):
    y1, y2, y3 = tensor_add.call(x1, x2)  # tensor_add_2_3: 2 input, 3 output
    y4, y5 = tensor_sub.call(x3, x4, x5)   # tensor_sub_3_2: 3 input, 2 output
    return y1+y4

@register.register_method(output_names="y")
def predict(x1, x2, x3, x4, x5):
    y = register.add_stage(call_model, x1, x2, x3, x4, x5, outputs_count=1)
    return y
    """
    base = start_serving_server(servable_content, model_file=["tensor_add_2_3.mindir", "tensor_sub_3_2.mindir"])
    # Client
    x1 = np.array([[1.1, 2.2], [3.3, 4.4]], np.float32)
    x2 = np.array([[5.5, 6.6], [7.7, 8.8]], np.float32)
    x3 = np.array([[7.5, 8.6], [9.7, 10.8]], np.float32)
    x4 = np.array([[8.5, 10.6], [6.7, 12.8]], np.float32)
    x5 = np.array([[9.5, 11.6], [8.7, 13.8]], np.float32)
    y = (x1 + x2) + (x3 - x4 - x5)
    instances = [{"x1": x1, "x2": x2, "x3": x3, "x4": x4, "x5": x5}]

    client = create_client("localhost:5500", base.servable_name, "predict")
    result = client.infer(instances)
    print("result", result)
    assert (result[0]["y"] == y).all()

@serving_test
def test_call_model_diff_input_output_two_model2_success():
    servable_content = r"""
import numpy as np
from mindspore_serving.server import register
tensor_add = register.declare_model(model_file="tensor_add_3_2.mindir", model_format="MindIR", with_batch_dim=False)
tensor_sub = register.declare_model(model_file="tensor_sub_2_3.mindir", model_format="MindIR", with_batch_dim=False)

def call_model(x1, x2, x3, x4, x5):
    y1, y2 = tensor_add.call(x1, x2, x3)  # tensor_add_3_2: 3 input, 2 output
    y3, y4, y5 = tensor_sub.call(x4, x5)   # tensor_sub_2_3: 2 input, 3 output
    return y1+y3

@register.register_method(output_names="y")
def predict(x1, x2, x3, x4, x5):
    y = register.add_stage(call_model, x1, x2, x3, x4, x5, outputs_count=1)
    return y
    """
    base = start_serving_server(servable_content, model_file=["tensor_add_3_2.mindir", "tensor_sub_2_3.mindir"])
    # Client
    x1 = np.array([[1.1, 2.2], [3.3, 4.4]], np.float32)
    x2 = np.array([[5.5, 6.6], [7.7, 8.8]], np.float32)
    x3 = np.array([[7.5, 8.6], [9.7, 10.8]], np.float32)
    x4 = np.array([[8.5, 10.6], [6.7, 12.8]], np.float32)
    x5 = np.array([[9.5, 11.6], [8.7, 13.8]], np.float32)
    y = (x1 + x2 + x3) + (x4 - x5)
    instances = [{"x1": x1, "x2": x2, "x3": x3, "x4": x4, "x5": x5}]

    client = create_client("localhost:5500", base.servable_name, "predict")
    result = client.infer(instances)
    print("result", result)
    assert (result[0]["y"] == y).all()

@serving_test
def test_call_model_diff_input_output_two_model_inputs_count_not_match_failed():
    servable_content = r"""
import numpy as np
from mindspore_serving.server import register
tensor_add = register.declare_model(model_file="tensor_add_3_2.mindir", model_format="MindIR", with_batch_dim=False)
tensor_sub = register.declare_model(model_file="tensor_sub_2_3.mindir", model_format="MindIR", with_batch_dim=False)

def call_model(x1, x2, x3, x4, x5):
    y1, y2 = tensor_add.call(x1, x2)  # tensor_add_3_2: 3 input, 2 output
    y3, y4, y5 = tensor_sub.call(x4, x5)   # tensor_sub_2_3: 2 input, 3 output
    return y1+y3

@register.register_method(output_names="y")
def predict(x1, x2, x3, x4, x5):
    y = register.add_stage(call_model, x1, x2, x3, x4, x5, outputs_count=1)
    return y
    """
    base = start_serving_server(servable_content, model_file=["tensor_add_3_2.mindir", "tensor_sub_2_3.mindir"])
    # Client
    x1 = np.array([[1.1, 2.2], [3.3, 4.4]], np.float32)
    x2 = np.array([[5.5, 6.6], [7.7, 8.8]], np.float32)
    x3 = np.array([[7.5, 8.6], [9.7, 10.8]], np.float32)
    x4 = np.array([[8.5, 10.6], [6.7, 12.8]], np.float32)
    x5 = np.array([[9.5, 11.6], [8.7, 13.8]], np.float32)
    instances = [{"x1": x1, "x2": x2, "x3": x3, "x4": x4, "x5": x5}]

    client = create_client("localhost:5500", base.servable_name, "predict")
    result = client.infer(instances)
    print("result", result)
    assert "The inputs count 2 of instance 0 is not equal to the inputs count 3 of the model" in result["error"]

@serving_test
def test_call_model_diff_input_output_two_model_with_bach_dim_success():
    servable_content = r"""
import numpy as np
from mindspore_serving.server import register
tensor_add = register.declare_model(model_file="tensor_add_3_2.mindir", model_format="MindIR", with_batch_dim=True)
tensor_sub = register.declare_model(model_file="tensor_sub_2_3.mindir", model_format="MindIR", with_batch_dim=True)

def call_model(x1, x2, x3, x4, x5):
    y1, y2 = tensor_add.call(x1, x2, x3)  # tensor_add_3_2: 3 input, 2 output
    y3, y4, y5 = tensor_sub.call(x4, x5)   # tensor_sub_2_3: 2 input, 3 output
    return y1+y3

@register.register_method(output_names="y")
def predict(x1, x2, x3, x4, x5):
    y = register.add_stage(call_model, x1, x2, x3, x4, x5, outputs_count=1)
    return y
    """
    base = start_serving_server(servable_content, model_file=["tensor_add_3_2.mindir", "tensor_sub_2_3.mindir"])
    # Client
    x1 = np.array([[3.3, 4.4]], np.float32)
    x2 = np.array([[7.7, 8.8]], np.float32)
    x3 = np.array([[9.7, 10.8]], np.float32)
    x4 = np.array([[6.7, 12.8]], np.float32)
    x5 = np.array([[8.7, 13.8]], np.float32)
    y = (x1 + x2 + x3) + (x4 - x5)
    instances = [{"x1": x1, "x2": x2, "x3": x3, "x4": x4, "x5": x5}]

    client = create_client("localhost:5500", base.servable_name, "predict")
    result = client.infer(instances)
    print("result", result)
    assert (result[0]["y"] == y).all()

max_stars_repo_name: mindspore-ai/serving

max_issues_repo_name: mindspore-ai/serving

max_forks_repo_name: mindspore-ai/serving


text:
import pkg_resources
import math
import tempfile
import os

import numpy as np

from scipy.io import wavfile
from subprocess import Popen
from typing import *

JAR_FILE_0_5_6 = pkg_resources.resource_filename(__name__, 'BeatRoot.jar')
JAR_FILE_0_5_8 = pkg_resources.resource_filename(__name__,
                                                 'beatroot-0.5.8.jar')

version_map = {
    '0.5.6': JAR_FILE_0_5_6,
    '0.5.8': JAR_FILE_0_5_8,
}

SR = 44100

class BeatrootInput:
    '''
    Returns a filename to an audio file to input beatroot.

    It can be initiated with the filename of a already existing audio file
    or a list of onset times (in seconds). In the latter case, the input is
    transformed into an audio file that can be used by beatroot.

    This temp file is deleted upon exit.
    '''

    def __init__(self, input: Union[str, List[float]]):
        '''
        Returns a filename to an audio file.

        Args:
            input (as str): filename of an audio file
            input (as list of floats): onset times (in seconds)
        '''
        self.temp = None
        self.input = None

        if isinstance(input, str):
            self.input = input
            return

        top = math.ceil(max(input))
        if (top == max(input)):
            top += 1

        signal = np.zeros(SR * top)
        for o in input:
            signal[int(o * SR)] = 1

        h, self.temp = tempfile.mkstemp()
        wavfile.write(self.temp, SR, signal)

    def __enter__(self):
        if self.temp is not None:
            return self.temp
        else:
            return self.input

    def __exit__(self, type, value, traceback):
        if self.temp is not None:
            os.remove(self.temp)

def beatroot(input, onsets=False, output=None, version='0.5.8'):
    '''
    Args:
        input: list of onset times (in seconds) or path to audio file to analyze
        onsets: whether to output onset information
        output: if not None, write output to given path
    '''
    args = ['java', '-jar', version_map[version]]
    if onsets:
        args.append('-O')

    _output = 'temp.out' if output is None else output

    args.extend(['-o', _output])

    with BeatrootInput(input) as audio_file:
        args.append(audio_file)

        p = Popen(args)
        p.communicate()
        if output is None:
            with open(_output, 'r') as f:
                beats = [float(l) for l in f]
                return beats
            os.remove(_output)

ext: py

reason: import numpy,from scipy


text:
# Copyright 2011-2016 Anton Akhmerov, Christoph Groth, and Michael Wimmer and
# Copyright 2017 Bas Nijholt.
#
# This file is part of mumpy. It is subject to the license terms in the file
# LICENSE found in the top-level directory of this distribution. A list of
# mumpy authors can be found in the file AUTHORS.md at the top-level
# directory of this distribution and at https://github.com/basnijholt/mumpy.

"""Interface to the MUMPS sparse solver library"""

__all__ = ['MUMPSContext', 'schur_complement', 'AnalysisStatistics',
           'FactorizationStatistics', 'MUMPSError']

import time
import numpy as np
import scipy.sparse
import warnings
from . import _mumps
from .fortran_helpers import prepare_for_fortran

orderings = {'amd': 0, 'amf': 2, 'scotch': 3, 'pord': 4, 'metis': 5,
             'qamd': 6, 'auto': 7}

ordering_name = ['amd', 'user-defined', 'amf',
                 'scotch', 'pord', 'metis', 'qamd']

def possible_orderings():
    """Return the ordering options that are available in the current
    installation of MUMPS.

    Which ordering options are actually available depends how MUMPs was
    compiled. Note that passing an ordering that is not avaialble in the
    current installation of MUMPS will not fail, instead MUMPS will fall back
    to a supported one.

    Returns
    -------
    orderings : list of strings
       A list of installed orderings that can be used in the `ordering` option
       of MUMPS.
    """

    if not possible_orderings.cached:
        # Try all orderings on a small test matrix, and check which one was
        # actually used.

        possible_orderings.cached = ['auto']
        for ordering in [0, 2, 3, 4, 5, 6]:
            data = np.asfortranarray([1, 1], dtype=np.complex128)
            row = np.asfortranarray([1, 2], dtype=_mumps.int_dtype)
            col = np.asfortranarray([1, 2], dtype=_mumps.int_dtype)

            instance = _mumps.zmumps()
            instance.set_assembled_matrix(2, row, col, data)
            instance.icntl[7] = ordering
            instance.job = 1
            instance.call()

            if instance.infog[7] == ordering:
                possible_orderings.cached.append(ordering_name[ordering])

    return possible_orderings.cached

possible_orderings.cached = None

error_messages = {
    -5 : "Not enough memory during analysis phase",
    -6 : "Matrix is singular in structure",
    -7 : "Not enough memory during analysis phase",
    -10 : "Matrix is numerically singular",
    -11 : "The authors of MUMPS would like to hear about this",
    -12 : "The authors of MUMPS would like to hear about this",
    -13 : "Not enough memory"
}

class MUMPSError(RuntimeError):
    def __init__(self, infog):
        self.error = infog[1]
        if self.error in error_messages:
            msg = "{}. (MUMPS error {})".format(
                error_messages[self.error], self.error)
        else:
            msg = "MUMPS failed with error {}.".format(self.error)

        RuntimeError.__init__(self, msg)

class AnalysisStatistics:
    def __init__(self, inst, time=None):
        self.est_mem_incore = inst.infog[17]
        self.est_mem_ooc = inst.infog[27]
        self.est_nonzeros = (inst.infog[20] if inst.infog[20] > 0 else
                             -inst.infog[20] * 1000000)
        self.est_flops = inst.rinfog[1]
        self.ordering = ordering_name[inst.infog[7]]
        self.time = time

    def __str__(self):
        parts = ["estimated memory for in-core factorization:",
                 str(self.est_mem_incore), "mbytes\n",
                 "estimated memory for out-of-core factorization:",
                 str(self.est_mem_ooc), "mbytes\n",
                 "estimated number of nonzeros in factors:",
                 str(self.est_nonzeros), "\n",
                 "estimated number of flops:", str(self.est_flops), "\n",
                 "ordering used:", self.ordering]
        if hasattr(self, "time"):
            parts.extend(["\n analysis time:", str(self.time), "secs"])
        return " ".join(parts)

class FactorizationStatistics:
    def __init__(self, inst, time=None, include_ordering=False):
        # information about pivoting
        self.offdiag_pivots = inst.infog[12] if inst.sym == 0 else 0
        self.delayed_pivots = inst.infog[13]
        self.tiny_pivots = inst.infog[25]

        # possibly include ordering (used in schur_complement)
        if include_ordering:
            self.ordering = ordering_name[inst.infog[7]]

        # information about runtime effiency
        self.memory = inst.infog[22]
        self.nonzeros = (inst.infog[29] if inst.infog[29] > 0 else
                         -inst.infog[29] * 1000000)
        self.flops = inst.rinfog[3]
        if time:
            self.time = time

    def __str__(self):
        parts = ["off-diagonal pivots:", str(self.offdiag_pivots), "\n",
                 "delayed pivots:", str(self.delayed_pivots), "\n",
                 "tiny pivots:", str(self.tiny_pivots), "\n"]
        if hasattr(self, "ordering"):
            parts.extend(["ordering used:", self.ordering, "\n"])
        parts.extend(["memory used during factorization:", str(self.memory),
                      "mbytes\n",
                      "nonzeros in factored matrix:", str(self.nonzeros), "\n",
                      "floating point operations:", str(self.flops)])
        if hasattr(self, "time"):
            parts.extend(["\n factorization time:", str(self.time), "secs"])
        return " ".join(parts)

class MUMPSContext:
    """MUMPSContext contains the internal data structures needed by the
    MUMPS library and contains a user-friendly interface.

    WARNING: Only complex numbers supported.

    Examples
    --------

    Solving a small system of equations.

    >>> import scipy.sparse as sp
    >>> a = sp.coo_matrix([[1.,0],[0,2.]], dtype=complex)
    >>> ctx = mumpy.mumps.MUMPSContext()
    >>> ctx.factor(a)
    >>> ctx.solve([1., 1.])
    array([ 1.0+0.j,  0.5+0.j])

    Instance variables
    ------------------

    analysis_stats : `AnalysisStatistics`
        contains MUMPS statistics after an analysis step (i.e.  after a call to
        `analyze` or `factor`)
    factor_stats : `FactorizationStatistics`
        contains MUMPS statistics after a factorization step (i.e.  after a
        call to `factor`)

    """

    def __init__(self, verbose=False):
        """Init the MUMPSContext class

        Parameters
        ----------

        verbose : True or False
            control whether MUMPS prints lots of internal statistics
            and debug information to screen.
        """
        self.mumps_instance = None
        self.dtype = None
        self.verbose = verbose
        self.factored = False

    def analyze(self, a, ordering='auto', overwrite_a=False):
        """Perform analysis step of MUMPS.

        In the analyis step, MUMPS figures out a reordering for the matrix and
        estimates number of operations and memory needed for the factorization
        time. This step usually needs not be called separately (it is done
        automatically by `factor`), but it can be useful to test which ordering
        would give best performance in the actual factorization, as MUMPS
        estimates are available in `analysis_stats`.

        Parameters
        ----------

        a : sparse SciPy matrix
            input matrix. Internally, the matrix is converted to `coo` format
            (so passing this format is best for performance)
        ordering : { 'auto', 'amd', 'amf', 'scotch', 'pord', 'metis', 'qamd' }
            ordering to use in the factorization. The availability of a
            particular ordering depends on the MUMPS installation.  Default is
            'auto'.
        overwrite_a : True or False
            whether the data in a may be overwritten, which can lead to a small
            performance gain. Default is False.
        """

        a = a.tocoo()

        if a.ndim != 2 or a.shape[0] != a.shape[1]:
            raise ValueError("Input matrix must be square!")

        if not ordering in orderings.keys():
            raise ValueError("Unknown ordering '"+ordering+"'!")

        dtype, row, col, data = _make_assembled_from_coo(a, overwrite_a)

        if dtype != self.dtype:
            self.mumps_instance = getattr(_mumps, dtype+"mumps")(self.verbose)
            self.dtype = dtype

        self.n = a.shape[0]
        self.row = row
        self.col = col
        self.data = data
        # Note: if I don't store them, they go out of scope and are
        #       deleted. I however need the memory to stay around!

        self.mumps_instance.set_assembled_matrix(a.shape[0], row, col, data)
        self.mumps_instance.icntl[7] = orderings[ordering]
        self.mumps_instance.job = 1
        t1 = time.clock()
        self.mumps_instance.call()
        t2 = time.clock()
        self.factored = False

        if self.mumps_instance.infog[1] < 0:
            raise MUMPSError(self.mumps_instance.infog)

        self.analysis_stats = AnalysisStatistics(self.mumps_instance,
                                                 t2 - t1)

    def factor(self, a, ordering='auto', ooc=False, pivot_tol=0.01,
               reuse_analysis=False, overwrite_a=False):
        """Perform the LU factorization of the matrix.

        This LU factorization can then later be used to solve a linear system
        with `solve`. Statistical data of the factorization is stored in
        `factor_stats`.

        Parameters
        ----------

        a : sparse SciPy matrix
            input matrix. Internally, the matrix is converted to `coo` format
            (so passing this format is best for performance)
        ordering : { 'auto', 'amd', 'amf', 'scotch', 'pord', 'metis', 'qamd' }
            ordering to use in the factorization. The availability of a
            particular ordering depends on the MUMPS installation.  Default is
            'auto'.
        ooc : True or False
            whether to use the out-of-core functionality of MUMPS.
            (out-of-core means that data is written to disk to reduce memory
            usage.) Default is False.
        pivot_tol: number in the range [0, 1]
            pivoting threshold. Pivoting is typically limited in sparse
            solvers, as too much pivoting destroys sparsity. 1.0 means full
            pivoting, whereas 0.0 means no pivoting. Default is 0.01.
        reuse_analysis: True or False
            whether to reuse the analysis done in a previous call to `analyze`
            or `factor`. If the structure of the matrix stays the same, and the
            numerical values do not change much, the previous analysis can be
            reused, saving some time.  WARNING: There is no check whether the
            structure of your matrix is compatible with the previous
            analysis. Also, if the values are not similar enough, there might
            be loss of accuracy, without a warning. Default is False.
        overwrite_a : True or False
            whether the data in a may be overwritten, which can lead to a small
            performance gain. Default is False.
        """
        a = a.tocoo()

        if a.ndim != 2 or a.shape[0] != a.shape[1]:
            raise ValueError("Input matrix must be square!")

        # Analysis phase must be done before factorization
        # Note: previous analysis is reused only if reuse_analysis == True

        if reuse_analysis:
            if self.mumps_instance is None:
                warnings.warn("Missing analysis although reuse_analysis=True. "
                              "New analysis is performed.", RuntimeWarning)
                self.analyze(a, ordering=ordering, overwrite_a=overwrite_a)
            else:
                dtype, row, col, data = _make_assembled_from_coo(a,
                                                                 overwrite_a)
                if self.dtype != dtype:
                    raise ValueError("MUMPSContext dtype and matrix dtype "
                                     "incompatible!")

                self.n = a.shape[0]
                self.row = row
                self.col = col
                self.data = data
                self.mumps_instance.set_assembled_matrix(a.shape[0],
                                                         row, col, data)
        else:
            self.analyze(a, ordering=ordering, overwrite_a=overwrite_a)

        self.mumps_instance.icntl[22] = 1 if ooc else 0
        self.mumps_instance.job = 2
        self.mumps_instance.cntl[1] = pivot_tol

        done = False
        while not done:
            t1 = time.clock()
            self.mumps_instance.call()
            t2 = time.clock()

            # error -8, -9 (not enough allocated memory) is treated
            # specially, by increasing the memory relaxation parameter
            if self.mumps_instance.infog[1] < 0:
                if self.mumps_instance.infog[1] in (-8, -9):
                    # double the additional memory
                    self.mumps_instance.icntl[14] *= 2
                else:
                    raise MUMPSError(self.mumps_instance.infog)
            else:
                done = True

        self.factored = True
        self.factor_stats = FactorizationStatistics(self.mumps_instance,
                                                    t2 - t1)

    def _solve_sparse(self, b):
        b = b.tocsc()
        x = np.empty((b.shape[0], b.shape[1]),
                     order='F', dtype=self.data.dtype)

        dtype, col_ptr, row_ind, data = _make_sparse_rhs_from_csc(
            b, self.data.dtype)

        if b.shape[0] != self.n:
            raise ValueError("Right hand side has wrong size")

        if self.dtype != dtype:
            raise ValueError("Data type of right hand side is not "
                             "compatible with the dtype of the "
                             "linear system")

        self.mumps_instance.set_sparse_rhs(col_ptr, row_ind, data)
        self.mumps_instance.set_dense_rhs(x)
        self.mumps_instance.job = 3
        self.mumps_instance.icntl[20] = 1
        self.mumps_instance.call()

        return x

    def _solve_dense(self, b, overwrite_b=False):
        dtype, b = prepare_for_fortran(overwrite_b, b,
                                       np.zeros(1, dtype=self.data.dtype))[:2]

        if b.shape[0] != self.n:
            raise ValueError("Right hand side has wrong size")

        if self.dtype != dtype:
            raise ValueError("Data type of right hand side is not "
                             "compatible with the dtype of the "
                             "linear system")

        self.mumps_instance.set_dense_rhs(b)
        self.mumps_instance.job = 3
        self.mumps_instance.call()

        return b

    def solve(self, b, overwrite_b=False):
        """Solve a linear system after the LU factorization has previously
        been performed by `factor`.

        Supports both dense and sparse right hand sides.

        Parameters
        ----------

        b : dense (NumPy) matrix or vector or sparse (SciPy) matrix
            the right hand side to solve. Accepts both dense and sparse input;
            if the input is sparse 'csc' format is used internally (so passing
            a 'csc' matrix gives best performance).
        overwrite_b : True or False
            whether the data in b may be overwritten, which can lead to a small
            performance gain. Default is False.

        Returns
        -------

        x : NumPy array
            the solution to the linear system as a dense matrix (a vector is
            returned if b was a vector, otherwise a matrix is returned).
        """

        if not self.factored:
            raise RuntimeError("Factorization must be done before solving!")

        if scipy.sparse.isspmatrix(b):
            return self._solve_sparse(b)
        else:
            return self._solve_dense(b, overwrite_b)

def schur_complement(a, indices, ordering='auto', ooc=False, pivot_tol=0.01,
                     calc_stats=False, overwrite_a=False):
    """Compute the Schur complement block of matrix a using MUMPS.

    Parameters:
    a : sparse matrix
        input matrix. Internally, the matrix is converted to `coo` format (so
        passing this format is best for performance)
    indices : 1d array
        indices (row and column) of the desired Schur complement block.  (The
        Schur complement block is square, so that the indices are both row and
        column indices.)
    ordering : { 'auto', 'amd', 'amf', 'scotch', 'pord', 'metis', 'qamd' }
        ordering to use in the factorization. The availability of a particular
        ordering depends on the MUMPS installation.  Default is 'auto'.
    ooc : True or False
        whether to use the out-of-core functionality of MUMPS.  (out-of-core
        means that data is written to disk to reduce memory usage.) Default is
        False.
    pivot_tol: number in the range [0, 1]
        pivoting threshold. Pivoting is typically limited in sparse solvers, as
        too much pivoting destroys sparsity. 1.0 means full pivoting, whereas
        0.0 means no pivoting. Default is 0.01.
    calc_stats: True or False
        whether to return the analysis and factorization statistics collected
        by MUMPS. Default is False.
    overwrite_a : True or False
        whether the data in a may be overwritten, which can lead to a small
        performance gain. Default is False.

    Returns
    -------

    s : NumPy array
        Schur complement block
    factor_stats: `FactorizationStatistics`
        statistics of the factorization as collected by MUMPS.  Only returned
        if ``calc_stats==True``.
    """

    if not scipy.sparse.isspmatrix(a):
        raise ValueError("a must be a sparse SciPy matrix!")

    a = a.tocoo()

    if a.ndim != 2 or a.shape[0] != a.shape[1]:
        raise ValueError("Input matrix must be square!")

    indices = np.asanyarray(indices)

    if indices.ndim != 1:
        raise ValueError("Schur indices must be specified in a 1d array!")

    if not ordering in orderings.keys():
        raise ValueError("Unknown ordering '"+ordering+"'!")

    dtype, row, col, data = _make_assembled_from_coo(a, overwrite_a)
    indices = _make_mumps_index_array(indices)

    mumps_instance = getattr(_mumps, dtype+"mumps")()

    mumps_instance.set_assembled_matrix(a.shape[0], row, col, data)
    mumps_instance.icntl[7] = orderings[ordering]
    mumps_instance.icntl[19] = 1
    mumps_instance.icntl[31] = 1  # discard factors, from 4.10.0
                                  # has no effect in earlier versions

    schur_compl = np.empty((indices.size, indices.size),
                           order='C', dtype=data.dtype)
    mumps_instance.set_schur(schur_compl, indices)

    mumps_instance.job = 4   # job=4 -> 1 and 2 after each other
    t1 = time.clock()
    mumps_instance.call()
    t2 = time.clock()

    if not calc_stats:
        return schur_compl
    else:
        return (schur_compl, FactorizationStatistics(
            mumps_instance, time=t2 - t1, include_ordering=True))

# Some internal helper functions
def _make_assembled_from_coo(a, overwrite_a):
    dtype, data = prepare_for_fortran(overwrite_a, a.data)

    row = np.asfortranarray(a.row.astype(_mumps.int_dtype))
    col = np.asfortranarray(a.col.astype(_mumps.int_dtype))

    # MUMPS uses Fortran indices.
    row += 1
    col += 1

    return dtype, row, col, data

def _make_sparse_rhs_from_csc(b, dtype):
    dtype, data = prepare_for_fortran(True, b.data,
                                      np.zeros(1, dtype=dtype))[:2]

    col_ptr = np.asfortranarray(b.indptr.astype(_mumps.int_dtype))
    row_ind = np.asfortranarray(b.indices.astype(_mumps.int_dtype))

    # MUMPS uses Fortran indices.
    col_ptr += 1
    row_ind += 1

    return dtype, col_ptr, row_ind, data

def _make_mumps_index_array(a):
    a = np.asfortranarray(a.astype(_mumps.int_dtype))
    a += 1                      # Fortran indices

    return a

max_stars_repo_name: basnijholt/mumpy

max_issues_repo_name: basnijholt/mumpy

max_forks_repo_name: basnijholt/mumpy

reason: import numpy,import scipy


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

from scipy.io import arff
from abc import ABC, abstractmethod
from torch.utils.data import DataLoader, TensorDataset

class BaseADDataset(ABC):
    """Anomaly detection dataset base class."""

    def __init__(self, root: str):
        super().__init__()
        self.root = root  # root path to data

        self.n_classes = 2  # 0: normal, 1: outlier
        self.normal_classes = None  # tuple with original class labels that define the normal class
        self.outlier_classes = None  # tuple with original class labels that define the outlier class

        self.train_set = None  # must be of type torch.utils.data.Dataset
        self.test_set = None  # must be of type torch.utils.data.Dataset

    @abstractmethod
    def loaders(self, batch_size: int, shuffle_train=True, shuffle_test=False, num_workers: int = 0) -> (
            DataLoader, DataLoader):
        """Implement data loaders of type torch.utils.data.DataLoader for train_set and test_set."""
        pass

    def __repr__(self):
        return self.__class__.__name__

class TorchvisionDataset(BaseADDataset):
    """TorchvisionDataset class for datasets already implemented in torchvision.datasets."""

    def __init__(self, root: str):
        super().__init__(root)

    def loaders(self, batch_size: int, shuffle_train=True, shuffle_test=False, num_workers: int = 0) -> (
            DataLoader, DataLoader):
        train_loader = DataLoader(dataset=self.train_set, batch_size=batch_size, shuffle=shuffle_train,
                                  num_workers=num_workers)
        test_loader = DataLoader(dataset=self.test_set, batch_size=batch_size, shuffle=shuffle_test,
                                 num_workers=num_workers)
        return train_loader, test_loader

class SAD_Dataset(TorchvisionDataset):

    def __init__(self, root: str, normal_class):

        super().__init__(root)
        self.n_classes = 2
        self.normal_class = normal_class

        # train set
        #load data file path
        url1_train = 'data/sad/SpokenArabicDigitsDimension1_TRAIN.arff'
        url2_train = 'data/sad/SpokenArabicDigitsDimension2_TRAIN.arff'
        url3_train = 'data/sad/SpokenArabicDigitsDimension3_TRAIN.arff'
        url4_train = 'data/sad/SpokenArabicDigitsDimension4_TRAIN.arff'
        url5_train = 'data/sad/SpokenArabicDigitsDimension5_TRAIN.arff'
        url6_train = 'data/sad/SpokenArabicDigitsDimension6_TRAIN.arff'
        url7_train = 'data/sad/SpokenArabicDigitsDimension7_TRAIN.arff'
        url8_train = 'data/sad/SpokenArabicDigitsDimension8_TRAIN.arff'
        url9_train = 'data/sad/SpokenArabicDigitsDimension9_TRAIN.arff'
        url10_train = 'data/sad/SpokenArabicDigitsDimension10_TRAIN.arff'
        url11_train = 'data/sad/SpokenArabicDigitsDimension11_TRAIN.arff'
        url12_train = 'data/sad/SpokenArabicDigitsDimension12_TRAIN.arff'
        url13_train = 'data/sad/SpokenArabicDigitsDimension13_TRAIN.arff'

        # get x and y as dataframe
        x_dim1_train, target_train = get_data(url1_train)
        x_dim2_train, __ = get_data(url2_train)
        x_dim3_train, __ = get_data(url3_train)
        x_dim4_train, __ = get_data(url4_train)
        x_dim5_train, __ = get_data(url5_train)
        x_dim6_train, __ = get_data(url6_train)
        x_dim7_train, __ = get_data(url7_train)
        x_dim8_train, __ = get_data(url8_train)
        x_dim9_train, __ = get_data(url9_train)
        x_dim10_train, __ = get_data(url10_train)
        x_dim11_train, __ = get_data(url11_train)
        x_dim12_train, __ = get_data(url12_train)
        x_dim13_train, __ = get_data(url13_train)

        x_dim1_train = get_features(x_dim1_train)
        x_dim2_train = get_features(x_dim2_train)
        x_dim3_train = get_features(x_dim3_train)
        x_dim4_train = get_features(x_dim4_train)
        x_dim5_train = get_features(x_dim5_train)
        x_dim6_train = get_features(x_dim6_train)
        x_dim7_train = get_features(x_dim7_train)
        x_dim8_train = get_features(x_dim8_train)
        x_dim9_train = get_features(x_dim9_train)
        x_dim10_train = get_features(x_dim10_train)
        x_dim11_train = get_features(x_dim11_train)
        x_dim12_train = get_features(x_dim12_train)
        x_dim13_train = get_features(x_dim13_train)

        # combine 13 dimensions of x
        x_train = np.dstack([x_dim1_train, x_dim2_train, x_dim3_train, x_dim4_train, x_dim5_train, x_dim6_train, x_dim7_train, x_dim8_train, x_dim9_train, x_dim10_train, x_dim11_train, x_dim12_train, x_dim13_train])
        # process output y and produce index
        y_train, index_train = get_target(target_train, normal_class)

        # train only on normal data, extracting normal data
        x_final_train, y_final_train, index_final_train = get_training_set(x_train, y_train, index_train)

        # print("size: ", x_final_train.shape)
        train_set = TensorDataset(torch.Tensor(x_final_train), torch.Tensor(y_final_train), torch.Tensor(index_final_train))
        self.train_set = train_set

        # set up testing set
        url1_test = 'data/sad/SpokenArabicDigitsDimension1_TEST.arff'
        url2_test = 'data/sad/SpokenArabicDigitsDimension2_TEST.arff'
        url3_test = 'data/sad/SpokenArabicDigitsDimension3_TEST.arff'
        url4_test = 'data/sad/SpokenArabicDigitsDimension4_TEST.arff'
        url5_test = 'data/sad/SpokenArabicDigitsDimension5_TEST.arff'
        url6_test = 'data/sad/SpokenArabicDigitsDimension6_TEST.arff'
        url7_test = 'data/sad/SpokenArabicDigitsDimension7_TEST.arff'
        url8_test = 'data/sad/SpokenArabicDigitsDimension8_TEST.arff'
        url9_test = 'data/sad/SpokenArabicDigitsDimension9_TEST.arff'
        url10_test = 'data/sad/SpokenArabicDigitsDimension10_TEST.arff'
        url11_test = 'data/sad/SpokenArabicDigitsDimension11_TEST.arff'
        url12_test = 'data/sad/SpokenArabicDigitsDimension12_TEST.arff'
        url13_test = 'data/sad/SpokenArabicDigitsDimension13_TEST.arff'

        x_dim1_test, target_test = get_data(url1_test)
        x_dim2_test, __ = get_data(url2_test)
        x_dim3_test, __ = get_data(url3_test)
        x_dim4_test, __ = get_data(url4_test)
        x_dim5_test, __ = get_data(url5_test)
        x_dim6_test, __ = get_data(url6_test)
        x_dim7_test, __ = get_data(url7_test)
        x_dim8_test, __ = get_data(url8_test)
        x_dim9_test, __ = get_data(url9_test)
        x_dim10_test, __ = get_data(url10_test)
        x_dim11_test, __ = get_data(url11_test)
        x_dim12_test, __ = get_data(url12_test)
        x_dim13_test, __ = get_data(url13_test)

        x_dim1_test = get_features(x_dim1_test)
        x_dim2_test = get_features(x_dim2_test)
        x_dim3_test = get_features(x_dim3_test)
        x_dim4_test = get_features(x_dim4_test)
        x_dim5_test = get_features(x_dim5_test)
        x_dim6_test = get_features(x_dim6_test)
        x_dim7_test = get_features(x_dim7_test)
        x_dim8_test = get_features(x_dim8_test)
        x_dim9_test = get_features(x_dim9_test)
        x_dim10_test = get_features(x_dim10_test)
        x_dim11_test = get_features(x_dim11_test)
        x_dim12_test = get_features(x_dim12_test)
        x_dim13_test = get_features(x_dim13_test)

        x_final_test = np.dstack([x_dim1_test, x_dim2_test, x_dim3_test, x_dim4_test, x_dim5_test, x_dim6_test, x_dim7_test, x_dim8_test, x_dim9_test, x_dim10_test, x_dim11_test, x_dim12_test, x_dim13_test])
        y_final_test, index_test = get_target(target_test, normal_class)

        test_set = TensorDataset(torch.Tensor(x_final_test), torch.Tensor(y_final_test), torch.Tensor(index_test))
        self.test_set = test_set

def get_data(url):
    """
    input: path to arff data file
    This function loads the arff file, then converts into dataframe.
    The dataframe is then split into x and y.
    output: x is dataframe object without the last column. y is series.
    """
    loaded = arff.loadarff(url)
    df = pd.DataFrame(loaded[0])

    # dropping the last column of dataframe
    # it is still a dataframe object
    x = df.iloc[:, :-1].to_numpy()

    # getting last column as series, not dataframe object
    # as dataframe object is using iloc[:, -1:]
    y = df.iloc[:, -1]

    return x, y

def get_features(x):
    """
    input: unprocessed features data
    This function replaces missing values with zeroes.
    output: processed features data
    """
    for i in range(0, len(x)):
        for j in range(0, 93):
            if pd.isna(x[i][j]):
                x[i][j] = 0

    return x

def get_target(y, normal_class):
    """
    input: pandas series. last column of dataframe.
    This function converts the byte string of series and compare to each classification group
    Each class is represented as a number.
    output: returns numpy array of numbers and index array
    """
    y_new = []
    y_temp = []
    idx = []
    length = len(y)

    for i in range(0, length):
        if y[i].decode('UTF-8') == '1':
            y_temp.append(0)
        elif y[i].decode('UTF-8') == '2':
            y_temp.append(1)
        elif y[i].decode('UTF-8') == '3':
            y_temp.append(2)
        elif y[i].decode('UTF-8') == '4':
            y_temp.append(3)
        elif y[i].decode('UTF-8') == '5':
            y_temp.append(4)
        elif y[i].decode('UTF-8') == '6':
            y_temp.append(5)
        elif y[i].decode('UTF-8') == '7':
            y_temp.append(6)
        elif y[i].decode('UTF-8') == '8':
            y_temp.append(7)
        elif y[i].decode('UTF-8') == '9':
            y_temp.append(8)
        elif y[i].decode('UTF-8') == '10':
            y_temp.append(9)
        idx.append(i)

    for i in range(0, length):
        if y_temp[i] == normal_class:
            y_new.append(1) # normal
        else:
            y_new.append(0) # anomaly

    return np.array(y_new), np.array(idx)

def get_training_set(x, y, idx):
    """
    Input: x, y, index of training set from data file
    This function only collects the normal data from train set.
    The model only trains on normal data of the train set.
    Output: x, y, index of normal data only in train set.
    """
    x_final = []
    y_final = []
    idx_final = []

    for i in range(0, len(x)):
        if y[i] == 1: # normal
            x_final.append(x[i])
            y_final.append(y[i])

    for i in range(0, len(x_final)):
        idx_final.append(i)

    return np.array(x_final), np.array(y_final), np.array(idx_final)

max_stars_repo_name: sodabeans/EdgeML

max_issues_repo_name: sodabeans/EdgeML

max_forks_repo_name: sodabeans/EdgeML