import torch
import numpy as np
import torch_npu
import unittest
from torch_npu.testing.testcase import TestCase, run_tests
from torch_npu.testing.common_utils import create_common_tensor, SupportedDevices


class TestTopK(TestCase):
    def cpu_op_exec(self, input1, k):
        output, indices = torch.topk(input1, k)
        output = output.numpy()
        indices = indices.numpy().astype(np.int32)
        return output, indices

    def npu_op_exec(self, input1, k):
        output, indices = torch.topk(input1, k)
        output = output.to("cpu")
        indices = indices.to("cpu")
        output = output.numpy()
        indices = indices.numpy().astype(np.int32)
        return output, indices

    def npu_op_exec_out(self, input1, k, output, indices):
        torch.topk(input1, k, out=(output, indices))
        output = output.to("cpu").numpy()
        indices = indices.to("cpu").numpy().astype(np.int32)
        return output, indices

    def topk_result(self, shape_format):
        for item in shape_format:
            cpu_input, npu_input = create_common_tensor(item, 0, 100)
            if cpu_input.dtype == torch.float16:
                cpu_input = cpu_input.to(torch.float32)
            cpu_output, cpu_indices = self.cpu_op_exec(cpu_input, 5)
            npu_output, npu_indices = self.npu_op_exec(npu_input, 5)
            cpu_output = cpu_output.astype(npu_output.dtype)
            self.assertRtolEqual(cpu_output, npu_output, prec=1.e-1)

    def topk_large_result(self, shape_format):
        for item in shape_format:
            cpu_input, npu_input = create_common_tensor(item, 0, 65504)
            if cpu_input.dtype == torch.float16:
                cpu_input = cpu_input.to(torch.float32)
            k = 4000000
            cpu_output, cpu_indices = self.cpu_op_exec(cpu_input, k)
            npu_output, npu_indices = self.npu_op_exec(npu_input, k)
            cpu_output = cpu_output.astype(npu_output.dtype)

            self.assertRtolEqual(cpu_output, npu_output)

    def test_topk_out_result_fp32(self):
        shape_format = [
            [np.float32, 0, [18]],
            [np.float32, 0, [5, 256]],
            [np.float32, 0, [32, 8, 8]],
            [np.float32, 0, [64, 112, 7, 7]],
        ]
        for item in shape_format:
            cpu_input1, npu_input1 = create_common_tensor(item, 0, 100)
            cpu_input2, npu_input2 = create_common_tensor(item, 0, 100)
            cpu_input3, npu_input3 = create_common_tensor(item, 0, 100)
            cpu_output, cpu_indices = self.cpu_op_exec(cpu_input1, 5)
            npu_output, npu_indices = self.npu_op_exec_out(npu_input1, 5, npu_input2, npu_input3.to(torch.int64))
            cpu_output = cpu_output.astype(npu_output.dtype)
            self.assertRtolEqual(cpu_output, npu_output, prec=1.e-1)

    def test_topk_out_result_fp16(self):
        shape_format = [
            [np.float16, 0, [18]],
            [np.float16, 0, [5, 256]],
            [np.float16, 0, [32, 8, 8]],
            [np.float16, 0, [64, 112, 7, 7]],
        ]
        for item in shape_format:
            cpu_input1, npu_input1 = create_common_tensor(item, 0, 100)
            cpu_input2, npu_input2 = create_common_tensor(item, 0, 100)
            cpu_input3, npu_input3 = create_common_tensor(item, 0, 100)
            if cpu_input1.dtype == torch.float16:
                cpu_input1 = cpu_input1.to(torch.float32)
            cpu_output, cpu_indices = self.cpu_op_exec(cpu_input1, 5)
            npu_output, npu_indices = self.npu_op_exec_out(npu_input1, 5, npu_input2, npu_input3.to(torch.int64))
            cpu_output = cpu_output.astype(npu_output.dtype)
            self.assertRtolEqual(cpu_output, npu_output)

    def test_topk_shape_format_fp16_1d(self):
        format_list = [0, 3, 4, 29]
        shape_format = [
            [np.float16, i, [18]] for i in format_list
        ]
        self.topk_result(shape_format)

    def test_topk_shape_format_fp16_large_1d(self):
        format_list = [-1]
        shape_format = [
            [np.float16, i, [104857600]] for i in format_list
        ]
        self.topk_large_result(shape_format)

    def test_topk_shape_format_fp32_1d(self):
        format_list = [0, 3, 4, 29]
        shape_format = [
            [np.float32, i, [18]] for i in format_list
        ]
        self.topk_result(shape_format)

    def test_topk_shape_format_fp16_2d(self):
        format_list = [0, 3, 4, 29]
        shape_format = [
            [np.float16, i, [5, 256]] for i in format_list
        ]
        self.topk_result(shape_format)

    def test_topk_shape_format_fp32_2d(self):
        format_list = [0, 3, 4, 29]
        shape_format = [
            [np.float32, i, [5, 256]] for i in format_list
        ]
        self.topk_result(shape_format)

    def test_topk_shape_format_fp16_3d(self):
        format_list = [0, 3, 4, 29]
        shape_format = [
            [np.float16, i, [32, 8, 8]] for i in format_list
        ]
        self.topk_result(shape_format)

    def test_topk_shape_format_fp32_3d(self):
        format_list = [0, 3, 4, 29]
        shape_format = [
            [np.float32, i, [32, 8, 8]] for i in format_list
        ]
        self.topk_result(shape_format)

    def test_topk_shape_format_fp16_4d(self):
        format_list = [0, 3, 4, 29]
        shape_format = [
            [np.float16, i, [64, 112, 7, 7]] for i in format_list
        ]
        self.topk_result(shape_format)

    def test_topk_shape_format_fp32_4d(self):
        format_list = [0, 3, 4, 29]
        shape_format = [
            [np.float32, i, [64, 112, 7, 7]] for i in format_list
        ]
        self.topk_result(shape_format)

    @SupportedDevices(['Ascend910B'])
    def test_topk_indices_dtype_zero_dim(self):
        cpu_input = torch.tensor(42.5)
        npu_input = cpu_input.npu()
        k = 0
        values, indices = torch.topk(npu_input, k)
        values_cpu, indices_cpu = torch.topk(cpu_input, k)
        self.assertRtolEqual(values, values_cpu)
        self.assertEqual(indices.cpu(), indices_cpu)

    @unittest.skip("skip test_topk_zero_dim now")
    @SupportedDevices(['Ascend910B'])
    def test_topk_zero_dim(self):
        inputs = torch.tensor(3).npu()
        inputs_cpu = inputs.cpu()
        values, indices = torch.topk(inputs, 0, 0, False, True)
        values_cpu, indices_cpu = torch.topk(inputs_cpu, 0, 0, False, True)
        self.assertRtolEqual(values, values_cpu)
        self.assertRtolEqual(indices, indices_cpu)

        values_cpu, indices_cpu = torch.topk(inputs_cpu, 1, 0, False, True)
        values, indices = torch.topk(inputs, 1, 0, False, True)
        self.assertRtolEqual(values, values_cpu)
        self.assertRtolEqual(indices, indices_cpu)

    @unittest.skip("skip test_topk_zero_dim now")
    @SupportedDevices(['Ascend910B'])
    def test_topk_zero_dim_out(self):
        inputs = torch.tensor(3).npu()
        inputs_cpu = inputs.cpu()
        values = torch.tensor(0).npu()
        indices = torch.tensor(0).npu()
        values_cpu = values.cpu()
        indices_cpu = indices.cpu()
        out = (values, indices)
        out_cpu = (values_cpu, indices_cpu)
        torch.topk(inputs, 0, 0, False, True, out=out)
        torch.topk(inputs_cpu, 0, 0, False, True, out=out_cpu)
        self.assertRtolEqual(values, values_cpu)
        self.assertRtolEqual(indices, indices_cpu)

        values = torch.tensor(0).npu()
        indices = torch.tensor(0).npu()
        values_cpu = values.cpu()
        indices_cpu = indices.cpu()
        out = (values, indices)
        out_cpu = (values_cpu, indices_cpu)
        torch.topk(inputs, 1, 0, False, True, out=out)
        torch.topk(inputs_cpu, 1, 0, False, True, out=out_cpu)
        self.assertRtolEqual(values, values_cpu)
        self.assertRtolEqual(indices, indices_cpu)

        values = torch.tensor(0).npu().to(torch.int32)
        indices = torch.tensor(0).npu()
        out = (values, indices)
        with self.assertRaisesRegex(RuntimeError, "Expected out tensor to have dtype long"):
            torch.topk(inputs, 0, 0, False, True, out=out)


if __name__ == "__main__":
    run_tests()