Mark `fn` for line coverage analysis.

    Uses the 'hotshot' module for fast coverage analysis.

    BUG: Produces inaccurate results.

    See the docstring of `coverage` for usage examples.
168416
def coverage_with_hotshot(fn):
    
    fp = HotShotFuncCoverage(fn)
    # We cannot return fp or fp.__call__ directly as that would break method
    # definitions, instead we need to return a plain function.
    def new_fn(*args, **kw):
        return fp(*args, **kw)
    new_fn.__doc__ = fn.__doc__
    new_fn.__name__ = fn.__name__
    new_fn.__dict__ = fn.__dict__
    new_fn.__module__ = fn.__module__
    return new_fn
373163
def blat(args):
    
    p = OptionParser(blat.__doc__)
    p.set_align(pctid=95, hitlen=30)
    p.set_cpus()
    opts, args = p.parse_args(args)

    if len(args) != 2:
        sys.exit(not p.print_help())

    reffasta, queryfasta = args
    blastfile = get_outfile(reffasta, queryfasta, suffix="blat")

    run_blat(infile=queryfasta, outfile=blastfile, db=reffasta,
             pctid=opts.pctid, hitlen=opts.hitlen, cpus=opts.cpus,
             overwrite=False)

    return blastfile
589080
def _annotate_fn_args(stack, fn_opname, nargs, nkw=-1, consume_fn_name=True):
  """"""
  kwarg_names = []
  if nkw == -1:
    if sys.version_info[0] < 3:
      # Compute nkw and nargs from nargs for python 2.7
      nargs, nkw = (nargs % 256, 2 * nargs // 256)
    else:
      if fn_opname == 'CALL_FUNCTION_KW':
        if qj._DEBUG_QJ:
          assert len(stack) and stack[-1].opname == 'LOAD_CONST'
        if not len(stack) or stack[-1].opname != 'LOAD_CONST':
          return
        se = stack.pop()
        kwarg_names = se.oparg_repr[::-1]
        se.oparg_repr = ['']
        nkw = len(kwarg_names)
        nargs -= nkw
        if qj._DEBUG_QJ:
          assert nargs >= 0 and nkw > 0
      else:
        nkw = 0

  for i in range(nkw):
    se = stack.pop()
    if se.stack_depth == 0 and (len(se.oparg_repr) == 0 or se.oparg_repr[0] == ''):
      # Skip stack entries that don't have any effect on the stack
      continue
    if i % 2 == 1 and sys.version_info[0] < 3:
      if qj._DEBUG_QJ:
        assert se.opname == 'LOAD_CONST'
      if se.opname == 'LOAD_CONST':
        # kwargs are pairs of key=value in code
        se.oparg_repr += ['=']
    else:
      pops = []
      if se.opname.startswith('CALL_FUNCTION'):
        _annotate_fn_args(stack[:], se.opname, se.oparg, -1, True)
      pops = _collect_pops(stack, se.stack_depth - 1 if se.opname.startswith('CALL_FUNCTION') else 0, [], False)
      if i > 1 and len(pops):
        pops[-1].oparg_repr += [',']
      if sys.version_info[0] >= 3:
        target_se = pops[-1] if len(pops) else se
        target_se.oparg_repr = [kwarg_names[i], '='] + target_se.oparg_repr

  for i in range(nargs):
    se = stack.pop()
    if se.opname.startswith('CALL_FUNCTION'):
      _annotate_fn_args(stack, se.opname, se.oparg, -1, True)
    elif len(se.oparg_repr) and se.oparg_repr[0] in {']', '}', ')'}:
      if (i > 0 or nkw > 0):
        se.oparg_repr += [',']
    else:
      pops = _collect_pops(stack, se.stack_depth, [], False)
      if (i > 0 or nkw > 0) and len(pops):
        pops[-1].oparg_repr += [',']

  if consume_fn_name:
    _collect_pops(stack, -1, [], False)
732806
def fastaIterator(fastaFN):
    '''
    
    '''
    if fastaFN[len(fastaFN)-3:] == '.gz':
        fp = gzip.open(fastaFN, 'r')
    else:
        fp = open(fastaFN, 'r')
    
    label = ''
    segments = []
    line = ''
    
    for line in fp:
        if line[0] == '>':
            if label != '':
                yield (label, ''.join(segments))
            label = (line.strip('\n')[1:]).split(' ')[0]
            segments = []
        else:
            segments.append(line.strip('\n'))
            
    if label != '' and len(segments) > 0:
        yield (label, ''.join(segments))
    
    fp.close()
888641
def pasteprepare(args):
    
    p = OptionParser(pasteprepare.__doc__)
    p.add_option("--flank", default=5000, type="int",
                 help="Get the seq of size on two ends [default: %default]")
    opts, args = p.parse_args(args)

    if len(args) != 1:
        sys.exit(not p.print_help())

    goodfasta, = args
    flank = opts.flank
    pf = goodfasta.rsplit(".", 1)[0]
    extbed = pf + ".ext.bed"

    sizes = Sizes(goodfasta)
    fw = open(extbed, "w")
    for bac, size in sizes.iter_sizes():
        print("\t".join(str(x) for x in \
                               (bac, 0, min(flank, size), bac + "L")), file=fw)
        print("\t".join(str(x) for x in \
                               (bac, max(size - flank, 0), size, bac + "R")), file=fw)
    fw.close()

    fastaFromBed(extbed, goodfasta, name=True)
312448
def _trace(self, frame, event, arg):
        

        # Return if this is not a function call
        if event != 'call':
            return

        # Filter calling and called functions by module names
        src_mod = current_module_name(frame.f_back)
        dst_mod = current_module_name(frame)

        # Avoid tracing the tracer (specifically, call from
        # ContextCallTracer.__exit__ to CallTracer.stop)
        if src_mod == __modulename__ or dst_mod == __modulename__:
            return

        # Apply source and destination module filters
        if not self.srcmodflt.match(src_mod):
            return
        if not self.dstmodflt.match(dst_mod):
            return

        # Get calling and called functions
        src_func = current_function(frame.f_back)
        dst_func = current_function(frame)

        # Filter calling and called functions by qnames
        if not self.srcqnmflt.match(function_qname(src_func)):
            return
        if not self.dstqnmflt.match(function_qname(dst_func)):
            return

        # Get calling and called function full names
        src_name = function_fqname(src_func)
        dst_name = function_fqname(dst_func)

        # Modify full function names if necessary
        if self.fnmsub is not None:
            src_name = re.sub(self.fnmsub[0], self.fnmsub[1], src_name)
            dst_name = re.sub(self.fnmsub[0], self.fnmsub[1], dst_name)

        # Update calling function count
        if src_func is not None:
            if src_name in self.fncts:
                self.fncts[src_name][0] += 1
            else:
                self.fncts[src_name] = [1, 0]

        # Update called function count
        if dst_func is not None and src_func is not None:
            if dst_name in self.fncts:
                self.fncts[dst_name][1] += 1
            else:
                self.fncts[dst_name] = [0, 1]

        # Update caller/calling pair count
        if dst_func is not None and src_func is not None:
            key = (src_name, dst_name)
            if key in self.calls:
                self.calls[key] += 1
            else:
                self.calls[key] = 1
823183
private void processFunctions(FileData coverageData, StringBuilder lcov) {
        int total = 0;
        int hit = 0;
        for (int functionNumber = 0; functionNumber < coverageData.getFunctions().size(); functionNumber++) {
            total++;
            Integer functionHits = coverageData.getFunctions().get(functionNumber);
            if (functionHits > 0)
                hit++;

            String functionName = "" + functionNumber;
            lcov.append(format(functionData, functionHits, functionName));
        }

        lcov.append(format(functionFound, total));
        lcov.append(format(functionHit, hit));
    }
310071
def guess_lexer_using_modeline(text):
    

    lexer, accuracy = None, None

    file_type = None
    try:
        file_type = get_filetype_from_buffer(text)
    except:  # pragma: nocover
        log.traceback(logging.DEBUG)

    if file_type is not None:
        try:
            lexer = get_lexer_by_name(file_type)
        except ClassNotFound:
            log.traceback(logging.DEBUG)

    if lexer is not None:
        try:
            accuracy = lexer.analyse_text(text)
        except:  # pragma: nocover
            log.traceback(logging.DEBUG)

    return lexer, accuracy
389632
def determine_coverage(cls, coverage_file):
        
        lines = []
        source_file = 'ERROR'
        for line in coverage_file:
            m = title_re.match(line)
            if m:
                if m.group(2) == '100':
                    return ('', [])
                source_file = m.group(1)
                continue
            m = source_re.match(line)
            if m:
                lines.append(int(m.group(1)))
                continue
            if end_re.match(line):
                break
        line_ranges = cls.make_ranges(lines)
        return (source_file, line_ranges)
620387
function coverage(filename, data) {
  var ret = {
    filename: filename,
    coverage: 0,
    hits: 0,
    misses: 0,
    sloc: 0,
    source: {}
  };

  data.source.forEach(function(line, num){
    num++;

    if (data[num] === 0) {
      ret.misses++;
      ret.sloc++;
    } else if (data[num] !== undefined) {
      ret.hits++;
      ret.sloc++;
    }

    ret.source[num] = {
        source: line
      , coverage: data[num] === undefined
        ? ''
        : data[num]
    };
  });

  ret.coverage = ret.hits / ret.sloc * 100;

  return ret;
}
426542
def _find_current_fn_call(co, lasti):
  """"""
  qj._DEBUG_QJ and qj.LOG_FN('co = {}'.format(co))

  if sys.version_info[0] < 3:
    stack = _build_instruction_stack(co, lasti)
  else:
    stack = _build_instruction_stack3(co, lasti)

  source_lines, source_offset = inspect.getsourcelines(co)
  source_lines = [l.strip() for l in source_lines]

  # Apply stack effects backwards until we arrive at the stack entries for a complete function call
  fn_stack = _collect_pops(stack[:-1], stack[-1].stack_depth, [], 0)

  if not fn_stack and stack[-1].stack_depth == 0:
    # The function call took 0 arguments, so return early with a special string.
    return '<empty log>'

  qj._DEBUG_QJ and qj.LOG_FN('collected fn_stack:\n%s\n\n' % '\n'.join(str(se) for se in fn_stack))

  # Prepare to annotate the stack with extra symbols, and filter out MAKE_FUNCTION and MAKE_CLOSURE calls, which are no longer needed.
  annotate_stack = [se for se in fn_stack if not se.opname.startswith('MAKE_')]
  annotate_stack.reverse()
  _annotate_fn_args(annotate_stack, stack[-1].opname, stack[-1].oparg, -1, False)

  qj._DEBUG_QJ and qj.LOG_FN('annotated fn_stack:\n%s\n\n' % '\n'.join(str(se) for se in fn_stack))

  # Find the range of lines to search over.
  min_l = stack[-1].curr_l
  max_l = stack[0].curr_l
  for se in fn_stack:
    min_l = min(min_l, se.curr_l)
    max_l = max(max_l, se.curr_l)

  qj._DEBUG_QJ and qj.LOG_FN('source lines range: %d -> %d (indices: %d, %d)' % (min_l, max_l, min_l - source_offset, max_l - source_offset + 1))
  source_chunk = ' '.join(
      [l for l in source_lines[min_l - source_offset:max_l - source_offset + 1] if l and not l.startswith('#')])

  qj._DEBUG_QJ and qj.LOG_FN('source_chunk: %r' % source_chunk)

  # Build up all of the tokens for the function call.
  tokens = []
  for se in fn_stack:
    opname = se.opname
    for oparg_repr in se.oparg_repr[::-1]:
      # Clean up the tokens
      if not oparg_repr:
        continue
      oparg_repr = re.escape(oparg_repr.replace('\n', '\\n'))
      tokens.append(oparg_repr)

  tokens.reverse()
  qj._DEBUG_QJ and qj.LOG_FN('extracted tokens: {}'.format(tokens))
  if qj._DEBUG_QJ:
    assert tokens

  # Add tokens for the function call we're extracting.
  tokens = ['\\('] + tokens + ['\\)']

  reg = '\b*?.*?\b*?'.join(tokens)
  qj._DEBUG_QJ and qj.LOG_FN(reg)

  # Search for the function call using the full set of tokens.
  # Expand the search with source lines after the current set if we don't find a match.
  shortest_match = None
  match_attempts = 0
  max_match_attempts = 10
  while not shortest_match and match_attempts < max_match_attempts:
    (shortest_match, _, _) = (
        _find_earliest_shortest_match(source_chunk, reg, 0, len(source_chunk), num_attempts=10))
    if not shortest_match:
      match_attempts += 1

      min_l -= 1
      prev_line = ''
      while not prev_line and min_l - source_offset >= 0 and min_l - source_offset < len(source_lines):
        prev_line = source_lines[min_l - source_offset]
        if prev_line.startswith('#'):
          prev_line = ''
        if not prev_line:
          min_l -= 1
      prev_line += ' ' if prev_line else ''
      source_chunk = prev_line + source_chunk

      max_l += 1
      next_line = ''
      while not next_line and max_l - source_offset >= 0 and max_l - source_offset < len(source_lines):
        next_line = source_lines[max_l - source_offset]
        if next_line.startswith('#'):
          next_line = ''
        if not next_line:
          max_l += 1
      next_line = (' ' if next_line else '') + next_line
      source_chunk += next_line

  # Return the string for the function call
  if qj._DEBUG_QJ:
    assert shortest_match
  if shortest_match is not None:
    match = shortest_match.group(0)
    if qj._DEBUG_QJ:
      assert match.startswith('(') and match.endswith(')')
    # Do some parentheses cleanup.
    if match.startswith('('):
      match = match[1:]
    lparens = match.count('(')
    rparens = match.count(')')
    if lparens > rparens:
      match += ')' * (lparens - rparens)
    elif lparens < rparens:
      rparens_to_remove = rparens - lparens
      while len(match) and match[-1] == ')' and rparens_to_remove > 0:
        match = match[:-1]
        rparens_to_remove -= 1
    match = match.strip()
    return match
  else:
    return ''
1251209
def script_lines_workaround(line_info, coverage_info, count_info)
      is_repeated = lambda do |div|
        n = line_info.size / div
        break false unless line_info.size % div == 0 && n > 1
        different = false
        n.times do |i|
        
          things = (0...div).map { |j| line_info[i + j * n] }
          if things.uniq.size != 1
            different = true
            break
          end
        end

        ! different
      end

      factors = braindead_factorize(line_info.size)
      factors.each do |n|
        if is_repeated[n]
          line_info = line_info[0, line_info.size / n]
          coverage_info = coverage_info[0, coverage_info.size / n]
          count_info = count_info[0, count_info.size / n]
        end
      end if factors.size > 1   # don't even try if it's prime

      [line_info, coverage_info, count_info]
    end
246430
def count(args):
    
    p = OptionParser(count.__doc__)
    opts, args = p.parse_args(args)

    if len(args) != 2:
        sys.exit(not p.print_help())

    coveragefile, fastafile = args

    countsfile = coveragefile.split(".")[0] + ".bin"
    if op.exists(countsfile):
        logging.error("`{0}` file exists. Remove before proceed."\
                .format(countsfile))
        return

    fastasize, sizes, offsets = get_offsets(fastafile)
    logging.debug("Initialize array of uint8 with size {0}".format(fastasize))
    ar = np.zeros(fastasize, dtype=np.uint8)

    update_array(ar, coveragefile, sizes, offsets)

    ar.tofile(countsfile)
    logging.debug("Array written to `{0}`".format(countsfile))
1362848
def main(argv=None, directory=None):
    
    logging.basicConfig(format='%(message)s')

    argv = argv or sys.argv
    arg_dict = parse_coverage_args(argv[1:])
    GitPathTool.set_cwd(directory)
    fail_under = arg_dict.get('fail_under')
    percent_covered = generate_coverage_report(
        arg_dict['coverage_xml'],
        arg_dict['compare_branch'],
        html_report=arg_dict['html_report'],
        css_file=arg_dict['external_css_file'],
        ignore_staged=arg_dict['ignore_staged'],
        ignore_unstaged=arg_dict['ignore_unstaged'],
        exclude=arg_dict['exclude'],
        src_roots=arg_dict['src_roots'],
    )

    if percent_covered >= fail_under:
        return 0
    else:
        LOGGER.error("Failure. Coverage is below {}%.".format(fail_under))
        return 1
469291
def debug(text):
    
    frame = inspect.currentframe().f_back
    module = frame.f_globals['__name__']
    func = frame.f_code.co_name
    msg = "%s.%s: %s" % (module, func, text)
    _LOGGER.debug(msg)
839247
def guard(ctx, opts=""):
    
    # TODO if coverage was run via pytest-cov, we could add coverage here too
    return test(ctx, include_slow=True, loop_on_fail=True, opts=opts)
529671
def mark_bl(self) -> int:
        
        counter = 0
        stringy_lines = find_stringy_lines(self.node, self.first_line_no)
        for relative_line_number, line in enumerate(self.lines):
            if relative_line_number not in stringy_lines and line.strip() == '':
                counter += 1
                self.line_markers[relative_line_number] = LineType.blank_line

        return counter