import torch

# ModuleList with elements of type Module
class FooModule(torch.nn.Module):
    pass

class BarModule(torch.nn.Module):
    pass


            

Reported by Pylint.

              
ml: torch.nn.ModuleList = torch.nn.ModuleList([FooModule(), BarModule()])
ml[0].children() == []  # noqa: B015
reveal_type(ml)  # E: {ModuleList}

            

Reported by Pylint.

                  pass

ml: torch.nn.ModuleList = torch.nn.ModuleList([FooModule(), BarModule()])
ml[0].children() == []  # noqa: B015
reveal_type(ml)  # E: {ModuleList}

            

Reported by Pylint.

              import torch

# ModuleList with elements of type Module
class FooModule(torch.nn.Module):
    pass

class BarModule(torch.nn.Module):
    pass


            

Reported by Pylint.

              import torch

# ModuleList with elements of type Module
class FooModule(torch.nn.Module):
    pass

class BarModule(torch.nn.Module):
    pass


            

Reported by Pylint.

              import torch

# ModuleList with elements of type Module
class FooModule(torch.nn.Module):
    pass

class BarModule(torch.nn.Module):
    pass


            

Reported by Pylint.

              class FooModule(torch.nn.Module):
    pass

class BarModule(torch.nn.Module):
    pass

ml: torch.nn.ModuleList = torch.nn.ModuleList([FooModule(), BarModule()])
ml[0].children() == []  # noqa: B015
reveal_type(ml)  # E: {ModuleList}

            

Reported by Pylint.

              class FooModule(torch.nn.Module):
    pass

class BarModule(torch.nn.Module):
    pass

ml: torch.nn.ModuleList = torch.nn.ModuleList([FooModule(), BarModule()])
ml[0].children() == []  # noqa: B015
reveal_type(ml)  # E: {ModuleList}

            

