import torch
import tensorrt as trt
from torch.fx.experimental.fx2trt.fx2trt import tensorrt_converter

from .helper_functions import mark_as_int8_layer, extend_attr_to_tuple

def common_maxpool(network, mod, dimension, input_val, layer_name):
    kernel_size = extend_attr_to_tuple(mod, "kernel_size", dimension)
    stride = extend_attr_to_tuple(mod, "stride", dimension)

            

Reported by Pylint.

              import tensorrt as trt
from torch.fx.experimental.fx2trt.fx2trt import tensorrt_converter

from .helper_functions import mark_as_int8_layer, extend_attr_to_tuple

def common_maxpool(network, mod, dimension, input_val, layer_name):
    kernel_size = extend_attr_to_tuple(mod, "kernel_size", dimension)
    stride = extend_attr_to_tuple(mod, "stride", dimension)
    padding = extend_attr_to_tuple(mod, "padding", dimension)

            

Reported by Pylint.

              import torch
import tensorrt as trt
from torch.fx.experimental.fx2trt.fx2trt import tensorrt_converter

from .helper_functions import mark_as_int8_layer, extend_attr_to_tuple

def common_maxpool(network, mod, dimension, input_val, layer_name):
    kernel_size = extend_attr_to_tuple(mod, "kernel_size", dimension)
    stride = extend_attr_to_tuple(mod, "stride", dimension)

            

Reported by Pylint.

              
from .helper_functions import mark_as_int8_layer, extend_attr_to_tuple

