def _setup_logger(name: Optional[str] = None):
    log = logging.getLogger(name)
    log.setLevel(os.environ.get("LOGLEVEL", get_log_level()))
    return log


def _derive_module_name(depth: int = 1) -> Optional[str]:

            

Reported by Pylint.

              # LICENSE file in the root directory of this source tree.

import inspect
import logging
import os
import warnings
from typing import Optional

from torch.distributed.elastic.utils.log_level import get_log_level

            

Reported by Pylint.

                          filename = frame_info[1]
            module_name = os.path.splitext(os.path.basename(filename))[0]
        return module_name
    except Exception as e:
        warnings.warn(
            f"Error deriving logger module name, using <None>. Exception: {e}",
            RuntimeWarning,
        )
        return None

            

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 inspect

            

Reported by Pylint.

                  """
    try:
        stack = inspect.stack()
        assert depth < len(stack)
        # FrameInfo is just a named tuple: (frame, filename, lineno, function, code_context, index)
        frame_info = stack[depth]

        module = inspect.getmodule(frame_info[0])
        if module:

            

Reported by Bandit.

                          filename = frame_info[1]
            module_name = os.path.splitext(os.path.basename(filename))[0]
        return module_name
    except Exception as e:
        warnings.warn(
            f"Error deriving logger module name, using <None>. Exception: {e}",
            RuntimeWarning,
        )
        return None

            

Reported by Pylint.

              from .register_dispatch_key import RegisterDispatchKey as RegisterDispatchKey
from .native_functions import compute_native_function_declaration as compute_native_function_declaration

            

Reported by Pylint.

              from .register_dispatch_key import RegisterDispatchKey as RegisterDispatchKey
from .native_functions import compute_native_function_declaration as compute_native_function_declaration

            

Reported by Pylint.

              from .register_dispatch_key import RegisterDispatchKey as RegisterDispatchKey
from .native_functions import compute_native_function_declaration as compute_native_function_declaration

            

Reported by Pylint.

              from .register_dispatch_key import RegisterDispatchKey as RegisterDispatchKey
from .native_functions import compute_native_function_declaration as compute_native_function_declaration

            

Reported by Pylint.

              from .register_dispatch_key import RegisterDispatchKey as RegisterDispatchKey
from .native_functions import compute_native_function_declaration as compute_native_function_declaration

            

Reported by Pylint.

              from .register_dispatch_key import RegisterDispatchKey as RegisterDispatchKey
from .native_functions import compute_native_function_declaration as compute_native_function_declaration

            

Reported by Pylint.

              
    def __iter__(self):
        # deterministically shuffle based on epoch
        g = torch.Generator()
        g.manual_seed(self.epoch)
        indices = (
            torch.randperm(len(self.dataset) - self.start_index, generator=g)
            .add(self.start_index)
            .tolist()

            

Reported by Pylint.

                      g = torch.Generator()
        g.manual_seed(self.epoch)
        indices = (
            torch.randperm(len(self.dataset) - self.start_index, generator=g)
            .add(self.start_index)
            .tolist()
        )

        # add extra samples to make it evenly divisible

            

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 math

            

Reported by Pylint.

              
    def __iter__(self):
        # deterministically shuffle based on epoch
        g = torch.Generator()
        g.manual_seed(self.epoch)
        indices = (
            torch.randperm(len(self.dataset) - self.start_index, generator=g)
            .add(self.start_index)
            .tolist()

            

Reported by Pylint.

              
        # add extra samples to make it evenly divisible
        indices += indices[: (self.total_size - len(indices))]
        assert len(indices) == self.total_size

        # subsample
        indices = indices[self.rank : self.total_size : self.num_replicas]
        assert len(indices) == self.num_samples


            

Reported by Bandit.

              
        # subsample
        indices = indices[self.rank : self.total_size : self.num_replicas]
        assert len(indices) == self.num_samples

        return iter(indices)

    def __len__(self):
        return self.num_samples

            

Reported by Bandit.

              
        error_file = os.path.join(clogdir, "error.json")
        error_files[local_rank] = error_file
        log.info(f"Setting worker{local_rank} reply file to: {error_file}")
        envs[local_rank]["TORCHELASTIC_ERROR_FILE"] = error_file

    context: PContext
    if isinstance(entrypoint, str):
        context = SubprocessContext(

            

Reported by Pylint.

              log = get_logger()


def start_processes(
    name: str,
    entrypoint: Union[Callable, str],
    args: Dict[int, Tuple],
    envs: Dict[int, Dict[str, str]],
    log_dir: str,

            

Reported by Pylint.

              log = get_logger()


def start_processes(
    name: str,
    entrypoint: Union[Callable, str],
    args: Dict[int, Tuple],
    envs: Dict[int, Dict[str, str]],
    log_dir: str,

            

Reported by Pylint.

                  #       |- ...
    #       |- (nprocs-1)
    redirs = to_map(redirects, nprocs)
    ts = to_map(tee, nprocs)

    # to tee stdout/stderr we first redirect into a file
    # then tail -f stdout.log/stderr.log so add tee settings to redirects
    for local_rank, tee_std in ts.items():
        redirect_std = redirs[local_rank]

            

Reported by Pylint.

                      clogdir = os.path.join(log_dir, str(local_rank))
        os.mkdir(clogdir)

        rd = redirs[local_rank]
        if (rd & Std.OUT) == Std.OUT:
            stdouts[local_rank] = os.path.join(clogdir, "stdout.log")
        if (rd & Std.ERR) == Std.ERR:
            stderrs[local_rank] = os.path.join(clogdir, "stderr.log")


            

Reported by Pylint.

                      if (rd & Std.ERR) == Std.ERR:
            stderrs[local_rank] = os.path.join(clogdir, "stderr.log")

        t = ts[local_rank]
        if t & Std.OUT == Std.OUT:
            tee_stdouts[local_rank] = stdouts[local_rank]
        if t & Std.ERR == Std.ERR:
            tee_stderrs[local_rank] = stderrs[local_rank]


            

Reported by Pylint.

              
from typing import Optional

from .api import (  # noqa: F401
    ConsoleMetricHandler,
    MetricData,
    MetricHandler,
    MetricsConfig,
    NullMetricHandler,

            

Reported by Pylint.

              

try:
    from torch.distributed.elastic.metrics.static_init import *  # type: ignore[import] # noqa: F401 F403
except ModuleNotFoundError:
    pass

            

Reported by Pylint.

              

try:
    from torch.distributed.elastic.metrics.static_init import *  # type: ignore[import] # noqa: F401 F403
except ModuleNotFoundError:
    pass

            

Reported by Pylint.

              )


def initialize_metrics(cfg: Optional[MetricsConfig] = None):
    pass


try:
    from torch.distributed.elastic.metrics.static_init import *  # type: ignore[import] # noqa: F401 F403

            

Reported by Pylint.

              )


def initialize_metrics(cfg: Optional[MetricsConfig] = None):
    pass


try:
    from torch.distributed.elastic.metrics.static_init import *  # type: ignore[import] # noqa: F401 F403

            

Reported by Pylint.

              

try:
    from torch.distributed.elastic.metrics.static_init import *  # type: ignore[import] # noqa: F401 F403
except ModuleNotFoundError:
    pass

            

Reported by Pylint.

                      return 0;
      }
      size_t nread = std::min(static_cast<size_t>(pos) + n, size_) - pos;
      memcpy(buf, data_ + pos, nread);
      return nread;
    }
  private:
    const char* data_;
    size_t size_;

            

Reported by FlawFinder.

              
InputArchive::InputArchive() : module_("Module", std::make_shared<jit::CompilationUnit>()) {}

void InputArchive::read(const std::string& key, c10::IValue& ivalue) {
  ivalue = module_.attr(key);
}

bool InputArchive::try_read(
    const std::string& key,

            

Reported by FlawFinder.

                return true;
}

void InputArchive::read(
    const std::string& key,
    Tensor& tensor,
    bool is_buffer) {
  TORCH_CHECK(
      try_read(key, tensor, is_buffer),

            

Reported by FlawFinder.

                return true;
}

void InputArchive::read(const std::string& key, InputArchive& archive) {
  TORCH_CHECK(
      try_read(key, archive),
      "No such serialized submodule: '",
      hierarchy_prefix_,
      key,

            

Reported by FlawFinder.

                    : data_(data), size_(size) {
    }
    size_t size() const override { return size_; }
    size_t read(uint64_t pos, void* buf, size_t n, const char* what = "")
      const override {
      (void) what;
      if (pos >= size_) {
        return 0;
      }

            

Reported by FlawFinder.

                      size_func_(size_func) {
    }
    size_t size() const override { return size_func_(); }
    size_t read(uint64_t pos, void* buf, size_t n, const char* what = "")
      const override {
      (void)what;
      return read_func_(pos, buf, n);
    }
  private:

            

Reported by FlawFinder.

              from tools.codegen.api.autograd import NativeFunctionWithDifferentiabilityInfo as NFWDI
from tools.codegen.context import native_function_manager
from tools.codegen.utils import T

import functools
from typing import Callable

# Like tools.api.context.with_native_function, but for
# NativeFunctionWithDifferentiabilityInfo.

            

Reported by Pylint.

              from tools.codegen.context import native_function_manager
from tools.codegen.utils import T

import functools
from typing import Callable

# Like tools.api.context.with_native_function, but for
# NativeFunctionWithDifferentiabilityInfo.
def with_native_function_with_differentiability_info(func: Callable[[NFWDI], T]) -> Callable[[NFWDI], T]:

            

Reported by Pylint.

              from tools.codegen.utils import T

import functools
from typing import Callable

# Like tools.api.context.with_native_function, but for
# NativeFunctionWithDifferentiabilityInfo.
def with_native_function_with_differentiability_info(func: Callable[[NFWDI], T]) -> Callable[[NFWDI], T]:
    @functools.wraps(func)

            

Reported by Pylint.

              
# Like tools.api.context.with_native_function, but for
# NativeFunctionWithDifferentiabilityInfo.
def with_native_function_with_differentiability_info(func: Callable[[NFWDI], T]) -> Callable[[NFWDI], T]:
    @functools.wraps(func)
    def wrapper(f: NFWDI) -> T:
        with native_function_manager(f.func):
            return func(f)
    return wrapper

            

Reported by Pylint.

              
# Like tools.api.context.with_native_function, but for
# NativeFunctionWithDifferentiabilityInfo.
def with_native_function_with_differentiability_info(func: Callable[[NFWDI], T]) -> Callable[[NFWDI], T]:
    @functools.wraps(func)
    def wrapper(f: NFWDI) -> T:
        with native_function_manager(f.func):
            return func(f)
    return wrapper

            

Reported by Pylint.

              # NativeFunctionWithDifferentiabilityInfo.
def with_native_function_with_differentiability_info(func: Callable[[NFWDI], T]) -> Callable[[NFWDI], T]:
    @functools.wraps(func)
    def wrapper(f: NFWDI) -> T:
        with native_function_manager(f.func):
            return func(f)
    return wrapper

            

Reported by Pylint.

              from typing import Any, Dict, List, Set, Tuple

from .coverage_record import CoverageRecord
from .llvm_coverage_segment import LlvmCoverageSegment, parse_segments


class LlvmCoverageParser:
    """
    Accepts a parsed json produced by llvm-cov export -- typically,

            

