from typing import cast, List, Optional, Tuple, TYPE_CHECKING
import uuid

import torch

if TYPE_CHECKING:
    # See the note in api.py for why this is necessary.
    from torch.utils.benchmark.utils.timer import Language
else:

            

Reported by Pylint.

              
if TYPE_CHECKING:
    # See the note in api.py for why this is necessary.
    from torch.utils.benchmark.utils.timer import Language
else:
    from torch.utils.benchmark import Language

from core.api import AutogradMode, AutoLabels, GroupedBenchmark, RuntimeMode, TimerArgs
from core.types import FlatDefinition, FlatIntermediateDefinition, Label

            

Reported by Pylint.

                  # See the note in api.py for why this is necessary.
    from torch.utils.benchmark.utils.timer import Language
else:
    from torch.utils.benchmark import Language

from core.api import AutogradMode, AutoLabels, GroupedBenchmark, RuntimeMode, TimerArgs
from core.types import FlatDefinition, FlatIntermediateDefinition, Label
from core.utils import get_temp_dir


            

Reported by Pylint.

              _ALL_MODES = tuple(it.product(
    RuntimeMode,
    AutogradMode,
    Language,
))


def _generate_torchscript_file(model_src: str, name: str) -> Optional[str]:
    """Returns the path a saved model if one can be constructed from `spec`.

            

Reported by Pylint.

                  else:
        assert runtime == RuntimeMode.JIT
        assert benchmark.signature_args is not None
        stmts = GroupedBenchmark._make_model_invocation(
            benchmark.signature_args, benchmark.signature_output, RuntimeMode.JIT)

    stmt = stmts[0 if is_python else 1]

    if autograd == AutogradMode.FORWARD_BACKWARD and stmt is not None:

            

Reported by Pylint.

              else:
    from torch.utils.benchmark import Language

from core.api import AutogradMode, AutoLabels, GroupedBenchmark, RuntimeMode, TimerArgs
from core.types import FlatDefinition, FlatIntermediateDefinition, Label
from core.utils import get_temp_dir


_ALL_MODES = tuple(it.product(

            

Reported by Pylint.

                  from torch.utils.benchmark import Language

from core.api import AutogradMode, AutoLabels, GroupedBenchmark, RuntimeMode, TimerArgs
from core.types import FlatDefinition, FlatIntermediateDefinition, Label
from core.utils import get_temp_dir


_ALL_MODES = tuple(it.product(
    RuntimeMode,

            

Reported by Pylint.

              
from core.api import AutogradMode, AutoLabels, GroupedBenchmark, RuntimeMode, TimerArgs
from core.types import FlatDefinition, FlatIntermediateDefinition, Label
from core.utils import get_temp_dir


_ALL_MODES = tuple(it.product(
    RuntimeMode,
    AutogradMode,

            

Reported by Pylint.

                  `model_src` must contain `jit_model = ...`, which `materialize` will supply.
    """
    # Double check.
    assert "jit_model = " in model_src, f"Missing jit_model definition:\n{model_src}"

    # `torch.utils.benchmark.Timer` will automatically import torch, so we
    # need to match that convention.
    model_src = f"import torch\n{model_src}"


            

Reported by Bandit.

                      # to confirm.
        raise ValueError(f"File {module_path} already exists.")

    with open(module_path, "wt") as f:
        f.write(model_src)

    # Import magic to actually load our function.
    module_spec = importlib.util.spec_from_file_location(f"torchscript__{name}", module_path)
    module = importlib.util.module_from_spec(module_spec)

            

Reported by Pylint.

              
    try:
        from caffe2.python.caffe2_pybind11_state_gpu import *  # noqa
        if num_cuda_devices():  # noqa
            has_gpu_support = has_cuda_support = True
    except ImportError as gpu_e:
        logging.info('Failed to import cuda module: {}'.format(gpu_e))
        try:
            from caffe2.python.caffe2_pybind11_state_hip import *  # noqa

            

Reported by Pylint.

              
# libcaffe2_python contains a global Workspace that we need to properly delete
# when exiting. Otherwise, cudart will cause segfaults sometimes.
atexit.register(on_module_exit)  # noqa


