#
# This source code is licensed under the BSD license found in the
# LICENSE file in the root directory of this source tree.
import pytest
import torch
from torch import nn

from torch.distributed.pipeline.sync import Pipe


            

Reported by Pylint.

              # This source code is licensed under the BSD license found in the
# LICENSE file in the root directory of this source tree.
import pytest
import torch
from torch import nn

from torch.distributed.pipeline.sync import Pipe



            

Reported by Pylint.

              # LICENSE file in the root directory of this source tree.
import pytest
import torch
from torch import nn

from torch.distributed.pipeline.sync import Pipe


def test_inplace_on_requires_grad(setup_rpc):

            

Reported by Pylint.

              import torch
from torch import nn

from torch.distributed.pipeline.sync import Pipe


def test_inplace_on_requires_grad(setup_rpc):
    model = nn.Sequential(nn.Linear(1, 1), nn.ReLU(inplace=True))
    model = Pipe(model, checkpoint="always")

            

Reported by Pylint.

              from torch.distributed.pipeline.sync import Pipe


def test_inplace_on_requires_grad(setup_rpc):
    model = nn.Sequential(nn.Linear(1, 1), nn.ReLU(inplace=True))
    model = Pipe(model, checkpoint="always")

    x = torch.rand(1)
    y = model(x).local_value()

            

Reported by Pylint.

              

@pytest.mark.xfail(strict=True)
def test_inplace_on_not_requires_grad(setup_rpc):
    # In-place operation on a tensor not requiring grad doesn't cause a
    # RuntimeError. Currently, we cannot detect this case.
    model = nn.Sequential(nn.ReLU(inplace=True))
    model = Pipe(model, [1], devices=["cpu"], checkpoint="always")


            

Reported by Pylint.

              

@pytest.mark.xfail(strict=True)
def test_inplace_incorrect_grad(setup_rpc):
    class M(nn.Module):
        def forward(self, foo_bar):
            # 'foo' requires grad but 'bar' does not. In-place operation on
            # 'bar' won't cause a RuntimeError.
            foo, bar = foo_bar

            

Reported by Pylint.

              # Copyright 2019 Kakao Brain
#
# Copyright (c) Facebook, Inc. and its affiliates. All rights reserved.
#
# This source code is licensed under the BSD license found in the
# LICENSE file in the root directory of this source tree.
import pytest
import torch
from torch import nn

            

Reported by Pylint.

              from torch.distributed.pipeline.sync import Pipe


def test_inplace_on_requires_grad(setup_rpc):
    model = nn.Sequential(nn.Linear(1, 1), nn.ReLU(inplace=True))
    model = Pipe(model, checkpoint="always")

    x = torch.rand(1)
    y = model(x).local_value()

            

Reported by Pylint.

                  model = nn.Sequential(nn.Linear(1, 1), nn.ReLU(inplace=True))
    model = Pipe(model, checkpoint="always")

    x = torch.rand(1)
    y = model(x).local_value()

    message = r"a leaf Variable that requires grad .* used in an in-place operation."
    with pytest.raises(RuntimeError, match=message):
        y.backward()

            