Reported by Pylint.

              from typing import Any, Dict, List, Set, Tuple

from .coverage_record import CoverageRecord
from .llvm_coverage_segment import LlvmCoverageSegment, parse_segments


class LlvmCoverageParser:
    """
    Accepts a parsed json produced by llvm-cov export -- typically,

            

Reported by Pylint.

                      uncovered_lines.difference_update(covered_lines)
        return sorted(covered_lines), sorted(uncovered_lines)

    def parse(self, repo_name: str) -> List[CoverageRecord]:
        # The JSON format is described in the LLVM source code
        # https://github.com/llvm-mirror/llvm/blob/master/tools/llvm-cov/CoverageExporterJson.cpp
        records: List[CoverageRecord] = []
        for export_unit in self._llvm_coverage["data"]:
            for file_info in export_unit["files"]:

            

Reported by Pylint.

              from typing import Any, Dict, List, Set, Tuple

from .coverage_record import CoverageRecord
from .llvm_coverage_segment import LlvmCoverageSegment, parse_segments


class LlvmCoverageParser:
    """
    Accepts a parsed json produced by llvm-cov export -- typically,

            

Reported by Pylint.

              from .llvm_coverage_segment import LlvmCoverageSegment, parse_segments


class LlvmCoverageParser:
    """
    Accepts a parsed json produced by llvm-cov export -- typically,
    representing a single C++ test and produces a list
    of CoverageRecord(s).


            

Reported by Pylint.

                      uncovered_lines.difference_update(covered_lines)
        return sorted(covered_lines), sorted(uncovered_lines)

    def parse(self, repo_name: str) -> List[CoverageRecord]:
        # The JSON format is described in the LLVM source code
        # https://github.com/llvm-mirror/llvm/blob/master/tools/llvm-cov/CoverageExporterJson.cpp
        records: List[CoverageRecord] = []
        for export_unit in self._llvm_coverage["data"]:
            for file_info in export_unit["files"]:

            

Reported by Pylint.

              result = "package_c"


class PackageCObject:
    __slots__ = ["obj"]

    def __init__(self, obj):
        self.obj = obj


            

Reported by Pylint.

              result = "package_c"


class PackageCObject:
    __slots__ = ["obj"]

    def __init__(self, obj):
        self.obj = obj


            

Reported by Pylint.

              result = "package_c"


class PackageCObject:
    __slots__ = ["obj"]

    def __init__(self, obj):
        self.obj = obj


            

Reported by Pylint.

              result = "package_c"


class PackageCObject:
    __slots__ = ["obj"]

    def __init__(self, obj):
        self.obj = obj


            

Reported by Pylint.

                  def __init__(self, obj):
        self.obj = obj

    def return_result(self):
        return result

            

Reported by Pylint.

                  def __init__(self, obj):
        self.obj = obj

    def return_result(self):
        return result

            

Reported by Pylint.

              __import__("subpackage_1", globals(), fromlist=["PackageBSubpackage1Object_0"], level=1)
__import__("subpackage_0.subsubpackage_0", globals(), fromlist=[""], level=1)
__import__("subpackage_2", globals=globals(), locals=locals(), fromlist=["*"], level=1)

result = "package_b"


class PackageBObject:
    __slots__ = ["obj"]

            

Reported by Pylint.

              __import__("subpackage_0.subsubpackage_0", globals(), fromlist=[""], level=1)
__import__("subpackage_2", globals=globals(), locals=locals(), fromlist=["*"], level=1)

result = "package_b"


class PackageBObject:
    __slots__ = ["obj"]


            

Reported by Pylint.

              result = "package_b"


class PackageBObject:
    __slots__ = ["obj"]

    def __init__(self, obj):
        self.obj = obj


            

Reported by Pylint.

              result = "package_b"


class PackageBObject:
    __slots__ = ["obj"]

    def __init__(self, obj):
        self.obj = obj


            

Reported by Pylint.

                  def __init__(self, obj):
        self.obj = obj

    def return_result(self):
        return result

            

Reported by Pylint.

                  def __init__(self, obj):
        self.obj = obj

    def return_result(self):
        return result

            

Reported by Pylint.