def common_maxpool(network, mod, dimension, input_val, layer_name):
    kernel_size = extend_attr_to_tuple(mod, "kernel_size", dimension)
    stride = extend_attr_to_tuple(mod, "stride", dimension)
    padding = extend_attr_to_tuple(mod, "padding", dimension)

    layer = network.add_pooling(

            

Reported by Pylint.

              

@tensorrt_converter(torch.nn.modules.pooling.MaxPool2d)
def maxpool2d(network, submod, args, kwargs, layer_name):
    # args/kwargs should have already been normalized to kwargs
    assert len(args) == 0
    input_val = kwargs["input"]

    if not isinstance(input_val, trt.tensorrt.ITensor):

            

Reported by Pylint.

              @tensorrt_converter(torch.nn.modules.pooling.MaxPool2d)
def maxpool2d(network, submod, args, kwargs, layer_name):
    # args/kwargs should have already been normalized to kwargs
    assert len(args) == 0
    input_val = kwargs["input"]

    if not isinstance(input_val, trt.tensorrt.ITensor):
        raise RuntimeError(f"MaxPool2d received input {input_val} that is not part "
                           "of the TensorRT region!")

            

Reported by Bandit.

              import torch
import tensorrt as trt


def extend_attr_to_tuple(mod: torch.nn.Module, name: str, size: int):
    """
    Extend an attribute of `mod` that named `name` to a tuple of `size`.
    """
    val = getattr(mod, name)

            

Reported by Pylint.

                  """
    Get the dynamic range of a tensor based on its scale, zero_point and dtype.
    """
    if dtype == torch.quint8:
        min_val, max_val = 0, 255
    elif dtype == torch.qint8:
        min_val, max_val = -128, 127
    else:
        raise RuntimeError(f"Unsupported quantized dtype {dtype}")

            

Reported by Pylint.

                  """
    if dtype == torch.quint8:
        min_val, max_val = 0, 255
    elif dtype == torch.qint8:
        min_val, max_val = -128, 127
    else:
        raise RuntimeError(f"Unsupported quantized dtype {dtype}")

    return (min_val - zero_point) * scale, (max_val - zero_point) * scale

            

Reported by Pylint.

              import torch
import tensorrt as trt


def extend_attr_to_tuple(mod: torch.nn.Module, name: str, size: int):
    """
    Extend an attribute of `mod` that named `name` to a tuple of `size`.
    """
    val = getattr(mod, name)

            

Reported by Pylint.

                  Set the precision of a layer to int8 as well as the type of its first output.
    Also set the dynamic range of its first output.
    """
    if layer.type not in {trt.LayerType.SHUFFLE, trt.LayerType.CONCATENATION, trt.LayerType.CONSTANT, trt.LayerType.SHAPE}:
        layer.precision = trt.int8

    for i in range(layer.num_outputs):
        output_val = layer.get_output(i)
        output_val.dynamic_range = dynamic_range

            

Reported by Pylint.

                      # output_val.dtype = trt.int8


def get_inputs_from_args_and_kwargs(args, kwargs, input_names):
    inputs = []
    for i, key in enumerate(input_names):
        if key not in kwargs:
            inputs.append(args[i])
        else:

            

Reported by Pylint.

              from libfb.py import testutil

import test_deploy_python_ext

class TestDeployFromPython(testutil.BaseFacebookTestCase):
    def test_deploy_from_python(self):
        self.assertTrue(test_deploy_python_ext.run())

            

Reported by Pylint.

              from libfb.py import testutil

import test_deploy_python_ext

class TestDeployFromPython(testutil.BaseFacebookTestCase):
    def test_deploy_from_python(self):
        self.assertTrue(test_deploy_python_ext.run())

            

Reported by Pylint.

              from libfb.py import testutil

import test_deploy_python_ext

class TestDeployFromPython(testutil.BaseFacebookTestCase):
    def test_deploy_from_python(self):
        self.assertTrue(test_deploy_python_ext.run())

            

Reported by Pylint.

              
import test_deploy_python_ext

class TestDeployFromPython(testutil.BaseFacebookTestCase):
    def test_deploy_from_python(self):
        self.assertTrue(test_deploy_python_ext.run())

            

Reported by Pylint.

              
import test_deploy_python_ext

class TestDeployFromPython(testutil.BaseFacebookTestCase):
    def test_deploy_from_python(self):
        self.assertTrue(test_deploy_python_ext.run())

            

Reported by Pylint.

              import test_deploy_python_ext

class TestDeployFromPython(testutil.BaseFacebookTestCase):
    def test_deploy_from_python(self):
        self.assertTrue(test_deploy_python_ext.run())

            

Reported by Pylint.

              import torch

avg_pool1 = torch.nn.AdaptiveAvgPool2d((1, None))
reveal_type(avg_pool1)  # E: {AdaptiveAvgPool2d}
avg_pool2 = torch.nn.AdaptiveAvgPool2d((None, 1))
reveal_type(avg_pool2)  # E: {AdaptiveAvgPool2d}
max_pool1 = torch.nn.AdaptiveMaxPool2d((1, None))
reveal_type(max_pool1)  # E: {AdaptiveMaxPool2d}
max_pool2 = torch.nn.AdaptiveMaxPool2d((None, 1))

            

Reported by Pylint.

              import torch

avg_pool1 = torch.nn.AdaptiveAvgPool2d((1, None))
reveal_type(avg_pool1)  # E: {AdaptiveAvgPool2d}
avg_pool2 = torch.nn.AdaptiveAvgPool2d((None, 1))
reveal_type(avg_pool2)  # E: {AdaptiveAvgPool2d}
max_pool1 = torch.nn.AdaptiveMaxPool2d((1, None))
reveal_type(max_pool1)  # E: {AdaptiveMaxPool2d}
max_pool2 = torch.nn.AdaptiveMaxPool2d((None, 1))

            

Reported by Pylint.

              avg_pool1 = torch.nn.AdaptiveAvgPool2d((1, None))
reveal_type(avg_pool1)  # E: {AdaptiveAvgPool2d}
avg_pool2 = torch.nn.AdaptiveAvgPool2d((None, 1))
reveal_type(avg_pool2)  # E: {AdaptiveAvgPool2d}
max_pool1 = torch.nn.AdaptiveMaxPool2d((1, None))
reveal_type(max_pool1)  # E: {AdaptiveMaxPool2d}
max_pool2 = torch.nn.AdaptiveMaxPool2d((None, 1))
reveal_type(max_pool2)  # E: {AdaptiveMaxPool2d}

            

Reported by Pylint.

              avg_pool2 = torch.nn.AdaptiveAvgPool2d((None, 1))
reveal_type(avg_pool2)  # E: {AdaptiveAvgPool2d}
max_pool1 = torch.nn.AdaptiveMaxPool2d((1, None))
reveal_type(max_pool1)  # E: {AdaptiveMaxPool2d}
max_pool2 = torch.nn.AdaptiveMaxPool2d((None, 1))
reveal_type(max_pool2)  # E: {AdaptiveMaxPool2d}

            

Reported by Pylint.

              max_pool1 = torch.nn.AdaptiveMaxPool2d((1, None))
reveal_type(max_pool1)  # E: {AdaptiveMaxPool2d}
max_pool2 = torch.nn.AdaptiveMaxPool2d((None, 1))
reveal_type(max_pool2)  # E: {AdaptiveMaxPool2d}

            

Reported by Pylint.

              import torch

avg_pool1 = torch.nn.AdaptiveAvgPool2d((1, None))
reveal_type(avg_pool1)  # E: {AdaptiveAvgPool2d}
avg_pool2 = torch.nn.AdaptiveAvgPool2d((None, 1))
reveal_type(avg_pool2)  # E: {AdaptiveAvgPool2d}
max_pool1 = torch.nn.AdaptiveMaxPool2d((1, None))
reveal_type(max_pool1)  # E: {AdaptiveMaxPool2d}
max_pool2 = torch.nn.AdaptiveMaxPool2d((None, 1))

            

Reported by Pylint.

              from torch.distributions.transformed_distribution import TransformedDistribution


class LogNormal(TransformedDistribution):
    r"""
    Creates a log-normal distribution parameterized by
    :attr:`loc` and :attr:`scale` where::

        X ~ Normal(loc, scale)

            

Reported by Pylint.

              from torch.distributions import constraints
from torch.distributions.transforms import ExpTransform
from torch.distributions.normal import Normal
from torch.distributions.transformed_distribution import TransformedDistribution


class LogNormal(TransformedDistribution):
    r"""
    Creates a log-normal distribution parameterized by

            

Reported by Pylint.

              
    def __init__(self, loc, scale, validate_args=None):
        base_dist = Normal(loc, scale, validate_args=validate_args)
        super(LogNormal, self).__init__(base_dist, ExpTransform(), validate_args=validate_args)

    def expand(self, batch_shape, _instance=None):
        new = self._get_checked_instance(LogNormal, _instance)
        return super(LogNormal, self).expand(batch_shape, _instance=new)


            

Reported by Pylint.

              
    def expand(self, batch_shape, _instance=None):
        new = self._get_checked_instance(LogNormal, _instance)
        return super(LogNormal, self).expand(batch_shape, _instance=new)

    @property
    def loc(self):
        return self.base_dist.loc


            

Reported by Pylint.

                      return super(LogNormal, self).expand(batch_shape, _instance=new)

    @property
    def loc(self):
        return self.base_dist.loc

    @property
    def scale(self):
        return self.base_dist.scale

            

Reported by Pylint.

                      return self.base_dist.loc

    @property
    def scale(self):
        return self.base_dist.scale

    @property
    def mean(self):
        return (self.loc + self.scale.pow(2) / 2).exp()

            

Reported by Pylint.

              import torch
input = []
input.append(torch.tensor([1.0, 2.0, 3.0, 4.0]))
input.append(torch.tensor([[1.0, 2.0, 3.0, 4.0]]))
input.append(torch.tensor([[[1.0, 2.0, 3.0, 4.0]]]))
reveal_type(input[0].shape[0])  # E: int
reveal_type(input[1].shape[1])  # E: int
reveal_type(input[2].shape[2])  # E: int

            

Reported by Pylint.

              input.append(torch.tensor([1.0, 2.0, 3.0, 4.0]))
input.append(torch.tensor([[1.0, 2.0, 3.0, 4.0]]))
input.append(torch.tensor([[[1.0, 2.0, 3.0, 4.0]]]))
reveal_type(input[0].shape[0])  # E: int
reveal_type(input[1].shape[1])  # E: int
reveal_type(input[2].shape[2])  # E: int

            

Reported by Pylint.

              input.append(torch.tensor([[1.0, 2.0, 3.0, 4.0]]))
input.append(torch.tensor([[[1.0, 2.0, 3.0, 4.0]]]))
reveal_type(input[0].shape[0])  # E: int
reveal_type(input[1].shape[1])  # E: int
reveal_type(input[2].shape[2])  # E: int

            

Reported by Pylint.

              input.append(torch.tensor([[[1.0, 2.0, 3.0, 4.0]]]))
reveal_type(input[0].shape[0])  # E: int
reveal_type(input[1].shape[1])  # E: int
reveal_type(input[2].shape[2])  # E: int

            

Reported by Pylint.

              import torch
input = []
input.append(torch.tensor([1.0, 2.0, 3.0, 4.0]))
input.append(torch.tensor([[1.0, 2.0, 3.0, 4.0]]))
input.append(torch.tensor([[[1.0, 2.0, 3.0, 4.0]]]))
reveal_type(input[0].shape[0])  # E: int
reveal_type(input[1].shape[1])  # E: int
reveal_type(input[2].shape[2])  # E: int

            

Reported by Pylint.

              import torch
input = []
input.append(torch.tensor([1.0, 2.0, 3.0, 4.0]))
input.append(torch.tensor([[1.0, 2.0, 3.0, 4.0]]))
input.append(torch.tensor([[[1.0, 2.0, 3.0, 4.0]]]))
reveal_type(input[0].shape[0])  # E: int
reveal_type(input[1].shape[1])  # E: int
reveal_type(input[2].shape[2])  # E: int

            

Reported by Pylint.

                      importers = sys_importer

    def persistent_id(obj):
        # FIXME: the docs say that persistent_id should only return a string
        # but torch store returns tuples. This works only in the binary protocol
        # see
        # https://docs.python.org/2/library/pickle.html#pickling-and-unpickling-external-objects
        # https://github.com/python/cpython/blob/master/Lib/pickle.py#L527-L537
        if torch.is_storage(obj):

            

Reported by Pylint.

                  data_value = data_buf.getvalue()
    return data_value, serialized_storages, serialized_dtypes, importer.zip_reader if importer else None

def _load_storages(id, zip_reader, obj_bytes, serialized_storages):

    def persistent_load(saved_id):
        assert isinstance(saved_id, tuple)
        typename = _maybe_decode_ascii(saved_id[0])
        data = saved_id[1:]

            

Reported by Pylint.

              import io
import torch
from torch.package._package_pickler import create_pickler
from torch.package._package_unpickler import PackageUnpickler
from torch.package import sys_importer, OrderedImporter, PackageImporter, Importer
from torch.serialization import _maybe_decode_ascii

def _save_storages(importer, obj):
    serialized_storages = []

            

Reported by Pylint.

                  pickler.persistent_id = persistent_id
    pickler.dump(obj)
    data_value = data_buf.getvalue()
    return data_value, serialized_storages, serialized_dtypes, importer.zip_reader if importer else None

def _load_storages(id, zip_reader, obj_bytes, serialized_storages):

    def persistent_load(saved_id):
        assert isinstance(saved_id, tuple)

            

Reported by Pylint.

                  data_value = data_buf.getvalue()
    return data_value, serialized_storages, serialized_dtypes, importer.zip_reader if importer else None

def _load_storages(id, zip_reader, obj_bytes, serialized_storages):

    def persistent_load(saved_id):
        assert isinstance(saved_id, tuple)
        typename = _maybe_decode_ascii(saved_id[0])
        data = saved_id[1:]

            

Reported by Pylint.

              def _load_storages(id, zip_reader, obj_bytes, serialized_storages):

    def persistent_load(saved_id):
        assert isinstance(saved_id, tuple)
        typename = _maybe_decode_ascii(saved_id[0])
        data = saved_id[1:]

        if typename == 'storage':
            return serialized_storages[data[0]]

            

Reported by Bandit.

              import torch


t = torch.randn(2, 3)
reveal_type(t)  # E: {Tensor}
u = torch.randn(2, 3)
reveal_type(u)  # E: {Tensor}
t.copy_(u)
reveal_type(t)  # E: {Tensor}

            

Reported by Pylint.

              

t = torch.randn(2, 3)
reveal_type(t)  # E: {Tensor}
u = torch.randn(2, 3)
reveal_type(u)  # E: {Tensor}
t.copy_(u)
reveal_type(t)  # E: {Tensor}
r = (t == u).all()

            

Reported by Pylint.

              t = torch.randn(2, 3)
reveal_type(t)  # E: {Tensor}
u = torch.randn(2, 3)
reveal_type(u)  # E: {Tensor}
t.copy_(u)
reveal_type(t)  # E: {Tensor}
r = (t == u).all()
reveal_type(r)  # E: {Tensor}

            

Reported by Pylint.

              u = torch.randn(2, 3)
reveal_type(u)  # E: {Tensor}
t.copy_(u)
reveal_type(t)  # E: {Tensor}
r = (t == u).all()
reveal_type(r)  # E: {Tensor}

            

Reported by Pylint.

              t.copy_(u)
reveal_type(t)  # E: {Tensor}
r = (t == u).all()
reveal_type(r)  # E: {Tensor}

            

Reported by Pylint.

              import torch


t = torch.randn(2, 3)
reveal_type(t)  # E: {Tensor}
u = torch.randn(2, 3)
reveal_type(u)  # E: {Tensor}
t.copy_(u)
reveal_type(t)  # E: {Tensor}

            

Reported by Pylint.

              import torch
import torchvision
from torch.utils.mobile_optimizer import optimize_for_mobile

model = torchvision.models.mobilenet_v2(pretrained=True)
model.eval()
example = torch.rand(1, 3, 224, 224)
traced_script_module = torch.jit.script(model, example)
optimized_scripted_module = optimize_for_mobile(traced_script_module)

            

Reported by Pylint.

              import torch
import torchvision
from torch.utils.mobile_optimizer import optimize_for_mobile

model = torchvision.models.mobilenet_v2(pretrained=True)
model.eval()
example = torch.rand(1, 3, 224, 224)
traced_script_module = torch.jit.script(model, example)
optimized_scripted_module = optimize_for_mobile(traced_script_module)

            

Reported by Pylint.

              import torch
import torchvision
from torch.utils.mobile_optimizer import optimize_for_mobile

model = torchvision.models.mobilenet_v2(pretrained=True)
model.eval()
example = torch.rand(1, 3, 224, 224)
traced_script_module = torch.jit.script(model, example)
optimized_scripted_module = optimize_for_mobile(traced_script_module)

            

Reported by Pylint.

              example = torch.rand(1, 3, 224, 224)
traced_script_module = torch.jit.script(model, example)
optimized_scripted_module = optimize_for_mobile(traced_script_module)
exported_optimized_scripted_module = optimized_scripted_module._save_for_lite_interpreter("model.ptl")

            

Reported by Pylint.

              import torch
import torchvision
from torch.utils.mobile_optimizer import optimize_for_mobile

model = torchvision.models.mobilenet_v2(pretrained=True)
model.eval()
example = torch.rand(1, 3, 224, 224)
traced_script_module = torch.jit.script(model, example)
optimized_scripted_module = optimize_for_mobile(traced_script_module)

            

Reported by Pylint.

              example = torch.rand(1, 3, 224, 224)
traced_script_module = torch.jit.script(model, example)
optimized_scripted_module = optimize_for_mobile(traced_script_module)
exported_optimized_scripted_module = optimized_scripted_module._save_for_lite_interpreter("model.ptl")

            

Reported by Pylint.

                  if (srcPayload.length) {
      // Empty vector's data() can return nullptr, use the length to avoid
      // coying into nullptr
      memcpy(dstPayload.data, srcPayload.data, srcPayload.length);
    }
  }
  EXPECT_EQ(
      recvingTpAllocation.tensors.size(), sendingTpMessage.tensors.size());
  for (int i = 0; i < recvingTpAllocation.tensors.size(); i++) {

            

Reported by FlawFinder.

                for (int i = 0; i < recvingTpAllocation.tensors.size(); i++) {
    tensorpipe::Message::Tensor& srcTensor = sendingTpMessage.tensors[i];
    tensorpipe::Allocation::Tensor& dstTensor = recvingTpAllocation.tensors[i];
    memcpy(
        dstTensor.buffer.unwrap<tensorpipe::CpuBuffer>().ptr,
        srcTensor.buffer.unwrap<tensorpipe::CpuBuffer>().ptr,
        srcTensor.length);
  }


            

Reported by FlawFinder.

                EXPECT_EQ(mtype, recvingRpcMessage->type());
  EXPECT_EQ(payloadCopy, recvingRpcMessage->payload());
  EXPECT_EQ(mId, recvingRpcMessage->id());
  EXPECT_TRUE(torch::equal(t1, recvingRpcMessage->tensors()[0]));
  EXPECT_TRUE(torch::equal(t2, recvingRpcMessage->tensors()[1]));
}