# Add functionalities for the TensorCPU interface.
def _TensorCPU_shape(self):
    return tuple(self._shape)

            

Reported by Pylint.

              def _TensorCPU_reshape(self, shape):
    return self._reshape(list(shape))

TensorCPU.shape = property(_TensorCPU_shape)  # noqa
TensorCPU.reshape = _TensorCPU_reshape  # noqa

            

Reported by Pylint.

                  return self._reshape(list(shape))

TensorCPU.shape = property(_TensorCPU_shape)  # noqa
TensorCPU.reshape = _TensorCPU_reshape  # noqa

            

Reported by Pylint.

                  has_gpu_support = False

    try:
        from caffe2.python.caffe2_pybind11_state_gpu import *  # noqa
        if num_cuda_devices():  # noqa
            has_gpu_support = has_cuda_support = True
    except ImportError as gpu_e:
        logging.info('Failed to import cuda module: {}'.format(gpu_e))
        try:

            

Reported by Pylint.

                      if num_cuda_devices():  # noqa
            has_gpu_support = has_cuda_support = True
    except ImportError as gpu_e:
        logging.info('Failed to import cuda module: {}'.format(gpu_e))
        try:
            from caffe2.python.caffe2_pybind11_state_hip import *  # noqa
            # we stop checking whether we have AMD GPU devices on the host,
            # because we may be constructing a net on a machine without GPU,
            # and run the net on another one with GPU

            

Reported by Pylint.

                  except ImportError as gpu_e:
        logging.info('Failed to import cuda module: {}'.format(gpu_e))
        try:
            from caffe2.python.caffe2_pybind11_state_hip import *  # noqa
            # we stop checking whether we have AMD GPU devices on the host,
            # because we may be constructing a net on a machine without GPU,
            # and run the net on another one with GPU
            has_gpu_support = has_hip_support = True
            logging.info('This caffe2 python run has AMD GPU support!')

            

Reported by Pylint.

                          has_gpu_support = has_hip_support = True
            logging.info('This caffe2 python run has AMD GPU support!')
        except ImportError as hip_e:
            logging.info('Failed to import AMD hip module: {}'.format(hip_e))

            logging.warning(
                'This caffe2 python run failed to load cuda module:{},'
                'and AMD hip module:{}.'
                'Will run in CPU only mode.'.format(gpu_e, hip_e))

            

Reported by Pylint.

                          logging.info('Failed to import AMD hip module: {}'.format(hip_e))

            logging.warning(
                'This caffe2 python run failed to load cuda module:{},'
                'and AMD hip module:{}.'
                'Will run in CPU only mode.'.format(gpu_e, hip_e))
            try:
                from caffe2.python.caffe2_pybind11_state import *  # noqa
            except ImportError as cpu_e:

            

Reported by Pylint.

                              'and AMD hip module:{}.'
                'Will run in CPU only mode.'.format(gpu_e, hip_e))
            try:
                from caffe2.python.caffe2_pybind11_state import *  # noqa
            except ImportError as cpu_e:
                logging.critical(
                    'Cannot load caffe2.python. Error: {0}'.format(str(cpu_e)))
                sys.exit(1)


            

Reported by Pylint.

              import torch
import torch.nn.quantized as nnq

import operator_benchmark as op_bench

r"""Microbenchmarks for the quantized activations."""