Reported by Pylint.

                code += std::string("namespace ") + FusionExecutor::kernelNamespace() +
      " {\n" + executor_utils::kernelPreamble() + kernel + "}\n";

  const char* debug_env = std::getenv("PYTORCH_CUDA_FUSER_DEBUG");
  if (debug_env && atoi(debug_env)) {
    std::cout << "\n==== codegen output for kernel: " << kernelName()
              << " ====" << std::endl
              << code << std::endl
              << "======================================\n"

            

Reported by FlawFinder.

                FusionGuard fg(&fusion_);
  options_ = options;

  const char* debug_env = std::getenv("PYTORCH_CUDA_FUSER_DEBUG");
  if (debug_env && atoi(debug_env)) {
    std::cout << "\n==== codegen output for kernel: " << kernelName()
              << " ====" << std::endl
              << code << std::endl
              << "======================================\n"

            

Reported by FlawFinder.

                lowered_ = GpuLower(&fusion_);
  const auto kernel = lowered_.kernel();

  const char* dump_kir_env = std::getenv("PYTORCH_CUDA_FUSER_DUMP_KIR");
  if (dump_kir_env && atoi(dump_kir_env)) {
    kernel->print();
  }

  const auto& kernel_summary = kernel->summary();

            

Reported by FlawFinder.

                lowered_ = GpuLower(&fusion_);
  const auto kernel = lowered_.kernel();

  const char* dump_kir_env = std::getenv("PYTORCH_CUDA_FUSER_DUMP_KIR");
  if (dump_kir_env && atoi(dump_kir_env)) {
    kernel->print();
  }

  const auto kernel_code = codegen::generateCudaKernel(kernel, kernelName());

            

Reported by FlawFinder.

                    " {\n" + executor_utils::kernelPreamble() + kernel + "}\n";

  const char* debug_env = std::getenv("PYTORCH_CUDA_FUSER_DEBUG");
  if (debug_env && atoi(debug_env)) {
    std::cout << "\n==== codegen output for kernel: " << kernelName()
              << " ====" << std::endl
              << code << std::endl
              << "======================================\n"
              << std::endl;

            

Reported by FlawFinder.

                options_ = options;

  const char* debug_env = std::getenv("PYTORCH_CUDA_FUSER_DEBUG");
  if (debug_env && atoi(debug_env)) {
    std::cout << "\n==== codegen output for kernel: " << kernelName()
              << " ====" << std::endl
              << code << std::endl
              << "======================================\n"
              << std::endl;

            

Reported by FlawFinder.

                const auto kernel = lowered_.kernel();

  const char* dump_kir_env = std::getenv("PYTORCH_CUDA_FUSER_DUMP_KIR");
  if (dump_kir_env && atoi(dump_kir_env)) {
    kernel->print();
  }

  const auto& kernel_summary = kernel->summary();
  has_block_reductions = kernel_summary.has_block_reductions;

            

Reported by FlawFinder.

                const auto kernel = lowered_.kernel();

  const char* dump_kir_env = std::getenv("PYTORCH_CUDA_FUSER_DUMP_KIR");
  if (dump_kir_env && atoi(dump_kir_env)) {
    kernel->print();
  }

  const auto kernel_code = codegen::generateCudaKernel(kernel, kernelName());
  const auto structured_code = getStructuredCode(kernel_code);

            

Reported by FlawFinder.

              if is_available():
    # Registers FAULTY_TENSORPIPE RPC backend.
    from . import faulty_agent_backend_registry
    from torch._C._distributed_rpc_testing import (
        FaultyTensorPipeRpcBackendOptions,
        FaultyTensorPipeAgent,
    )

            

Reported by Pylint.

              

def is_available():
    return hasattr(torch._C, "_faulty_agent_init")


if is_available() and not torch._C._faulty_agent_init():
    raise RuntimeError("Failed to initialize torch.distributed.rpc._testing")


            

Reported by Pylint.

                  return hasattr(torch._C, "_faulty_agent_init")


if is_available() and not torch._C._faulty_agent_init():
    raise RuntimeError("Failed to initialize torch.distributed.rpc._testing")

if is_available():
    # Registers FAULTY_TENSORPIPE RPC backend.
    from . import faulty_agent_backend_registry

            

Reported by Pylint.

                  return hasattr(torch._C, "_faulty_agent_init")


if is_available() and not torch._C._faulty_agent_init():
    raise RuntimeError("Failed to initialize torch.distributed.rpc._testing")

if is_available():
    # Registers FAULTY_TENSORPIPE RPC backend.
    from . import faulty_agent_backend_registry

            

Reported by Pylint.

              
if is_available():
    # Registers FAULTY_TENSORPIPE RPC backend.
    from . import faulty_agent_backend_registry
    from torch._C._distributed_rpc_testing import (
        FaultyTensorPipeRpcBackendOptions,
        FaultyTensorPipeAgent,
    )

            

Reported by Pylint.

              
import torch


def is_available():
    return hasattr(torch._C, "_faulty_agent_init")


if is_available() and not torch._C._faulty_agent_init():

            

Reported by Pylint.

              import torch


def is_available():
    return hasattr(torch._C, "_faulty_agent_init")


if is_available() and not torch._C._faulty_agent_init():
    raise RuntimeError("Failed to initialize torch.distributed.rpc._testing")

            

Reported by Pylint.

                  return hasattr(torch._C, "_faulty_agent_init")


if is_available() and not torch._C._faulty_agent_init():
    raise RuntimeError("Failed to initialize torch.distributed.rpc._testing")

if is_available():
    # Registers FAULTY_TENSORPIPE RPC backend.
    from . import faulty_agent_backend_registry

            

Reported by Pylint.

              AbstractStream = Union[torch.cuda.Stream, CPUStreamType]


def new_stream(device: torch.device) -> AbstractStream:
    """Creates a new stream for either CPU or CUDA device."""
    if device.type != "cuda":
        return CPUStream
    return torch.cuda.Stream(device)


            

Reported by Pylint.

                  return torch.cuda.Stream(device)


def current_stream(device: torch.device) -> AbstractStream:
    """:func:`torch.cuda.current_stream` for either CPU or CUDA device."""
    if device.type != "cuda":
        return CPUStream
    return torch.cuda.current_stream(device)


            

Reported by Pylint.

                  return torch.cuda.current_stream(device)


def default_stream(device: torch.device) -> AbstractStream:
    """:func:`torch.cuda.default_stream` for either CPU or CUDA device."""
    if device.type != "cuda":
        return CPUStream
    return torch.cuda.default_stream(device)


            

Reported by Pylint.

              

@contextmanager
def use_device(device: torch.device) -> Generator[None, None, None]:
    """:func:`torch.cuda.device` for either CPU or CUDA device."""
    if device.type != "cuda":
        yield
        return


            

Reported by Pylint.

                      yield


def get_device(stream: AbstractStream) -> torch.device:
    """Gets the device from CPU or CUDA stream."""
    if is_cuda(stream):
        return as_cuda(stream).device
    return torch.device("cpu")


            

Reported by Pylint.

                  """Gets the device from CPU or CUDA stream."""
    if is_cuda(stream):
        return as_cuda(stream).device
    return torch.device("cpu")


def wait_stream(source: AbstractStream, target: AbstractStream) -> None:
    """:meth:`torch.cuda.Stream.wait_stream` for either CPU or CUDA stream. It
    makes the source stream wait until the target stream completes work queued.

            

Reported by Pylint.

              __all__: List[str] = []


class CPUStreamType:
    pass


# The placeholder on place of streams for the CPU device instead of CUDA.
CPUStream = CPUStreamType()

            

Reported by Pylint.

              __all__: List[str] = []


class CPUStreamType:
    pass


# The placeholder on place of streams for the CPU device instead of CUDA.
CPUStream = CPUStreamType()

            

Reported by Pylint.

              TYPE_PATTERN = re.compile(r"(?:const\s+)?([A-Z]\w+)")

# Add 'at::' to types defined in ATen namespace, e.g. Tensor, TensorList, IntArrayRef and etc.
# TODO: maybe update the cpp argument API to take optional namespace argument?
def fully_qualified_type(argument_type: str) -> str:
    def maybe_optional_type(type: str, is_opt: bool) -> str:
        return f'c10::optional<{type}>' if is_opt else type

    opt_match = OPTIONAL_TYPE_PATTERN.match(argument_type)

            

Reported by Pylint.

              # Add 'at::' to types defined in ATen namespace, e.g. Tensor, TensorList, IntArrayRef and etc.
# TODO: maybe update the cpp argument API to take optional namespace argument?
def fully_qualified_type(argument_type: str) -> str:
    def maybe_optional_type(type: str, is_opt: bool) -> str:
        return f'c10::optional<{type}>' if is_opt else type

    opt_match = OPTIONAL_TYPE_PATTERN.match(argument_type)
    is_opt = opt_match is not None
    if opt_match:

            

Reported by Pylint.

              # Generates C++ functions that wrap ATen tensor factory methods to turn them into Variables.
#
# This writes one file: variable_factories.h

import re
from typing import Optional, List

from tools.codegen.api.types import CppSignatureGroup
from tools.codegen.api import cpp

            

Reported by Pylint.

              
# Add 'at::' to types defined in ATen namespace, e.g. Tensor, TensorList, IntArrayRef and etc.
# TODO: maybe update the cpp argument API to take optional namespace argument?
def fully_qualified_type(argument_type: str) -> str:
    def maybe_optional_type(type: str, is_opt: bool) -> str:
        return f'c10::optional<{type}>' if is_opt else type

    opt_match = OPTIONAL_TYPE_PATTERN.match(argument_type)
    is_opt = opt_match is not None

            

Reported by Pylint.

                  qualified_type = f'{argument_type[:index]}at::{argument_type[index:]}'
    return maybe_optional_type(qualified_type, is_opt)

def gen_variable_factories(out: str, native_yaml_path: str, template_path: str) -> None:
    native_functions = parse_native_yaml(native_yaml_path).native_functions
    fm = FileManager(install_dir=out, template_dir=template_path, dry_run=False)
    fm.write_with_template('variable_factories.h', 'variable_factories.h', lambda: {
        'generated_comment': '@' + f'generated from {fm.template_dir}/variable_factories.h',
        'function_definitions': list(mapMaybe(process_function, native_functions)),

            

Reported by Pylint.

              
def gen_variable_factories(out: str, native_yaml_path: str, template_path: str) -> None:
    native_functions = parse_native_yaml(native_yaml_path).native_functions
    fm = FileManager(install_dir=out, template_dir=template_path, dry_run=False)
    fm.write_with_template('variable_factories.h', 'variable_factories.h', lambda: {
        'generated_comment': '@' + f'generated from {fm.template_dir}/variable_factories.h',
        'function_definitions': list(mapMaybe(process_function, native_functions)),
    })


            

Reported by Pylint.

                  })

@with_native_function
def process_function(f: NativeFunction) -> Optional[str]:
    name = cpp.name(f.func)
    has_tensor_options = python.has_tensor_options(f)
    is_factory = has_tensor_options or name.endswith("_like")

    if Variant.function not in f.variants or not is_factory:

            

Reported by Pylint.

                  })

@with_native_function
def process_function(f: NativeFunction) -> Optional[str]:
    name = cpp.name(f.func)
    has_tensor_options = python.has_tensor_options(f)
    is_factory = has_tensor_options or name.endswith("_like")

    if Variant.function not in f.variants or not is_factory:

            

Reported by Pylint.

              """
This makes the functions in torch._C._VariableFunctions available as
    torch._VF.<funcname>
without mypy being able to find them.

A subset of those functions are mapped to ATen functions in
torch/jit/_builtins.py

See https://github.com/pytorch/pytorch/issues/21478 for the reason for

            

Reported by Pylint.

              
"""
import torch
import sys
import types


class VFModule(types.ModuleType):
    vf: types.ModuleType

            

Reported by Pylint.

              """
import torch
import sys
import types


class VFModule(types.ModuleType):
    vf: types.ModuleType


            

Reported by Pylint.

              import types


class VFModule(types.ModuleType):
    vf: types.ModuleType

    def __init__(self, name):
        super(VFModule, self).__init__(name)
        self.vf = torch._C._VariableFunctions

            

Reported by Pylint.

              import types


class VFModule(types.ModuleType):
    vf: types.ModuleType

    def __init__(self, name):
        super(VFModule, self).__init__(name)
        self.vf = torch._C._VariableFunctions

            

Reported by Pylint.

                  vf: types.ModuleType

    def __init__(self, name):
        super(VFModule, self).__init__(name)
        self.vf = torch._C._VariableFunctions

    def __getattr__(self, attr):
        return getattr(self.vf, attr)


            

Reported by Pylint.

              
    def __init__(self, name):
        super(VFModule, self).__init__(name)
        self.vf = torch._C._VariableFunctions

    def __getattr__(self, attr):
        return getattr(self.vf, attr)



            

Reported by Pylint.

              
    def __init__(self, name):
        super(VFModule, self).__init__(name)
        self.vf = torch._C._VariableFunctions

    def __getattr__(self, attr):
        return getattr(self.vf, attr)



            

Reported by Pylint.

              import unittest

from tools.linter import mypy_wrapper


class TestMypyWrapper(unittest.TestCase):
    configs = {
        'foo.ini': {
            'file1.abc',

            

Reported by Pylint.

              import unittest

from tools.linter import mypy_wrapper


class TestMypyWrapper(unittest.TestCase):
    configs = {
        'foo.ini': {
            'file1.abc',

            

Reported by Pylint.

              from tools.linter import mypy_wrapper


class TestMypyWrapper(unittest.TestCase):
    configs = {
        'foo.ini': {
            'file1.abc',
            'dir2',
            'dir3/file4.xyz',

            

Reported by Pylint.

                      },
    }

    def test_config_files(self) -> None:
        self.assertEqual(mypy_wrapper.config_files().keys(), {
            'mypy.ini',
            'mypy-strict.ini',
        })


            

Reported by Pylint.

                          'mypy-strict.ini',
        })

    def test_split_path(self) -> None:
        self.assertEqual(mypy_wrapper.split_path('file1.abc'), ['file1.abc'])
        self.assertEqual(
            mypy_wrapper.split_path('dir3/file4.xyz'),
            ['dir3', 'file4.xyz'],
        )

            

Reported by Pylint.

                          ['dir2', 'dir5', 'file6.def'],
        )

    def test_make_trie(self) -> None:
        self.assertEqual(mypy_wrapper.make_trie(self.configs), self.trie)

    def test_lookup(self) -> None:
        self.assertEqual(
            mypy_wrapper.lookup(self.trie, 'file1.abc'),

            

Reported by Pylint.

                  def test_make_trie(self) -> None:
        self.assertEqual(mypy_wrapper.make_trie(self.configs), self.trie)

    def test_lookup(self) -> None:
        self.assertEqual(
            mypy_wrapper.lookup(self.trie, 'file1.abc'),
            {'foo.ini', 'bar/baz.ini'},
        )
        self.assertEqual(

            

Reported by Pylint.

                          set(),
        )

    def test_make_plan(self) -> None:
        self.assertEqual(
            mypy_wrapper.make_plan(configs=self.configs, files=[
                'file8.xyz',
                'dir3/file11.abc',
            ]),

            

Reported by Pylint.

                  return text


def main(args: argparse.Namespace) -> None:
    with open(args.version_path) as f:
        version = f.read().strip()
    (major, minor, patch) = parse_version(version)

    replacements = {

            

Reported by Pylint.

                  # Create the output dir if it doesn't exist.
    os.makedirs(os.path.dirname(args.output_path), exist_ok=True)

    with open(args.template_path) as input:
        with open(args.output_path, "w") as output:
            for line in input.readlines():
                output.write(apply_replacements(replacements, line))



            

Reported by Pylint.

              # Ideally, there would be a way in Bazel to parse version.txt
# and use the version numbers from there as substitutions for
# an expand_template action. Since there isn't, this silly script exists.

import argparse
import os
from typing import Dict, Tuple, cast

Version = Tuple[int, int, int]

            

Reported by Pylint.

                  """
    # Extract version number part (i.e. toss any revision / hash parts).
    version_number_str = version
    for i in range(len(version)):
        c = version[i]
        if not (c.isdigit() or c == "."):
            version_number_str = version[:i]
            break


            

Reported by Pylint.

                  # Extract version number part (i.e. toss any revision / hash parts).
    version_number_str = version
    for i in range(len(version)):
        c = version[i]
        if not (c.isdigit() or c == "."):
            version_number_str = version[:i]
            break

    return cast(Version, tuple([int(n) for n in version_number_str.split(".")]))

            

Reported by Pylint.

                          version_number_str = version[:i]
            break

    return cast(Version, tuple([int(n) for n in version_number_str.split(".")]))


def apply_replacements(replacements: Dict[str, str], text: str) -> str:
    """
    Applies the given replacements within the text.

            

Reported by Pylint.

                  return text


def main(args: argparse.Namespace) -> None:
    with open(args.version_path) as f:
        version = f.read().strip()
    (major, minor, patch) = parse_version(version)

    replacements = {

            

Reported by Pylint.

              

def main(args: argparse.Namespace) -> None:
    with open(args.version_path) as f:
        version = f.read().strip()
    (major, minor, patch) = parse_version(version)

    replacements = {
        "@TORCH_VERSION_MAJOR@": str(major),

            

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

@tensorrt_converter(torch.flatten)
def torch_flatten(network, target, args, kwargs, name):
    # args/kwargs should have already been normalized to kwargs

            

Reported by Pylint.

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

from .helper_functions import mark_as_int8_layer

@tensorrt_converter(torch.flatten)
def torch_flatten(network, target, args, kwargs, name):
    # args/kwargs should have already been normalized to kwargs
    assert len(args) == 0

            

Reported by Pylint.

              
from .helper_functions import mark_as_int8_layer

@tensorrt_converter(torch.flatten)
def torch_flatten(network, target, args, kwargs, name):
    # args/kwargs should have already been normalized to kwargs
    assert len(args) == 0
    input_val = kwargs["input"]


            

Reported by Pylint.

              from .helper_functions import mark_as_int8_layer

@tensorrt_converter(torch.flatten)
def torch_flatten(network, target, args, kwargs, 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.

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

from .helper_functions import mark_as_int8_layer

@tensorrt_converter(torch.flatten)
def torch_flatten(network, target, args, kwargs, name):
    # args/kwargs should have already been normalized to kwargs

            

Reported by Pylint.

              from .helper_functions import mark_as_int8_layer

@tensorrt_converter(torch.flatten)
def torch_flatten(network, target, args, kwargs, 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.flatten)
def torch_flatten(network, target, args, kwargs, 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"Flatten received input {input_val} that is not part "
                           "of the TensorRT region!")

            

Reported by Bandit.

                  start_dim = kwargs["start_dim"] - 1
    end_dim = len(input_val.shape) if kwargs["end_dim"] == -1 else kwargs["end_dim"] - 1

    assert start_dim >= 0, "Expect non negtive start_dim, this probably due to flatten batch dim."

    new_shape = []
    flatten_dim = 1
    for i, dim in enumerate(input_val.shape):
        if i < start_dim:

            

Reported by Bandit.

              import torch.nn as nn
import torch.nn.init as init


class SuperResolutionNet(nn.Module):
    def __init__(self, upscale_factor):
        super(SuperResolutionNet, self).__init__()

        self.relu = nn.ReLU()

            

Reported by Pylint.

              import torch.nn as nn
import torch.nn.init as init


class SuperResolutionNet(nn.Module):
    def __init__(self, upscale_factor):
        super(SuperResolutionNet, self).__init__()

        self.relu = nn.ReLU()

            

Reported by Pylint.

              import torch.nn as nn
import torch.nn.init as init


class SuperResolutionNet(nn.Module):
    def __init__(self, upscale_factor):
        super(SuperResolutionNet, self).__init__()

        self.relu = nn.ReLU()

            

Reported by Pylint.

              import torch.nn.init as init


class SuperResolutionNet(nn.Module):
    def __init__(self, upscale_factor):
        super(SuperResolutionNet, self).__init__()

        self.relu = nn.ReLU()
        self.conv1 = nn.Conv2d(1, 64, (5, 5), (1, 1), (2, 2))

            

Reported by Pylint.

              import torch.nn.init as init


class SuperResolutionNet(nn.Module):
    def __init__(self, upscale_factor):
        super(SuperResolutionNet, self).__init__()

        self.relu = nn.ReLU()
        self.conv1 = nn.Conv2d(1, 64, (5, 5), (1, 1), (2, 2))

            

Reported by Pylint.

              
class SuperResolutionNet(nn.Module):
    def __init__(self, upscale_factor):
        super(SuperResolutionNet, self).__init__()

        self.relu = nn.ReLU()
        self.conv1 = nn.Conv2d(1, 64, (5, 5), (1, 1), (2, 2))
        self.conv2 = nn.Conv2d(64, 64, (3, 3), (1, 1), (1, 1))
        self.conv3 = nn.Conv2d(64, 32, (3, 3), (1, 1), (1, 1))

            

Reported by Pylint.

              
        self._initialize_weights()

    def forward(self, x):
        x = self.relu(self.conv1(x))
        x = self.relu(self.conv2(x))
        x = self.relu(self.conv3(x))
        x = self.pixel_shuffle(self.conv4(x))
        return x

            

Reported by Pylint.

              
        self._initialize_weights()

    def forward(self, x):
        x = self.relu(self.conv1(x))
        x = self.relu(self.conv2(x))
        x = self.relu(self.conv3(x))
        x = self.pixel_shuffle(self.conv4(x))
        return x

            

Reported by Pylint.