TEST(TensorpipeSerialize, RecopySparseTensors) {
  // Take a 1K row of a 1M tensors, and make sure we don't send across 1M rows.

            

Reported by FlawFinder.

                EXPECT_EQ(payloadCopy, recvingRpcMessage->payload());
  EXPECT_EQ(mId, recvingRpcMessage->id());
  EXPECT_TRUE(torch::equal(t1, recvingRpcMessage->tensors()[0]));
  EXPECT_TRUE(torch::equal(t2, recvingRpcMessage->tensors()[1]));
}

TEST(TensorpipeSerialize, RecopySparseTensors) {
  // Take a 1K row of a 1M tensors, and make sure we don't send across 1M rows.
  constexpr size_t k1K = 1024;

            

Reported by FlawFinder.

              
  EXPECT_EQ(tpBuffers.tensors.size(), 2);
  EXPECT_EQ(sendingTpMessage.tensors.size(), 2);
  EXPECT_TRUE(torch::equal(main, tpBuffers.tensors[0]));
  EXPECT_TRUE(torch::equal(tiny, tpBuffers.tensors[1]));
  // Test cloned storage
  EXPECT_EQ(
      main.storage().data(),
      sendingTpMessage.tensors[0].buffer.unwrap<tensorpipe::CpuBuffer>().ptr);

            

Reported by FlawFinder.

                EXPECT_EQ(tpBuffers.tensors.size(), 2);
  EXPECT_EQ(sendingTpMessage.tensors.size(), 2);
  EXPECT_TRUE(torch::equal(main, tpBuffers.tensors[0]));
  EXPECT_TRUE(torch::equal(tiny, tpBuffers.tensors[1]));
  // Test cloned storage
  EXPECT_EQ(
      main.storage().data(),
      sendingTpMessage.tensors[0].buffer.unwrap<tensorpipe::CpuBuffer>().ptr);
  EXPECT_NE(

            

Reported by FlawFinder.