qactivation_long_configs = op_bench.cross_product_configs(
    dims=(

            

Reported by Pylint.

              import torch
import torch.nn.quantized as nnq

import operator_benchmark as op_bench

r"""Microbenchmarks for the quantized activations."""

qactivation_long_configs = op_bench.cross_product_configs(
    dims=(

            

Reported by Pylint.

              
r"""Microbenchmarks for the quantized activations."""

qactivation_long_configs = op_bench.cross_product_configs(
    dims=(
        # VGG-16 relu's with original shape: (-1, 3, 224, 224)
        ( 64, 224, 224),  # ReLU-1   # noqa: E201
        (128, 112, 112),  # ReLU-6
        (256,  56,  56),  # ReLU-11  # noqa: E241

            

Reported by Pylint.

                  tags=('long',)
)

qactivation_short_configs = op_bench.cross_product_configs(
    dims=(
        (3, 4, 5),      # Rank=3
        (2, 3, 4, 5),    # Rank=4,
        # Dimensions from the floating point benchmarks
        (512, 512),

            

Reported by Pylint.

                  tags=('short',)
)

qactivation_ops = op_bench.op_list(
    attrs=(
        ('relu', torch.nn.ReLU()),
        ('relu6', torch.ops.quantized.relu6),
        ('functional.hardtanh', nnq.functional.hardtanh),
        ('functional.hardsigmoid', nnq.functional.hardsigmoid),

            

Reported by Pylint.

              )


class QActivationBenchmarkBase(op_bench.TorchBenchmarkBase):
    r"""Base class for all the activations."""
    def _setup(self, dims, contig, dtype):
        # Input
        f_input = (torch.rand(*dims) - 0.5) * 256
        self.scale = 1.0

            

Reported by Pylint.

                      return self.qop(q_input)


op_bench.generate_pt_tests_from_op_list(qactivation_ops,
                                        qactivation_short_configs + qactivation_long_configs,
                                        QActivationBenchmark)


qactivation_scale_zero_point_ops = op_bench.op_list(

            

Reported by Pylint.

                                                      QActivationBenchmark)


qactivation_scale_zero_point_ops = op_bench.op_list(
    attrs=(
        ('functional.hardswish', nnq.functional.hardswish),
        ('functional.elu', nnq.functional.elu),
        ('functional.celu', nnq.functional.celu),
    ),

            

Reported by Pylint.

                  def forward(self, q_input):
        return self.qop(q_input, scale=self.scale, zero_point=self.zero_point)

op_bench.generate_pt_tests_from_op_list(qactivation_scale_zero_point_ops,
                                        qactivation_short_configs + qactivation_long_configs,
                                        QActivationScaleZeroPointBenchmark)

if __name__ == "__main__":
    op_bench.benchmark_runner.main()

            

Reported by Pylint.

              
import operator_benchmark as op_bench

r"""Microbenchmarks for the quantized activations."""

qactivation_long_configs = op_bench.cross_product_configs(
    dims=(
        # VGG-16 relu's with original shape: (-1, 3, 224, 224)
        ( 64, 224, 224),  # ReLU-1   # noqa: E201

            

Reported by Pylint.

              from torch.autograd import Variable, Function
import torch.onnx

import onnx
import caffe2.python.onnx.backend

class MyFunction(Function):
    @staticmethod
    def forward(ctx, x, y):

            

Reported by Pylint.

              
f = tempfile.NamedTemporaryFile()
torch.onnx.export(MyModule(),
                  (Variable(torch.ones(3, 4)), Variable(torch.ones(3, 4))),
                  f, verbose=True)

# prints the graph for debugging:
# graph(%input : Float(3, 4, strides=[4, 1], requires_grad=0, device=cpu),
#       %y : Float(3, 4, strides=[4, 1], requires_grad=0, device=cpu)):

            

Reported by Pylint.

              
f = tempfile.NamedTemporaryFile()
torch.onnx.export(MyModule(),
                  (Variable(torch.ones(3, 4)), Variable(torch.ones(3, 4))),
                  f, verbose=True)

# prints the graph for debugging:
# graph(%input : Float(3, 4, strides=[4, 1], requires_grad=0, device=cpu),
#       %y : Float(3, 4, strides=[4, 1], requires_grad=0, device=cpu)):

            

Reported by Pylint.

              import onnx
import caffe2.python.onnx.backend

class MyFunction(Function):
    @staticmethod
    def forward(ctx, x, y):
        return x * x + y

    @staticmethod

            

Reported by Pylint.

              
class MyFunction(Function):
    @staticmethod
    def forward(ctx, x, y):
        return x * x + y

    @staticmethod
    def symbolic(graph, x, y):
        x2 = graph.at("mul", x, x)

            

Reported by Pylint.

              
class MyFunction(Function):
    @staticmethod
    def forward(ctx, x, y):
        return x * x + y

    @staticmethod
    def symbolic(graph, x, y):
        x2 = graph.at("mul", x, x)

            

Reported by Pylint.

                      return x * x + y

    @staticmethod
    def symbolic(graph, x, y):
        x2 = graph.at("mul", x, x)
        r = graph.at("add", x2, y)
        # x, y, x2, and r are 'Node' objects
        # print(r) or print(graph) will print out a textual representation for debugging.
        # this representation will be converted to ONNX protobufs on export.

            

Reported by Pylint.

                      return x * x + y

    @staticmethod
    def symbolic(graph, x, y):
        x2 = graph.at("mul", x, x)
        r = graph.at("add", x2, y)
        # x, y, x2, and r are 'Node' objects
        # print(r) or print(graph) will print out a textual representation for debugging.
        # this representation will be converted to ONNX protobufs on export.

            

Reported by Pylint.

                  @staticmethod
    def symbolic(graph, x, y):
        x2 = graph.at("mul", x, x)
        r = graph.at("add", x2, y)
        # x, y, x2, and r are 'Node' objects
        # print(r) or print(graph) will print out a textual representation for debugging.
        # this representation will be converted to ONNX protobufs on export.
        return r


            

Reported by Pylint.

                      return r

class MyModule(nn.Module):
    def forward(self, x, y):
        # you can combine your ATen ops with standard onnx ones
        x = nn.ReLU()(x)
        return MyFunction.apply(x, y)

f = tempfile.NamedTemporaryFile()

            

Reported by Pylint.

                      """
        best_scope = candidate_scope
        best_scope_idx = 0
        sub_scopes = candidate_scope.split(scope._NAMESCOPE_SEPARATOR)

        cur_scope = ''
        for idx, sub_scope in enumerate(sub_scopes):
            cur_scope = cur_scope + sub_scope + scope._NAMESCOPE_SEPARATOR
            if cur_scope in self._scope_overrides:

            

Reported by Pylint.

              
        cur_scope = ''
        for idx, sub_scope in enumerate(sub_scopes):
            cur_scope = cur_scope + sub_scope + scope._NAMESCOPE_SEPARATOR
            if cur_scope in self._scope_overrides:
                best_scope = self._scope_overrides[cur_scope]
                best_scope_idx = idx
        if best_scope == candidate_scope:
            return candidate_scope

            

Reported by Pylint.

                          return candidate_scope
        else:
            return (self._resolve_scope_overrides(best_scope) +
                    scope._NAMESCOPE_SEPARATOR.join(
                        sub_scopes[best_scope_idx + 1:]))

    def get_parameter_name(self, name):
        candidate_scope = scope.CurrentNameScope()
        best_scope = self._resolve_scope_overrides(candidate_scope)

            

Reported by Pylint.

                      candidate_scope = scope.CurrentNameScope()
        best_scope = self._resolve_scope_overrides(candidate_scope)
        if best_scope != candidate_scope:
            logger.info("Overwriting scope {0} with scope {1}".format(
                candidate_scope, best_scope))

        return best_scope + name

    def add_scope_overrides(self, shared_scopes):

            

Reported by Pylint.

              

def _normalize_namescope(namescope):
    if namescope and namescope[-1] != scope._NAMESCOPE_SEPARATOR:
        return namescope + scope._NAMESCOPE_SEPARATOR
    else:
        return namescope



            

Reported by Pylint.

              
def _normalize_namescope(namescope):
    if namescope and namescope[-1] != scope._NAMESCOPE_SEPARATOR:
        return namescope + scope._NAMESCOPE_SEPARATOR
    else:
        return namescope


@contextlib.contextmanager

            

Reported by Pylint.

              




from caffe2.python import scope

import contextlib
import logging

            

Reported by Pylint.

              
from caffe2.python import scope

import contextlib
import logging

logger = logging.getLogger(__name__)



            

Reported by Pylint.

              from caffe2.python import scope

import contextlib
import logging

logger = logging.getLogger(__name__)


class ParameterSharingContext(object):

            

Reported by Pylint.

              logger = logging.getLogger(__name__)


class ParameterSharingContext(object):
    """
    This class manages scope driven way of parameter sharing across different
    NameScopes.
    """


            

Reported by Pylint.

                  }

    try (InputStream is = context.getAssets().open(assetName)) {
      try (OutputStream os = new FileOutputStream(file)) {
        byte[] buffer = new byte[4 * 1024];
        int read;
        while ((read = is.read(buffer)) != -1) {
          os.write(buffer, 0, read);
        }

            

Reported by PMD.

                private static final String TAG = BuildConfig.LOGCAT_TAG;
  private static final int TEXT_TRIM_SIZE = 4096;

  private TextView mTextView;

  protected HandlerThread mBackgroundThread;
  protected Handler mBackgroundHandler;
  private Module mModule;
  private FloatBuffer mInputTensorBuffer;

            

Reported by PMD.

              
  private TextView mTextView;

  protected HandlerThread mBackgroundThread;
  protected Handler mBackgroundHandler;
  private Module mModule;
  private FloatBuffer mInputTensorBuffer;
  private Tensor mInputTensor;
  private StringBuilder mTextViewStringBuilder = new StringBuilder();

            

Reported by PMD.

                private TextView mTextView;

  protected HandlerThread mBackgroundThread;
  protected Handler mBackgroundHandler;
  private Module mModule;
  private FloatBuffer mInputTensorBuffer;
  private Tensor mInputTensor;
  private StringBuilder mTextViewStringBuilder = new StringBuilder();


            

Reported by PMD.

              
  protected HandlerThread mBackgroundThread;
  protected Handler mBackgroundHandler;
  private Module mModule;
  private FloatBuffer mInputTensorBuffer;
  private Tensor mInputTensor;
  private StringBuilder mTextViewStringBuilder = new StringBuilder();

  private final Runnable mModuleForwardRunnable =

            

Reported by PMD.

                protected HandlerThread mBackgroundThread;
  protected Handler mBackgroundHandler;
  private Module mModule;
  private FloatBuffer mInputTensorBuffer;
  private Tensor mInputTensor;
  private StringBuilder mTextViewStringBuilder = new StringBuilder();

  private final Runnable mModuleForwardRunnable =
      new Runnable() {

            

Reported by PMD.

                protected Handler mBackgroundHandler;
  private Module mModule;
  private FloatBuffer mInputTensorBuffer;
  private Tensor mInputTensor;
  private StringBuilder mTextViewStringBuilder = new StringBuilder();

  private final Runnable mModuleForwardRunnable =
      new Runnable() {
        @Override

            

Reported by PMD.

                private Module mModule;
  private FloatBuffer mInputTensorBuffer;
  private Tensor mInputTensor;
  private StringBuilder mTextViewStringBuilder = new StringBuilder();

  private final Runnable mModuleForwardRunnable =
      new Runnable() {
        @Override
        public void run() {

            

Reported by PMD.

                private Module mModule;
  private FloatBuffer mInputTensorBuffer;
  private Tensor mInputTensor;
  private StringBuilder mTextViewStringBuilder = new StringBuilder();

  private final Runnable mModuleForwardRunnable =
      new Runnable() {
        @Override
        public void run() {

            

Reported by PMD.

                private Module mModule;
  private FloatBuffer mInputTensorBuffer;
  private Tensor mInputTensor;
  private StringBuilder mTextViewStringBuilder = new StringBuilder();

  private final Runnable mModuleForwardRunnable =
      new Runnable() {
        @Override
        public void run() {

            

Reported by PMD.

              from caffe2.python import workspace


def OnGPU(gpu_id):
    """A utility function that returns a device option protobuf of the
  specified gpu id.
  """
    device_option = caffe2_pb2.DeviceOption()
    device_option.device_type = workspace.GpuDeviceType

            

Reported by Pylint.

                  return device_option


def OnCPU():
    device_option = caffe2_pb2.DeviceOption()
    device_option.device_type = caffe2_pb2.CPU
    return device_option



            

Reported by Pylint.

                  return device_option


def OnCPU():
    device_option = caffe2_pb2.DeviceOption()
    device_option.device_type = caffe2_pb2.CPU
    return device_option



            

Reported by Pylint.

                  return device_option


def Allreduce(net, blobs, reduced_affix="_reduced", gpu_indices=None):
    """The general Allreduce interface that reroutes the function calls.
    CPUs and AMD GPUs are not supported because
    GetGpuPeerAccessPattern is called to get gpu peer access pattern.
  """
    if gpu_indices is None:

            

Reported by Pylint.

                          (len(gpu_indices), len(blobs))
        )
    pattern = workspace.GetGpuPeerAccessPattern()
    if len(blobs) == 2 and pattern.shape[0] >= 2 and np.all(pattern[:2, :2]):
        return Allreduce2(net, blobs, reduced_affix, gpu_indices)
    elif len(blobs) == 4 and pattern.shape[0] >= 4 and np.all(pattern[:4, :4]):
        return Allreduce4(net, blobs, reduced_affix, gpu_indices)
    elif len(blobs) == 4 and pattern.shape[0] >= 4 and np.all(pattern[:2, :2]) and np.all(pattern[2:4, 2:4]):
        return Allreduce4Group2(net, blobs, reduced_affix, gpu_indices)

            

Reported by Pylint.

                      return Allreduce2(net, blobs, reduced_affix, gpu_indices)
    elif len(blobs) == 4 and pattern.shape[0] >= 4 and np.all(pattern[:4, :4]):
        return Allreduce4(net, blobs, reduced_affix, gpu_indices)
    elif len(blobs) == 4 and pattern.shape[0] >= 4 and np.all(pattern[:2, :2]) and np.all(pattern[2:4, 2:4]):
        return Allreduce4Group2(net, blobs, reduced_affix, gpu_indices)
    elif len(blobs) == 8 and pattern.shape[0] >= 8 and np.all(pattern[:8, :8]):
        return Allreduce8(net, blobs, reduced_affix, gpu_indices)
    else:
        return AllreduceFallback(net, blobs, reduced_affix, gpu_indices)

            

Reported by Pylint.

                      return AllreduceFallback(net, blobs, reduced_affix, gpu_indices)


def Allreduce2(net, blobs, reduced_affix, gpu_indices):
    """Allreduce for 2 gpus.

  Algorithm: 0r <- 0 + 1, 1r <- 0r, where r means "reduced"
  """
    a, b = blobs

            

Reported by Pylint.

              
  Algorithm: 0r <- 0 + 1, 1r <- 0r, where r means "reduced"
  """
    a, b = blobs
    gpu_a, gpu_b = gpu_indices
    a_reduced = net.Add([a, b], a + reduced_affix, device_option=OnGPU(gpu_a))
    b_reduced = a_reduced.Copy(
        [],
        b + reduced_affix,

            

Reported by Pylint.

              
  Algorithm: 0r <- 0 + 1, 1r <- 0r, where r means "reduced"
  """
    a, b = blobs
    gpu_a, gpu_b = gpu_indices
    a_reduced = net.Add([a, b], a + reduced_affix, device_option=OnGPU(gpu_a))
    b_reduced = a_reduced.Copy(
        [],
        b + reduced_affix,

            

Reported by Pylint.

                  return a_reduced, b_reduced


def Allreduce4(net, blobs, reduced_affix, gpu_indices):
    """Allreduce for 4 gpus.

  Algorithm: 2 level reduction.
      0r <- 0 + 1, 2r <- 2 + 3
      0r <- 0r + 2r

            

Reported by Pylint.

                      print("Failed to parse the response. Check if the Azure DevOps PAT is incorrect or expired.")
        sys.exit(-1)

    build_id = run_build_json['id']

    print("Submitted bulid: " + str(build_id))
    print("Bulid URL: " + run_build_json['url'])
    return build_id


            

Reported by Pylint.

              
def wait_for_build(_id):
    build_detail = get_build(_id)
    build_status = build_detail['status']

    while build_status == 'notStarted':
        print('Waiting for run to start: ' + str(_id))
        sys.stdout.flush()
        try:

            

Reported by Pylint.

                      try:
            build_detail = get_build(_id)
            build_status = build_detail['status']
        except Exception as e:
            print("Error getting build")
            print(e)

        time.sleep(30)


            

Reported by Pylint.

                      try:
            print("Waiting for log: " + str(_id))
            logs = get_build_logs(_id)
        except Exception as e:
            print("Error fetching logs")
            print(e)
            time.sleep(30)
            continue


            

Reported by Pylint.

                          try:
                log_content = get_log_content(log['url'])
                print(log_content)
            except Exception as e:
                print("Error getting log content")
                print(e)
            sys.stdout.flush()
        build_detail = get_build(_id)
        build_status = build_detail['status']

            

Reported by Pylint.

                      build_status = build_detail['status']
        time.sleep(30)

    build_result = build_detail['result']

    print("Bulid status: " + build_status)
    print("Bulid result: " + build_result)

    return build_status, build_result

            

Reported by Pylint.

              # Documentation: https://docs.microsoft.com/en-us/rest/api/azure/devops/build/?view=azure-devops-rest-6.0

import re
import json
import os
import sys
import requests
import time


            

Reported by Pylint.

              # Documentation: https://docs.microsoft.com/en-us/rest/api/azure/devops/build/?view=azure-devops-rest-6.0

import re
import json
import os
import sys
import requests
import time


            

Reported by Pylint.

              import os
import sys
import requests
import time

AZURE_PIPELINE_BASE_URL = "https://aiinfra.visualstudio.com/PyTorch/"
AZURE_DEVOPS_PAT_BASE64 = os.environ.get("AZURE_DEVOPS_PAT_BASE64_SECRET", "")
PIPELINE_ID = "911"
PROJECT_ID = "0628bce4-2d33-499e-bac5-530e12db160f"

            

Reported by Pylint.

              TARGET_BRANCH = os.environ.get("CIRCLE_BRANCH", "master")
TARGET_COMMIT = os.environ.get("CIRCLE_SHA1", "")

build_base_url = AZURE_PIPELINE_BASE_URL + "_apis/build/builds?api-version=6.0"

s = requests.Session()
s.headers.update({"Authorization": "Basic " + AZURE_DEVOPS_PAT_BASE64})

def submit_build(pipeline_id, project_id, source_branch, source_version):

            

Reported by Pylint.

              
import operator_benchmark as op_bench
import torch
import numpy as np

embeddingbag_conversion_short_configs = op_bench.cross_product_configs(
    num_embeddings=(80,),
    embedding_dim=(128, 256, 512),
    tags=('short',)

            

Reported by Pylint.

              import torch
import numpy as np

embeddingbag_conversion_short_configs = op_bench.cross_product_configs(
    num_embeddings=(80,),
    embedding_dim=(128, 256, 512),
    tags=('short',)
)


            

Reported by Pylint.

                  tags=('short',)
)

embeddingbag_conversion_long_configs = op_bench.cross_product_configs(
    num_embeddings=(100, 120, 1000),
    embedding_dim=(16, 64, 128, 256, 512, 1024, 2048),
    tags=('long',)
)


            

Reported by Pylint.

                  tags=('long',)
)

conversion_ops = op_bench.op_list(
    attrs=(
        ('qembeddingbag_byte_prepack', torch.ops.quantized.embedding_bag_byte_prepack),
        ('qembeddingbag_4bit_prepack', torch.ops.quantized.embedding_bag_4bit_prepack),
        ('qembeddingbag_2bit_prepack', torch.ops.quantized.embedding_bag_2bit_prepack),
    ),

            

Reported by Pylint.

                  attr_names=('op_name', 'op_func'),
)

unpack_ops = op_bench.op_list(
    attrs=(
        ('qembeddingbag_byte_unpack', torch.ops.quantized.embedding_bag_byte_unpack),
        ('qembeddingbag_4bit_unpack', torch.ops.quantized.embedding_bag_4bit_unpack),
        ('qembeddingbag_2bit_unpack', torch.ops.quantized.embedding_bag_2bit_unpack),
    ),

            

Reported by Pylint.

                  attr_names=('op_name', 'op_func'),
)

class EmbeddingBagFloatToFusedBase(op_bench.TorchBenchmarkBase):
    def init(self, num_embeddings, embedding_dim, op_func):
        self.inputs = {
            "weight": torch.from_numpy((np.random.random_sample((
                num_embeddings, embedding_dim)) + 1).astype(np.float32))
        }

            

Reported by Pylint.

                  def forward(self, weight):
        return self.op_func(weight)

class EmbeddingBagFusedToFloatBase(op_bench.TorchBenchmarkBase):
    def init(self, num_embeddings, embedding_dim, op_func):
        weight = torch.randn(num_embeddings, embedding_dim + 8, dtype=torch.float)
        self.inputs = {
            "packed_weight": weight.to(torch.uint8)
        }

            

Reported by Pylint.

                      return self.op_func(packed_weight)


op_bench.generate_pt_tests_from_op_list(conversion_ops,
                                        embeddingbag_conversion_short_configs + embeddingbag_conversion_long_configs,
                                        EmbeddingBagFloatToFusedBase)
op_bench.generate_pt_tests_from_op_list(unpack_ops,
                                        embeddingbag_conversion_short_configs + embeddingbag_conversion_long_configs,
                                        EmbeddingBagFusedToFloatBase)

            

Reported by Pylint.

              op_bench.generate_pt_tests_from_op_list(conversion_ops,
                                        embeddingbag_conversion_short_configs + embeddingbag_conversion_long_configs,
                                        EmbeddingBagFloatToFusedBase)
op_bench.generate_pt_tests_from_op_list(unpack_ops,
                                        embeddingbag_conversion_short_configs + embeddingbag_conversion_long_configs,
                                        EmbeddingBagFusedToFloatBase)

if __name__ == "__main__":
    op_bench.benchmark_runner.main()

            

Reported by Pylint.

              
class EmbeddingBagFloatToFusedBase(op_bench.TorchBenchmarkBase):
    def init(self, num_embeddings, embedding_dim, op_func):
        self.inputs = {
            "weight": torch.from_numpy((np.random.random_sample((
                num_embeddings, embedding_dim)) + 1).astype(np.float32))
        }
        self.op_func = op_func


            

Reported by Pylint.

              
from caffe2.python import brew

"""
Utilitiy for creating ShuffleNet
"ShuffleNet V2: Practical Guidelines for EfficientCNN Architecture Design" by Ma et. al. 2018
"""

OUTPUT_CHANNELS = {

            

Reported by Pylint.

                              )
                self.add_basic_unit(prefix, in_channels)

        self.last_conv = brew.conv(self.model, self.prev_blob, 'conv5',
                                   in_channels, self.output_channels[4],
                                   kernel=1)
        self.avg_pool = self.model.AveragePool(self.last_conv, 'avg_pool',
                                               kernel=7)
        self.last_out = brew.fc(self.model,

            

Reported by Pylint.

                      self.last_conv = brew.conv(self.model, self.prev_blob, 'conv5',
                                   in_channels, self.output_channels[4],
                                   kernel=1)
        self.avg_pool = self.model.AveragePool(self.last_conv, 'avg_pool',
                                               kernel=7)
        self.last_out = brew.fc(self.model,
                                self.avg_pool,
                                'last_out_L{}'.format(self.num_labels),
                                self.output_channels[4],

            

Reported by Pylint.

                                                 kernel=1)
        self.avg_pool = self.model.AveragePool(self.last_conv, 'avg_pool',
                                               kernel=7)
        self.last_out = brew.fc(self.model,
                                self.avg_pool,
                                'last_out_L{}'.format(self.num_labels),
                                self.output_channels[4],
                                self.num_labels)


            

Reported by Pylint.

              # Module caffe2.python.models.shufflenet






from caffe2.python import brew


            

Reported by Pylint.

              }


class ShuffleNetV2Builder():
    def __init__(
        self,
        model,
        data,
        num_input_channels,

            

Reported by Pylint.

              }


class ShuffleNetV2Builder():
    def __init__(
        self,
        model,
        data,
        num_input_channels,

            

Reported by Pylint.

              

class ShuffleNetV2Builder():
    def __init__(
        self,
        model,
        data,
        num_input_channels,
        num_labels,

            

Reported by Pylint.

                      self.detection = detection
        self.bn_epsilon = bn_epsilon

    def create(self):
        in_channels = self.output_channels[0]

        self.prev_blob = brew.conv(self.model, self.prev_blob, 'stage1_conv',
                                   self.num_input_channels, in_channels,
                                   weight_init=("MSRAFill", {}),

            

Reported by Pylint.

                                              self.num_labels)

    # spatial down sampling unit with stride=2
    def add_spatial_ds_unit(self, prefix, in_channels, out_channels, stride=2):
        right = left = self.prev_blob
        out_channels = out_channels // 2

        # Enlarge the receptive field for detection task
        if self.detection:

            

Reported by Pylint.