Reported by Pylint.

              
        # also add to the output list the intermediate outputs of fwd_step that
        # are used by backward.
        ssa, blob_versions = core.get_ssa(cell_net.Proto())
        scratches = [
            blob
            for blob, ver in viewitems(blob_versions)
            if (ver > 0 and
                blob in undefined and

            

Reported by Pylint.

                                      [output_blob],
                    )

    def map_to_dual_list(m):
        return [str(x) for x in list(m.keys())] + \
               [str(x) for x in list(m.values())]

    backward_args = {}
    if backward_cell_net is not None:

            

Reported by Pylint.

              ## @package recurrent
# Module caffe2.python.recurrent





from caffe2.python import core, workspace
from future.utils import viewitems, viewkeys

            

Reported by Pylint.

              from caffe2.python import core, workspace
from future.utils import viewitems, viewkeys

def recurrent_net(
        net, cell_net, inputs, initial_cell_inputs,
        links, timestep=None, scope=None, outputs_with_grads=(0,),
        recompute_blobs_on_backward=None, forward_only=False,
):
    '''

            

Reported by Pylint.

              from caffe2.python import core, workspace
from future.utils import viewitems, viewkeys

def recurrent_net(
        net, cell_net, inputs, initial_cell_inputs,
        links, timestep=None, scope=None, outputs_with_grads=(0,),
        recompute_blobs_on_backward=None, forward_only=False,
):
    '''

            

Reported by Pylint.

              from caffe2.python import core, workspace
from future.utils import viewitems, viewkeys

def recurrent_net(
        net, cell_net, inputs, initial_cell_inputs,
        links, timestep=None, scope=None, outputs_with_grads=(0,),
        recompute_blobs_on_backward=None, forward_only=False,
):
    '''

            

Reported by Pylint.

              from caffe2.python import core, workspace
from future.utils import viewitems, viewkeys

def recurrent_net(
        net, cell_net, inputs, initial_cell_inputs,
        links, timestep=None, scope=None, outputs_with_grads=(0,),
        recompute_blobs_on_backward=None, forward_only=False,
):
    '''

            

Reported by Pylint.

              
    forward_only: if True, only forward steps are executed
    '''
    assert len(inputs) == 1, "Only one input blob is supported so far"

    input_blobs = [str(i[0]) for i in inputs]
    initial_input_blobs = [str(x[1]) for x in initial_cell_inputs]
    op_name = net.NextName('recurrent')


            

Reported by Bandit.

                  initial_input_blobs = [str(x[1]) for x in initial_cell_inputs]
    op_name = net.NextName('recurrent')

    def s(name):
        # We have to manually scope due to our internal/external blob
        # relationships.
        scope_name = op_name if scope is None else scope
        return "{}/{}".format(str(scope_name), str(name))


            

Reported by Pylint.

                          recompute_blobs_on_backward = {str(b) for b in
                                           recompute_blobs_on_backward}

            for op in cell_net.Proto().op:
                if not recompute_blobs_on_backward.isdisjoint(set(op.output)):
                    backward_cell_net.Proto().op.extend([op])
                    # This fires if other outputs than the declared
                    # are computed by the ops that are recomputed
                    assert set(op.output).issubset(recompute_blobs_on_backward)

            

Reported by Pylint.

              import numpy as np
import numpy.testing as npt

from hypothesis import given, settings
import hypothesis.strategies as st

import functools



            

Reported by Pylint.

              import numpy.testing as npt

from hypothesis import given, settings
import hypothesis.strategies as st

import functools


def primefac(n):

            

Reported by Pylint.

                      workspace.RunNetOnce(net)

        for idx in range(3):
            self.assertEquals(workspace.FetchBlob(results[idx]), [1.0])

    def test_rebatching_queue_multi_enqueue_dequeue(self):
        net = core.Net('net')
        workspace.FeedBlob(
            "tensors", np.array([x for x in range(10)], np.int32)

            

Reported by Pylint.

                      ### Consumers ###
        outputs = []

        def append(ins, outs):
            # Extend is atomic
            outputs.extend(ins[0].data.tolist())

        consumer_steps = []
        for i in range(num_consumers):

            

Reported by Pylint.

                      # We check that the outputs are a permutation of inputs
        inputs.sort()
        outputs.sort()
        self.assertEquals(inputs, outputs)


if __name__ == "__main__":
    import unittest
    unittest.main()

            

Reported by Pylint.

              



from caffe2.python import core, workspace
from caffe2.python.test_util import TestCase

import numpy as np
import numpy.testing as npt

            

Reported by Pylint.

              from hypothesis import given, settings
import hypothesis.strategies as st

import functools


def primefac(n):
    ret = []
    divisor = 2

            

Reported by Pylint.

              import functools


def primefac(n):
    ret = []
    divisor = 2
    while divisor * divisor <= n:
        while (n % divisor) == 0:
            ret.append(divisor)

            

Reported by Pylint.

              import functools


def primefac(n):
    ret = []
    divisor = 2
    while divisor * divisor <= n:
        while (n % divisor) == 0:
            ret.append(divisor)

            

Reported by Pylint.

                  return ret


class TestReBatchingQueue(TestCase):
    def test_rebatching_queue_single_enqueue_dequeue(self):
        net = core.Net('net')

        tensors = [
            net.ConstantFill([], 1, value=1.0, run_once=False)

            

Reported by Pylint.

                      """
        try:
            faulthandler.enable(all_threads=True)
        except Exception as e:
            warnings.warn(f"Unable to enable fault handler. {type(e).__name__}: {e}")

    def _write_error_file(self, file_path: str, error_msg: str) -> None:
        """
        Writes error message to the file.

            

Reported by Pylint.

                      try:
            with open(file_path, "w") as fp:
                fp.write(error_msg)
        except Exception as e:
            warnings.warn(f"Unable to write error to file. {type(e).__name__}: {e}")

    def record_exception(self, e: BaseException) -> None:
        """
        Writes a structured information about the exception into an error file in

            

Reported by Pylint.

                          # is terminated by singals like SIGSEGV.
            if error_code:
                if "message" not in rootcause_error:
                    log.warning(
                        f"child error file ({rootcause_error_file}) does not have field `message`. \n"
                        f"cannot override error code: {error_code}"
                    )
                elif isinstance(rootcause_error["message"], str):
                    log.warning(

            

Reported by Pylint.

                                      f"cannot override error code: {error_code}"
                    )
                elif isinstance(rootcause_error["message"], str):
                    log.warning(
                        f"child error file ({rootcause_error_file}) has a new message format. \n"
                        f"skipping error code override"
                    )
                else:
                    rootcause_error["message"]["errorCode"] = error_code

            

Reported by Pylint.

                              else:
                    rootcause_error["message"]["errorCode"] = error_code

            log.info(
                f"child error file ({rootcause_error_file}) contents:\n"
                f"{json.dumps(rootcause_error, indent=2)}"
            )

        my_error_file = self._get_error_file_path()

            

Reported by Pylint.

                          # original error file contents and overwrite the error file.
            self._rm(my_error_file)
            self._write_error_file(my_error_file, json.dumps(rootcause_error))
            log.info(f"dumped error file to parent's {my_error_file}")
        else:
            log.error(
                f"no error file defined for parent, to copy child error file ({rootcause_error_file})"
            )


            

Reported by Pylint.

                          self._write_error_file(my_error_file, json.dumps(rootcause_error))
            log.info(f"dumped error file to parent's {my_error_file}")
        else:
            log.error(
                f"no error file defined for parent, to copy child error file ({rootcause_error_file})"
            )

    def _rm(self, my_error_file):
        if os.path.isfile(my_error_file):

            

Reported by Pylint.

                          with open(my_error_file, "r") as fp:
                try:
                    original = json.dumps(json.load(fp), indent=2)
                    log.warning(
                        f"{my_error_file} already exists"
                        f" and will be overwritten."
                        f" Original contents:\n{original}"
                    )
                except json.decoder.JSONDecodeError as err:

            

Reported by Pylint.

                                      f" and will be overwritten."
                        f" Original contents:\n{original}"
                    )
                except json.decoder.JSONDecodeError as err:
                    log.warning(
                        f"{my_error_file} already exists"
                        f" and will be overwritten."
                        f" Unable to load original contents:\n"
                    )

            

Reported by Pylint.

                                      f" Original contents:\n{original}"
                    )
                except json.decoder.JSONDecodeError as err:
                    log.warning(
                        f"{my_error_file} already exists"
                        f" and will be overwritten."
                        f" Unable to load original contents:\n"
                    )
            os.remove(my_error_file)

            

Reported by Pylint.

              from . import benchmark
import numpy as np


class MatMulBench(benchmark.Benchmark):
    def __init__(self, mode, device, dtype, B, M, N, K):
        super().__init__(mode, device, dtype)
        self.B = B
        self.M = M

            

Reported by Pylint.

              from . import benchmark
import numpy as np


class MatMulBench(benchmark.Benchmark):
    def __init__(self, mode, device, dtype, B, M, N, K):
        super().__init__(mode, device, dtype)
        self.B = B
        self.M = M

            

Reported by Pylint.

              from . import benchmark
import numpy as np


class MatMulBench(benchmark.Benchmark):
    def __init__(self, mode, device, dtype, B, M, N, K):
        super().__init__(mode, device, dtype)
        self.B = B
        self.M = M

            

Reported by Pylint.

              import numpy as np


class MatMulBench(benchmark.Benchmark):
    def __init__(self, mode, device, dtype, B, M, N, K):
        super().__init__(mode, device, dtype)
        self.B = B
        self.M = M
        self.N = N

            

Reported by Pylint.

              

class MatMulBench(benchmark.Benchmark):
    def __init__(self, mode, device, dtype, B, M, N, K):
        super().__init__(mode, device, dtype)
        self.B = B
        self.M = M
        self.N = N
        self.K = K

            

Reported by Pylint.

              

class MatMulBench(benchmark.Benchmark):
    def __init__(self, mode, device, dtype, B, M, N, K):
        super().__init__(mode, device, dtype)
        self.B = B
        self.M = M
        self.N = N
        self.K = K

            

Reported by Pylint.

              

class MatMulBench(benchmark.Benchmark):
    def __init__(self, mode, device, dtype, B, M, N, K):
        super().__init__(mode, device, dtype)
        self.B = B
        self.M = M
        self.N = N
        self.K = K

            

Reported by Pylint.

              

class MatMulBench(benchmark.Benchmark):
    def __init__(self, mode, device, dtype, B, M, N, K):
        super().__init__(mode, device, dtype)
        self.B = B
        self.M = M
        self.N = N
        self.K = K

            

Reported by Pylint.

              

class MatMulBench(benchmark.Benchmark):
    def __init__(self, mode, device, dtype, B, M, N, K):
        super().__init__(mode, device, dtype)
        self.B = B
        self.M = M
        self.N = N
        self.K = K

            

Reported by Pylint.

              class MatMulBench(benchmark.Benchmark):
    def __init__(self, mode, device, dtype, B, M, N, K):
        super().__init__(mode, device, dtype)
        self.B = B
        self.M = M
        self.N = N
        self.K = K
        self.d1 = self.rand([B, M, N], device=device, dtype=dtype, requires_grad=self.requires_grad)
        self.d2 = self.rand([B, N, K], device=device, dtype=dtype, requires_grad=self.requires_grad)

            

Reported by Pylint.

              from . import benchmark


class ConvImplBench(benchmark.Benchmark):
    def __init__(self, case, mode, device, dtype, kernel_size, N, iC, H, W, oC):
        super().__init__(mode, device, dtype)
        self.case = case
        self.kernel_size = kernel_size
        self.N = N

            

Reported by Pylint.

              from . import benchmark


class ConvImplBench(benchmark.Benchmark):
    def __init__(self, case, mode, device, dtype, kernel_size, N, iC, H, W, oC):
        super().__init__(mode, device, dtype)
        self.case = case
        self.kernel_size = kernel_size
        self.N = N

            

Reported by Pylint.

              from . import benchmark


class ConvImplBench(benchmark.Benchmark):
    def __init__(self, case, mode, device, dtype, kernel_size, N, iC, H, W, oC):
        super().__init__(mode, device, dtype)
        self.case = case
        self.kernel_size = kernel_size
        self.N = N

            

Reported by Pylint.

              from . import benchmark


class ConvImplBench(benchmark.Benchmark):
    def __init__(self, case, mode, device, dtype, kernel_size, N, iC, H, W, oC):
        super().__init__(mode, device, dtype)
        self.case = case
        self.kernel_size = kernel_size
        self.N = N

            

Reported by Pylint.

              

class ConvImplBench(benchmark.Benchmark):
    def __init__(self, case, mode, device, dtype, kernel_size, N, iC, H, W, oC):
        super().__init__(mode, device, dtype)
        self.case = case
        self.kernel_size = kernel_size
        self.N = N
        self.iC = iC

            

Reported by Pylint.

              

class ConvImplBench(benchmark.Benchmark):
    def __init__(self, case, mode, device, dtype, kernel_size, N, iC, H, W, oC):
        super().__init__(mode, device, dtype)
        self.case = case
        self.kernel_size = kernel_size
        self.N = N
        self.iC = iC

            

Reported by Pylint.

              

class ConvImplBench(benchmark.Benchmark):
    def __init__(self, case, mode, device, dtype, kernel_size, N, iC, H, W, oC):
        super().__init__(mode, device, dtype)
        self.case = case
        self.kernel_size = kernel_size
        self.N = N
        self.iC = iC

            

Reported by Pylint.

              

class ConvImplBench(benchmark.Benchmark):
    def __init__(self, case, mode, device, dtype, kernel_size, N, iC, H, W, oC):
        super().__init__(mode, device, dtype)
        self.case = case
        self.kernel_size = kernel_size
        self.N = N
        self.iC = iC

            

Reported by Pylint.

              

class ConvImplBench(benchmark.Benchmark):
    def __init__(self, case, mode, device, dtype, kernel_size, N, iC, H, W, oC):
        super().__init__(mode, device, dtype)
        self.case = case
        self.kernel_size = kernel_size
        self.N = N
        self.iC = iC

            

Reported by Pylint.

              

class ConvImplBench(benchmark.Benchmark):
    def __init__(self, case, mode, device, dtype, kernel_size, N, iC, H, W, oC):
        super().__init__(mode, device, dtype)
        self.case = case
        self.kernel_size = kernel_size
        self.N = N
        self.iC = iC

            

Reported by Pylint.

              
from torch.nn.modules.container import ModuleDict, ModuleList

from .parametrization import PruningParametrization, LinearActivationReconstruction, Conv2dActivationReconstruction

SUPPORTED_MODULES = {
    nn.Linear,
    nn.Conv2d
}

            

Reported by Pylint.

                              module = config['module']

            if getattr(module, 'mask', None) is None:
                module.register_buffer('mask', torch.tensor(module.weight.shape[0]))
            param = config.get('parametrization', PruningParametrization)
            parametrize.register_parametrization(module, 'weight',
                                                 param(module.mask),
                                                 unsafe=True)


            

Reported by Pylint.

                          stack = [model]
            while stack:
                module = stack.pop()
                for name, child in module.named_children():
                    if type(child) in SUPPORTED_MODULES:
                        self.config.append(child)
                    else:
                        stack.append(child)


            

Reported by Pylint.

                      format_string += ')'
        return format_string

    def bias_hook(self, module, input, output):
        if getattr(module, '_bias', None) is not None:
            idx = [1] * len(output.shape)
            idx[1] = output.shape[1]
            bias = module._bias.reshape(idx)
            output += bias

            

Reported by Pylint.

                      format_string += ')'
        return format_string

    def bias_hook(self, module, input, output):
        if getattr(module, '_bias', None) is not None:
            idx = [1] * len(output.shape)
            idx[1] = output.shape[1]
            bias = module._bias.reshape(idx)
            output += bias

            

Reported by Pylint.

                      if getattr(module, '_bias', None) is not None:
            idx = [1] * len(output.shape)
            idx[1] = output.shape[1]
            bias = module._bias.reshape(idx)
            output += bias
        return output

    def prepare(self, use_path=False, *args, **kwargs):
        r"""Adds mask parametrization to the layer weight

            

Reported by Pylint.

                          output += bias
        return output

    def prepare(self, use_path=False, *args, **kwargs):
        r"""Adds mask parametrization to the layer weight
        """
        for config in self.module_groups:
            if use_path:
                module = _path_to_module(self.model, config['path'])

            

Reported by Pylint.

                          output += bias
        return output

    def prepare(self, use_path=False, *args, **kwargs):
        r"""Adds mask parametrization to the layer weight
        """
        for config in self.module_groups:
            if use_path:
                module = _path_to_module(self.model, config['path'])

            

Reported by Pylint.

                          output += bias
        return output

    def prepare(self, use_path=False, *args, **kwargs):
        r"""Adds mask parametrization to the layer weight
        """
        for config in self.module_groups:
            if use_path:
                module = _path_to_module(self.model, config['path'])

            

Reported by Pylint.

                              module.bias = None
            self.bias_handles.append(module.register_forward_hook(self.bias_hook))

    def convert(self, use_path=False, *args, **kwargs):
        for config in self.module_groups:
            if use_path:
                module = _path_to_module(self.model, config['path'])
            else:
                module = config['module']

            

Reported by Pylint.

              """Gradient interface"""

import torch
from .modules.utils import _single, _pair, _triple
import warnings


def _grad_input_padding(grad_output, input_size, stride, padding, kernel_size, dilation=None):
    if dilation is None:

            

Reported by Pylint.

                  grad_input_padding = _grad_input_padding(grad_output, input_size, stride,
                                             padding, kernel_size, dilation)

    return torch.conv_transpose1d(
        grad_output, weight, None, stride, padding, grad_input_padding, groups,
        dilation)


def conv1d_weight(input, weight_size, grad_output, stride=1, padding=0, dilation=1, groups=1):

            

Reported by Pylint.

                  input = input.contiguous().view(1, input.shape[0] * input.shape[1],
                                    input.shape[2])

    grad_weight = torch.conv1d(input, grad_output, None, dilation, padding,
                               stride, in_channels * min_batch)

    grad_weight = grad_weight.contiguous().view(
        min_batch, grad_weight.shape[1] // min_batch, grad_weight.shape[2])


            

Reported by Pylint.

                  grad_input_padding = _grad_input_padding(grad_output, input_size, stride,
                                             padding, kernel_size, dilation)

    return torch.conv_transpose2d(
        grad_output, weight, None, stride, padding, grad_input_padding, groups,
        dilation)


def conv2d_weight(input, weight_size, grad_output, stride=1, padding=0, dilation=1, groups=1):

            

Reported by Pylint.

                  input = input.contiguous().view(1, input.shape[0] * input.shape[1],
                                    input.shape[2], input.shape[3])

    grad_weight = torch.conv2d(input, grad_output, None, dilation, padding,
                               stride, in_channels * min_batch)

    grad_weight = grad_weight.contiguous().view(
        min_batch, grad_weight.shape[1] // min_batch, grad_weight.shape[2],
        grad_weight.shape[3])

            

Reported by Pylint.

                  grad_input_padding = _grad_input_padding(grad_output, input_size, stride,
                                             padding, kernel_size, dilation)

    return torch.conv_transpose3d(
        grad_output, weight, None, stride, padding, grad_input_padding, groups,
        dilation)


def conv3d_weight(input, weight_size, grad_output, stride=1, padding=0, dilation=1, groups=1):

            

Reported by Pylint.

                                                  input.shape[2], input.shape[3],
                                    input.shape[4])

    grad_weight = torch.conv3d(input, grad_output, None, dilation, padding,
                               stride, in_channels * min_batch)

    grad_weight = grad_weight.contiguous().view(
        min_batch, grad_weight.shape[1] // min_batch, grad_weight.shape[2],
        grad_weight.shape[3], grad_weight.shape[4])

            

Reported by Pylint.

                      dilation)


def conv1d_weight(input, weight_size, grad_output, stride=1, padding=0, dilation=1, groups=1):
    r"""
    Computes the gradient of conv1d with respect to the weight of the convolution.

    Args:
        input: input tensor of shape (minibatch x in_channels x iW)

            

Reported by Pylint.

                      dilation)


def conv2d_weight(input, weight_size, grad_output, stride=1, padding=0, dilation=1, groups=1):
    r"""
    Computes the gradient of conv2d with respect to the weight of the convolution.

    Args:
        input: input tensor of shape (minibatch x in_channels x iH x iW)

            

Reported by Pylint.

                      dilation)


def conv3d_weight(input, weight_size, grad_output, stride=1, padding=0, dilation=1, groups=1):
    r"""
    Computes the gradient of conv3d with respect to the weight of the convolution.

    Args:
        input: input tensor of shape (minibatch x in_channels x iT x iH x iW)

            

Reported by Pylint.

              # pyre-fixme[9]: group has type `str`; used as `None`.
def configure(handler: MetricHandler, group: str = None):
    if group is None:
        global _default_metrics_handler
        # pyre-fixme[9]: _default_metrics_handler has type `NullMetricHandler`; used
        #  as `MetricHandler`.
        _default_metrics_handler = handler
    else:
        _metrics_map[group] = handler

            

Reported by Pylint.

              #!/usr/bin/env python3

# Copyright (c) Facebook, Inc. and its affiliates.
# All rights reserved.
#
# This source code is licensed under the BSD-style license found in the
# LICENSE file in the root directory of this source tree.

import abc

            

Reported by Pylint.

              MetricData = namedtuple("MetricData", ["timestamp", "group_name", "name", "value"])


class MetricsConfig:
    __slots__ = ["params"]

    def __init__(self, params: Optional[Dict[str, str]] = None):
        self.params = params
        if self.params is None:

            

Reported by Pylint.

              MetricData = namedtuple("MetricData", ["timestamp", "group_name", "name", "value"])


class MetricsConfig:
    __slots__ = ["params"]

    def __init__(self, params: Optional[Dict[str, str]] = None):
        self.params = params
        if self.params is None:

            

Reported by Pylint.

                          self.params = {}


class MetricHandler(abc.ABC):
    @abc.abstractmethod
    def emit(self, metric_data: MetricData):
        pass



            

Reported by Pylint.

                          self.params = {}


class MetricHandler(abc.ABC):
    @abc.abstractmethod
    def emit(self, metric_data: MetricData):
        pass



            

Reported by Pylint.

              
class MetricHandler(abc.ABC):
    @abc.abstractmethod
    def emit(self, metric_data: MetricData):
        pass


class ConsoleMetricHandler(MetricHandler):
    def emit(self, metric_data: MetricData):

            

Reported by Pylint.

                      pass


class ConsoleMetricHandler(MetricHandler):
    def emit(self, metric_data: MetricData):
        print(
            "[{}][{}]: {}={}".format(
                metric_data.timestamp,
                metric_data.group_name,

            

Reported by Pylint.

                      pass


class ConsoleMetricHandler(MetricHandler):
    def emit(self, metric_data: MetricData):
        print(
            "[{}][{}]: {}={}".format(
                metric_data.timestamp,
                metric_data.group_name,

            

Reported by Pylint.

                      )


class NullMetricHandler(MetricHandler):
    def emit(self, metric_data: MetricData):
        pass


class MetricStream:

            

Reported by Pylint.

              import torch
from torch.testing._internal.common_utils import TestCase, run_tests

class TestSetDefaultMobileCPUAllocator(TestCase):
    def test_no_exception(self):
        torch._C._set_default_mobile_cpu_allocator()
        torch._C._unset_default_mobile_cpu_allocator()

    def test_exception(self):

            

Reported by Pylint.

              import torch
from torch.testing._internal.common_utils import TestCase, run_tests

class TestSetDefaultMobileCPUAllocator(TestCase):
    def test_no_exception(self):
        torch._C._set_default_mobile_cpu_allocator()
        torch._C._unset_default_mobile_cpu_allocator()

    def test_exception(self):

            

Reported by Pylint.

              
class TestSetDefaultMobileCPUAllocator(TestCase):
    def test_no_exception(self):
        torch._C._set_default_mobile_cpu_allocator()
        torch._C._unset_default_mobile_cpu_allocator()

    def test_exception(self):
        with self.assertRaises(Exception):
            torch._C._unset_default_mobile_cpu_allocator()

            

Reported by Pylint.

              
class TestSetDefaultMobileCPUAllocator(TestCase):
    def test_no_exception(self):
        torch._C._set_default_mobile_cpu_allocator()
        torch._C._unset_default_mobile_cpu_allocator()

    def test_exception(self):
        with self.assertRaises(Exception):
            torch._C._unset_default_mobile_cpu_allocator()

            

Reported by Pylint.

              class TestSetDefaultMobileCPUAllocator(TestCase):
    def test_no_exception(self):
        torch._C._set_default_mobile_cpu_allocator()
        torch._C._unset_default_mobile_cpu_allocator()

    def test_exception(self):
        with self.assertRaises(Exception):
            torch._C._unset_default_mobile_cpu_allocator()


            

Reported by Pylint.

              class TestSetDefaultMobileCPUAllocator(TestCase):
    def test_no_exception(self):
        torch._C._set_default_mobile_cpu_allocator()
        torch._C._unset_default_mobile_cpu_allocator()

    def test_exception(self):
        with self.assertRaises(Exception):
            torch._C._unset_default_mobile_cpu_allocator()


            

Reported by Pylint.

              
    def test_exception(self):
        with self.assertRaises(Exception):
            torch._C._unset_default_mobile_cpu_allocator()

        with self.assertRaises(Exception):
            torch._C._set_default_mobile_cpu_allocator()
            torch._C._set_default_mobile_cpu_allocator()


            

Reported by Pylint.

              
    def test_exception(self):
        with self.assertRaises(Exception):
            torch._C._unset_default_mobile_cpu_allocator()

        with self.assertRaises(Exception):
            torch._C._set_default_mobile_cpu_allocator()
            torch._C._set_default_mobile_cpu_allocator()


            

Reported by Pylint.

                          torch._C._unset_default_mobile_cpu_allocator()

        with self.assertRaises(Exception):
            torch._C._set_default_mobile_cpu_allocator()
            torch._C._set_default_mobile_cpu_allocator()

        # Must reset to good state
        # For next test.
        torch._C._unset_default_mobile_cpu_allocator()

            

Reported by Pylint.

                          torch._C._unset_default_mobile_cpu_allocator()

        with self.assertRaises(Exception):
            torch._C._set_default_mobile_cpu_allocator()
            torch._C._set_default_mobile_cpu_allocator()

        # Must reset to good state
        # For next test.
        torch._C._unset_default_mobile_cpu_allocator()

            

Reported by Pylint.