import operator_benchmark as op_bench
import torch
import random
from typing import List


"""Microbenchmarks for Stack operator"""

# Configs for PT stack operator

            

Reported by Pylint.

              """Microbenchmarks for Stack operator"""

# Configs for PT stack operator
stack_configs_static_runtime = op_bench.config_list(
    attr_names=['sizes', 'N'],
    attrs=[
        [(20, 40), 5],
        [(1, 40), 5],
    ],

            

Reported by Pylint.

                  tags=['static_runtime'],
)

stack_configs_short = op_bench.config_list(
    attr_names=['sizes', 'N'],
    attrs=[
        [(1,    1,      1), 2],  # noqa: E241
        [(512,  512,    2), 2],  # noqa: E241
        [(128, 1024,    2), 2],  # noqa: E241

            

Reported by Pylint.

                  tags=['short'],
)

stack_configs_long = op_bench.config_list(
    attr_names=['sizes', 'N'],
    attrs=[
        [(2**10,    2**10,      2), 2],  # noqa: E241
        [(2**10+1,  2**10-1,    2), 2],  # noqa: E226,E241
        [(2**10,    2**10,      2), 2],  # noqa: E241

            

Reported by Pylint.

              )

# There is a different codepath on CUDA for >4 dimensions
stack_configs_multidim = op_bench.config_list(
    attr_names=['sizes', 'N'],
    attrs=[
        [(2**6,     2**5,   2**2,   2**4,   2**5), 2],  # noqa: E241
        [(2**4,     2**5,   2**2,   2**4,   2**5), 8],  # noqa: E241
        [(2**3+1,   2**5-1, 2**2+1, 2**4-1, 2**5+1), 17],  # noqa: E226,E241

            

Reported by Pylint.

                  tags=['multidim'],
)

class StackBenchmark(op_bench.TorchBenchmarkBase):
    def init(self, sizes, N, dim, device):
        random.seed(42)
        inputs = []
        gen_sizes = []
        if type(sizes) == list and N == -1:

            

Reported by Pylint.

                      return torch.stack(inputs, dim=dim, out=result)


op_bench.generate_pt_test(stack_configs_static_runtime +
                          stack_configs_short +
                          stack_configs_long +
                          stack_configs_multidim,
                          StackBenchmark)


            

Reported by Pylint.

              from typing import List


"""Microbenchmarks for Stack operator"""

# Configs for PT stack operator
stack_configs_static_runtime = op_bench.config_list(
    attr_names=['sizes', 'N'],
    attrs=[

            

Reported by Pylint.

                      if type(sizes) == list and N == -1:
            gen_sizes = sizes
        else:
            for i in range(N):
                gen_sizes.append([old_size() if callable(old_size) else old_size for old_size in sizes])

        for s in gen_sizes:
            inputs.append(torch.rand(s, device=device))
        result = torch.rand(gen_sizes[0], device=device)

            

Reported by Pylint.

                      for s in gen_sizes:
            inputs.append(torch.rand(s, device=device))
        result = torch.rand(gen_sizes[0], device=device)
        self.inputs = {
            "result": result,
            "inputs": inputs,
            "dim": dim
        }
        self.set_module_name('stack')

            

Reported by Pylint.

              import operator_benchmark as op_bench
from pt import (  # noqa: F401
    qactivation_test,
    qarithmetic_test,
    qbatchnorm_test,
    qcat_test,
    qcomparators_test,
    qconv_test,
    qgroupnorm_test,

            

Reported by Pylint.

              import operator_benchmark as op_bench
from pt import (  # noqa: F401
    qactivation_test,
    qarithmetic_test,
    qbatchnorm_test,
    qcat_test,
    qcomparators_test,
    qconv_test,
    qgroupnorm_test,

            

Reported by Pylint.

              import operator_benchmark as op_bench
from pt import (  # noqa: F401
    qactivation_test,
    qarithmetic_test,
    qbatchnorm_test,
    qcat_test,
    qcomparators_test,
    qconv_test,
    qgroupnorm_test,

            

Reported by Pylint.

              import operator_benchmark as op_bench
from pt import (  # noqa: F401
    qactivation_test,
    qarithmetic_test,
    qbatchnorm_test,
    qcat_test,
    qcomparators_test,
    qconv_test,
    qgroupnorm_test,

            

Reported by Pylint.

              import operator_benchmark as op_bench
from pt import (  # noqa: F401
    qactivation_test,
    qarithmetic_test,
    qbatchnorm_test,
    qcat_test,
    qcomparators_test,
    qconv_test,
    qgroupnorm_test,

            

Reported by Pylint.

              import operator_benchmark as op_bench
from pt import (  # noqa: F401
    qactivation_test,
    qarithmetic_test,
    qbatchnorm_test,
    qcat_test,
    qcomparators_test,
    qconv_test,
    qgroupnorm_test,

            

Reported by Pylint.

              import operator_benchmark as op_bench
from pt import (  # noqa: F401
    qactivation_test,
    qarithmetic_test,
    qbatchnorm_test,
    qcat_test,
    qcomparators_test,
    qconv_test,
    qgroupnorm_test,

            

Reported by Pylint.

              import operator_benchmark as op_bench
from pt import (  # noqa: F401
    qactivation_test,
    qarithmetic_test,
    qbatchnorm_test,
    qcat_test,
    qcomparators_test,
    qconv_test,
    qgroupnorm_test,

            

Reported by Pylint.

              import operator_benchmark as op_bench
from pt import (  # noqa: F401
    qactivation_test,
    qarithmetic_test,
    qbatchnorm_test,
    qcat_test,
    qcomparators_test,
    qconv_test,
    qgroupnorm_test,

            

Reported by Pylint.

              import operator_benchmark as op_bench
from pt import (  # noqa: F401
    qactivation_test,
    qarithmetic_test,
    qbatchnorm_test,
    qcat_test,
    qcomparators_test,
    qconv_test,
    qgroupnorm_test,

            

Reported by Pylint.

              import torch
import functools
import random
import operator
import numpy as np
import time

# shim for torch.cuda.Event when running on cpu
class Event(object):

            

Reported by Pylint.

                      pass

    def record(self):
        self.time = time.perf_counter()

    def elapsed_time(self, end_event):
        assert isinstance(end_event, Event)
        return end_event.time - self.time


            

Reported by Pylint.

                  dense = np.random.randn(*shape)
    values = []
    indices = [[], []]
    for n in range(nnz):
        row = random.randint(0, shape[0] - 1)
        col = random.randint(0, shape[1] - 1)
        indices[0].append(row)
        indices[1].append(col)
        values.append(dense[row, col])

            

Reported by Pylint.

              import torch
import functools
import random
import operator
import numpy as np
import time

# shim for torch.cuda.Event when running on cpu
class Event(object):

            

Reported by Pylint.

              import torch
import functools
import random
import operator
import numpy as np
import time

# shim for torch.cuda.Event when running on cpu
class Event(object):

            

Reported by Pylint.

              import torch
import functools
import random
import operator
import numpy as np
import time

# shim for torch.cuda.Event when running on cpu
class Event(object):

            

Reported by Pylint.

              import torch
import functools
import random
import operator
import numpy as np
import time

# shim for torch.cuda.Event when running on cpu
class Event(object):

            

Reported by Pylint.

              import random
import operator
import numpy as np
import time

# shim for torch.cuda.Event when running on cpu
class Event(object):
    def __init__(self, enable_timing):
        pass

            

Reported by Pylint.

              import time

# shim for torch.cuda.Event when running on cpu
class Event(object):
    def __init__(self, enable_timing):
        pass

    def record(self):
        self.time = time.perf_counter()

            

Reported by Pylint.

              import time

# shim for torch.cuda.Event when running on cpu
class Event(object):
    def __init__(self, enable_timing):
        pass

    def record(self):
        self.time = time.perf_counter()

            

Reported by Pylint.

                      self.reg_func(net, reg)
        net.Mul([weight, x], weighted_x)
        net.Add([weighted_x, reg], weighted_x_add_reg)
        net.SumElements(weighted_x_add_reg, self.output_schema())
        if enable_diagnose:
            for i in range(self.num):
                net.Slice(weight, self.weight_i[i], starts=[i], ends=[i + 1])

    def add_ops(self, net):

            

Reported by Pylint.

              from caffe2.python.regularizer import BoundedGradientProjection, LogBarrier


"""
Implementation of adaptive weighting: https://arxiv.org/pdf/1705.07115.pdf
"""


class AdaptiveWeight(ModelLayer):

            

Reported by Pylint.

                          "GivenTensorFill",
            {"values": values, "dtype": core.DataType.FLOAT},
        )
        self.mu = self.create_param(
            param_name="mu",
            shape=[self.num],
            initializer=initializer,
            optimizer=self.optimizer,
        )

            

Reported by Pylint.

                          optimizer=self.optimizer,
        )

    def log_std_weight(self, x, net, weight):
        """
        min 1 / 2 / e^mu X + mu / 2
        """
        mu_neg = net.NextScopedBlob("mu_neg")
        net.Negative(self.mu, mu_neg)

            

Reported by Pylint.

                                  self.pos_optim_method
                )
            )
        self.k = self.create_param(
            param_name="k",
            shape=[self.num],
            initializer=initializer,
            optimizer=self.optimizer,
            regularizer=regularizer,

            

Reported by Pylint.

                          regularizer=regularizer,
        )

    def inv_var_weight(self, x, net, weight):
        net.Scale(self.k, weight, scale=0.5)

    def inv_var_reg(self, net, reg):
        log_k = net.NextScopedBlob("log_k")
        net.Log(self.k, log_k)

            

Reported by Pylint.

              # @package adaptive_weight
# Module caffe2.fb.python.layers.adaptive_weight


import numpy as np
from caffe2.python import core, schema
from caffe2.python.layers.layers import ModelLayer
from caffe2.python.regularizer import BoundedGradientProjection, LogBarrier


            

Reported by Pylint.

              """


class AdaptiveWeight(ModelLayer):
    def __init__(
        self,
        model,
        input_record,
        name="adaptive_weight",

            

Reported by Pylint.

              """


class AdaptiveWeight(ModelLayer):
    def __init__(
        self,
        model,
        input_record,
        name="adaptive_weight",

            

Reported by Pylint.

              

class AdaptiveWeight(ModelLayer):
    def __init__(
        self,
        model,
        input_record,
        name="adaptive_weight",
        optimizer=None,

            

Reported by Pylint.

              




from caffe2.python import core, workspace
import caffe2.python.hypothesis_test_util as hu

import numpy as np

            

Reported by Pylint.

              import numpy as np


def FakeQuantization8BitsRowwise(data):
    min_el = np.min(data, axis=1)
    max_el = np.max(data, axis=1)
    scale = (max_el - min_el) / 255.
    bias = min_el
    inv_scale = 1. / scale

            

Reported by Pylint.

              import numpy as np


def FakeQuantization8BitsRowwise(data):
    min_el = np.min(data, axis=1)
    max_el = np.max(data, axis=1)
    scale = (max_el - min_el) / 255.
    bias = min_el
    inv_scale = 1. / scale

            

Reported by Pylint.

                  return data.T


class TestQuantize8bits(hu.HypothesisTestCase):

    def test_quantize_op(self):
        op = core.CreateOperator(
            'FloatToRowwiseQuantized8Bits',
            ['input_data'],

            

Reported by Pylint.

              
class TestQuantize8bits(hu.HypothesisTestCase):

    def test_quantize_op(self):
        op = core.CreateOperator(
            'FloatToRowwiseQuantized8Bits',
            ['input_data'],
            ['quantized_input', 'scale_bias'])
        input_data = np.float32(np.asarray([[801., 786, 235.2, 2353.3434],

            

Reported by Pylint.

              
class TestQuantize8bits(hu.HypothesisTestCase):

    def test_quantize_op(self):
        op = core.CreateOperator(
            'FloatToRowwiseQuantized8Bits',
            ['input_data'],
            ['quantized_input', 'scale_bias'])
        input_data = np.float32(np.asarray([[801., 786, 235.2, 2353.3434],

            

Reported by Pylint.

              class TestQuantize8bits(hu.HypothesisTestCase):

    def test_quantize_op(self):
        op = core.CreateOperator(
            'FloatToRowwiseQuantized8Bits',
            ['input_data'],
            ['quantized_input', 'scale_bias'])
        input_data = np.float32(np.asarray([[801., 786, 235.2, 2353.3434],
                                            [5., 11., 9., -2.]]))

            

Reported by Pylint.

                      np.testing.assert_array_almost_equal(
            result, ground_truth)

    def test_quantize_tensor_with_const_row_op(self):
        op = core.CreateOperator(
            'FloatToRowwiseQuantized8Bits',
            ['input_data'],
            ['quantized_input', 'scale_bias'])
        input_data = np.float32(np.asarray([[801., 786, 235.2, 2353.3434],

            

Reported by Pylint.

                      np.testing.assert_array_almost_equal(
            result, ground_truth)

    def test_quantize_tensor_with_const_row_op(self):
        op = core.CreateOperator(
            'FloatToRowwiseQuantized8Bits',
            ['input_data'],
            ['quantized_input', 'scale_bias'])
        input_data = np.float32(np.asarray([[801., 786, 235.2, 2353.3434],

            

Reported by Pylint.

                          result, ground_truth)

    def test_quantize_tensor_with_const_row_op(self):
        op = core.CreateOperator(
            'FloatToRowwiseQuantized8Bits',
            ['input_data'],
            ['quantized_input', 'scale_bias'])
        input_data = np.float32(np.asarray([[801., 786, 235.2, 2353.3434],
                                            [9., 9., 9., 9.]]))

            

Reported by Pylint.

              import operator_benchmark as op_bench
import torch


"""Microbenchmarks for remainder operators."""


# Benchmark ops performance with broadcast
remainder_ops_list = op_bench.op_list(

            

Reported by Pylint.

              

# Benchmark ops performance with broadcast
remainder_ops_list = op_bench.op_list(
    attr_names=['op_name', 'op_func'],
    attrs=[
        ['fmod', torch.fmod],
        ['remainder', torch.remainder],
    ],

            

Reported by Pylint.

                  ],
)

remainder_short_configs = op_bench.config_list(
    attr_names=['M', 'N', 'K'],
    attrs=[
        [1, 1, 1],
        [64, 64, 64],
        [64, 64, 128],

            

Reported by Pylint.

                  tags=['short'],
)

remainder_long_configs = op_bench.cross_product_configs(
    M=[8, 128],
    N=[32, 64],
    K=[256, 512],
    device=['cpu', 'cuda'],
    dtype=[torch.int32, torch.float, torch.double],

            

Reported by Pylint.

              )


class RemainderOpBenchmark(op_bench.TorchBenchmarkBase):
    def init(self, M, N, K, device, dtype, op_func):
        self.dividend = torch.rand(M, N, K, device=device)
        self.dividend = (self.dividend * 1000 - 500).to(dtype=dtype)

        self.divisor = torch.rand(M, N, K, device=device)

            

Reported by Pylint.

                      return self.op_func(dividend, divisor)


op_bench.generate_pt_tests_from_op_list(remainder_ops_list,
                                        remainder_short_configs + remainder_long_configs,
                                        RemainderOpBenchmark)


if __name__ == "__main__":

            

Reported by Pylint.

              import torch


"""Microbenchmarks for remainder operators."""


# Benchmark ops performance with broadcast
remainder_ops_list = op_bench.op_list(
    attr_names=['op_name', 'op_func'],

            

Reported by Pylint.

              
class RemainderOpBenchmark(op_bench.TorchBenchmarkBase):
    def init(self, M, N, K, device, dtype, op_func):
        self.dividend = torch.rand(M, N, K, device=device)
        self.dividend = (self.dividend * 1000 - 500).to(dtype=dtype)

        self.divisor = torch.rand(M, N, K, device=device)
        # +1 so we don't divide by zero
        self.divisor = (self.divisor * 40 + 1).to(dtype=dtype)

            

Reported by Pylint.

              class RemainderOpBenchmark(op_bench.TorchBenchmarkBase):
    def init(self, M, N, K, device, dtype, op_func):
        self.dividend = torch.rand(M, N, K, device=device)
        self.dividend = (self.dividend * 1000 - 500).to(dtype=dtype)

        self.divisor = torch.rand(M, N, K, device=device)
        # +1 so we don't divide by zero
        self.divisor = (self.divisor * 40 + 1).to(dtype=dtype)


            

Reported by Pylint.

                      self.dividend = torch.rand(M, N, K, device=device)
        self.dividend = (self.dividend * 1000 - 500).to(dtype=dtype)

        self.divisor = torch.rand(M, N, K, device=device)
        # +1 so we don't divide by zero
        self.divisor = (self.divisor * 40 + 1).to(dtype=dtype)

        self.inputs = {
            "dividend": self.dividend,

            

Reported by Pylint.

              Weight Normalization from https://arxiv.org/abs/1602.07868
"""
from torch.nn.parameter import Parameter, UninitializedParameter
from torch import _weight_norm, norm_except_dim
from typing import Any, TypeVar
from ..modules import Module


class WeightNorm(object):

            

Reported by Pylint.

              Weight Normalization from https://arxiv.org/abs/1602.07868
"""
from torch.nn.parameter import Parameter, UninitializedParameter
from torch import _weight_norm, norm_except_dim
from typing import Any, TypeVar
from ..modules import Module


class WeightNorm(object):

            

Reported by Pylint.

              from torch.nn.parameter import Parameter, UninitializedParameter
from torch import _weight_norm, norm_except_dim
from typing import Any, TypeVar
from ..modules import Module


class WeightNorm(object):
    name: str
    dim: int

            

Reported by Pylint.

                      self.name = name
        self.dim = dim

    # TODO Make return type more specific
    def compute_weight(self, module: Module) -> Any:
        g = getattr(module, self.name + '_g')
        v = getattr(module, self.name + '_v')
        return _weight_norm(v, g, self.dim)


            

Reported by Pylint.

              
    @staticmethod
    def apply(module, name: str, dim: int) -> 'WeightNorm':
        for k, hook in module._forward_pre_hooks.items():
            if isinstance(hook, WeightNorm) and hook.name == name:
                raise RuntimeError("Cannot register two weight_norm hooks on "
                                   "the same parameter {}".format(name))

        if dim is None:

            

Reported by Pylint.

              
    @staticmethod
    def apply(module, name: str, dim: int) -> 'WeightNorm':
        for k, hook in module._forward_pre_hooks.items():
            if isinstance(hook, WeightNorm) and hook.name == name:
                raise RuntimeError("Cannot register two weight_norm hooks on "
                                   "the same parameter {}".format(name))

        if dim is None:

            

Reported by Pylint.

                              'The module passed to `WeightNorm` can\'t have uninitialized parameters. '
                'Make sure to run the dummy forward before applying weight normalization')
        # remove w from parameter list
        del module._parameters[name]

        # add g and v as new parameters and express w as g/||v|| * v
        module.register_parameter(name + '_g', Parameter(norm_except_dim(weight, 2, dim).data))
        module.register_parameter(name + '_v', Parameter(weight.data))
        setattr(module, name, fn.compute_weight(module))

            

Reported by Pylint.

                  def remove(self, module: Module) -> None:
        weight = self.compute_weight(module)
        delattr(module, self.name)
        del module._parameters[self.name + '_g']
        del module._parameters[self.name + '_v']
        setattr(module, self.name, Parameter(weight.data))

    def __call__(self, module: Module, inputs: Any) -> None:
        setattr(module, self.name, self.compute_weight(module))

            

Reported by Pylint.

                      weight = self.compute_weight(module)
        delattr(module, self.name)
        del module._parameters[self.name + '_g']
        del module._parameters[self.name + '_v']
        setattr(module, self.name, Parameter(weight.data))

    def __call__(self, module: Module, inputs: Any) -> None:
        setattr(module, self.name, self.compute_weight(module))


            

Reported by Pylint.

                      >>> m = weight_norm(nn.Linear(20, 40))
        >>> remove_weight_norm(m)
    """
    for k, hook in module._forward_pre_hooks.items():
        if isinstance(hook, WeightNorm) and hook.name == name:
            hook.remove(module)
            del module._forward_pre_hooks[k]
            return module


            

Reported by Pylint.

              import sys
import warnings

from . import _prctl_pr_set_pdeathsig  # type: ignore[attr-defined]


class ProcessException(Exception):
    __slots__ = ["error_index", "error_pid"]


            

Reported by Pylint.

                      if self.error_queues[error_index].empty():
            exitcode = self.processes[error_index].exitcode
            if exitcode < 0:
                name = signal.Signals(-exitcode).name
                raise ProcessExitedException(
                    "process %d terminated with signal %s" %
                    (error_index, name),
                    error_index=error_index,
                    error_pid=failed_process.pid,

            

Reported by Pylint.

                  Exception is thrown when the process failed due to exception
    raised by the code.
    """
    def __init__(
        self,
        msg: str,
        error_index: int,
        error_pid: int,
    ):

            

Reported by Pylint.

                      fn(i, *args)
    except KeyboardInterrupt:
        pass  # SIGINT; Killed by parent, do nothing
    except Exception:
        # Propagate exception to parent process, keeping original traceback
        import traceback
        error_queue.put(traceback.format_exc())
        sys.exit(1)


            

Reported by Pylint.

                  def __init__(self, processes, error_queues):
        warnings.warn('SpawnContext is renamed to ProcessContext since 1.4 release.')
        super(SpawnContext, self).__init__(processes, error_queues)
    pass


# Note: [start_processes]
# mp.start_processes handles both start_method='spawn' and 'fork'. It's supposed to be a
# more generalized API than mp.spawn. Currently we only document mp.spawn as it's the

            

Reported by Pylint.

              
from typing import Optional
import multiprocessing
import multiprocessing.connection
import signal
import sys
import warnings

from . import _prctl_pr_set_pdeathsig  # type: ignore[attr-defined]

            

Reported by Pylint.

              from . import _prctl_pr_set_pdeathsig  # type: ignore[attr-defined]


class ProcessException(Exception):
    __slots__ = ["error_index", "error_pid"]

    def __init__(self, msg: str, error_index: int, pid: int):
        super().__init__(msg)
        self.error_index = error_index

            

Reported by Pylint.

                  """
    __slots__ = ["exit_code"]

    def __init__(
            self, msg: str, error_index: int, error_pid: int,
            exit_code: int, signal_name: Optional[str] = None
    ):
        super().__init__(msg, error_index, error_pid)
        self.exit_code = exit_code

            

Reported by Pylint.

                      self.signal_name = signal_name


def _wrap(fn, i, args, error_queue):
    # prctl(2) is a Linux specific system call.
    # On other systems the following function call has no effect.
    # This is set to ensure that non-daemonic child processes can
    # terminate if their parent terminates before they do.
    _prctl_pr_set_pdeathsig(signal.SIGINT)

            

Reported by Pylint.

                      pass  # SIGINT; Killed by parent, do nothing
    except Exception:
        # Propagate exception to parent process, keeping original traceback
        import traceback
        error_queue.put(traceback.format_exc())
        sys.exit(1)


class ProcessContext:

            

Reported by Pylint.

                      code = compile(dest_ast, "", "exec")
        globals_dict = copy.copy(fn.__globals__)
        keys_before = set(globals_dict.keys())
        exec(code, globals_dict)
        new_keys = list(set(globals_dict.keys()) - keys_before)
        assert len(new_keys) == 1
        fn_compiled = globals_dict[new_keys[0]]

        # Return the correct FunctionType object

            

Reported by Bandit.

                      code = compile(dest_ast, "", "exec")
        globals_dict = copy.copy(fn.__globals__)
        keys_before = set(globals_dict.keys())
        exec(code, globals_dict)
        new_keys = list(set(globals_dict.keys()) - keys_before)
        assert len(new_keys) == 1
        fn_compiled = globals_dict[new_keys[0]]

        # Return the correct FunctionType object

            

Reported by Pylint.

              import ast
import inspect
import textwrap
import copy
from types import FunctionType
from typing import cast, Union, Callable, Dict, Optional, Any
from torch.fx._symbolic_trace import Tracer
from torch.fx.graph import Graph
from torch._sources import normalize_source_lines

            

Reported by Pylint.

              from torch._sources import normalize_source_lines
import torch

class AST_Rewriter(ast.NodeTransformer):
    """
    Take a FunctionType object representing a `forward` method, then
    perform an AST rewrite to swap out nodes that are not symbolically
    traceable with a callsite to the FX alternative.


            

Reported by Pylint.

                  https://docs.python.org/3/library/ast.html#ast.NodeTransformer
    """

    def rewrite(self, fn: FunctionType):

        # Normalize the source lines
        sourcelines, _ = inspect.getsourcelines(fn)
        sourcelines = normalize_source_lines(sourcelines)
        source = ''.join(sourcelines)

            

Reported by Pylint.

                  https://docs.python.org/3/library/ast.html#ast.NodeTransformer
    """

    def rewrite(self, fn: FunctionType):

        # Normalize the source lines
        sourcelines, _ = inspect.getsourcelines(fn)
        sourcelines = normalize_source_lines(sourcelines)
        source = ''.join(sourcelines)

            

Reported by Pylint.

                      keys_before = set(globals_dict.keys())
        exec(code, globals_dict)
        new_keys = list(set(globals_dict.keys()) - keys_before)
        assert len(new_keys) == 1
        fn_compiled = globals_dict[new_keys[0]]

        # Return the correct FunctionType object
        return fn_compiled


            

Reported by Bandit.

                      # Return the correct FunctionType object
        return fn_compiled

    def visit_Assert(self, node):
        """
        Swap out the Assert node (Python's `assert`) with a callsite to the
        symbolically-traceable torch._assert function
        """
        # Create the Call node

            

Reported by Pylint.

                      # Return the correct FunctionType object
        return fn_compiled

    def visit_Assert(self, node):
        """
        Swap out the Assert node (Python's `assert`) with a callsite to the
        symbolically-traceable torch._assert function
        """
        # Create the Call node

            

Reported by Pylint.

                      symbolically-traceable torch._assert function
        """
        # Create the Call node
        n = ast.parse('torch._assert()', mode='eval')
        assert isinstance(n, ast.Expression)
        call_node = n.body
        assert isinstance(call_node, ast.Call)
        msg = node.msg if node.msg else ast.Constant(value="", kind=None)
        call_node.args = [node.test, msg]

            

Reported by Pylint.

              from collections import defaultdict, abc as container_abcs

import torch
from copy import deepcopy
from itertools import chain
import warnings
import functools



            

Reported by Pylint.

              from collections import defaultdict, abc as container_abcs

import torch
from copy import deepcopy
from itertools import chain
import warnings
import functools



            

Reported by Pylint.

              
import torch
from copy import deepcopy
from itertools import chain
import warnings
import functools


class _RequiredParameter(object):

            

Reported by Pylint.

              import torch
from copy import deepcopy
from itertools import chain
import warnings
import functools


class _RequiredParameter(object):
    """Singleton class representing a required parameter for an Optimizer."""

            

Reported by Pylint.

              from copy import deepcopy
from itertools import chain
import warnings
import functools


class _RequiredParameter(object):
    """Singleton class representing a required parameter for an Optimizer."""
    def __repr__(self):

            

Reported by Pylint.

              import functools


class _RequiredParameter(object):
    """Singleton class representing a required parameter for an Optimizer."""
    def __repr__(self):
        return "<required parameter>"

required = _RequiredParameter()

            

Reported by Pylint.

              import functools


class _RequiredParameter(object):
    """Singleton class representing a required parameter for an Optimizer."""
    def __repr__(self):
        return "<required parameter>"

required = _RequiredParameter()

            

Reported by Pylint.

              required = _RequiredParameter()


class Optimizer(object):
    r"""Base class for all optimizers.

    .. warning::
        Parameters need to be specified as collections that have a deterministic
        ordering that is consistent between runs. Examples of objects that don't

            

Reported by Pylint.

                      return format_string

    def _hook_for_profile(self):
        self._zero_grad_profile_name = "Optimizer.zero_grad#{}.zero_grad".format(self.__class__.__name__)

        def profile_hook_step(func):

            @functools.wraps(func)
            def wrapper(*args, **kwargs):

            

Reported by Pylint.

                                           "that doesn't match the size of optimizer's group")

        # Update the state
        id_map = {old_id: p for old_id, p in
                  zip(chain.from_iterable((g['params'] for g in saved_groups)),
                      chain.from_iterable((g['params'] for g in groups)))}

        def cast(param, value):
            r"""Make a deep copy of value, casting all tensors to device of param."""

            

Reported by Pylint.