comparison mercurial/linelog.py @ 43076:2372284d9457

formatting: blacken the codebase

This is using my patch to black
(https://github.com/psf/black/pull/826) so we don't un-wrap collection
literals.

Done with:

  hg files 'set:**.py - mercurial/thirdparty/** - "contrib/python-zstandard/**"' | xargs black -S

# skip-blame mass-reformatting only

# no-check-commit reformats foo_bar functions

Differential Revision: https://phab.mercurial-scm.org/D6971

43075:57875cf423c9

from __future__ import absolute_import, print_function

import abc
import struct

from .thirdparty import (
    attr,
)
from . import (
    pycompat,
)

_llentry = struct.Struct('>II')

class LineLogError(Exception):
    """Error raised when something bad happens internally in linelog."""

@attr.s
class lineinfo(object):
    # Introducing revision of this line.
    rev = attr.ib()
    # Line number for this line in its introducing revision.
    linenum = attr.ib()
    # Private. Offset in the linelog program of this line. Used internally.
    _offset = attr.ib()

@attr.s
class annotateresult(object):
    rev = attr.ib()
    lines = attr.ib()
    _eof = attr.ib()

    def __iter__(self):
        return iter(self.lines)

class _llinstruction(object):

    __metaclass__ = abc.ABCMeta

    @abc.abstractmethod

        Returns:
          The new value of pc. Returns None if exeuction should stop
          (that is, we've found the end of the file.)
        """

class _jge(_llinstruction):
    """If the current rev is greater than or equal to op1, jump to op2."""

    def __init__(self, op1, op2):
        self._cmprev = op1

    def __str__(self):
        return r'JGE %d %d' % (self._cmprev, self._target)

    def __eq__(self, other):
        return (type(self) == type(other)
                and self._cmprev == other._cmprev
                and self._target == other._target)

    def encode(self):
        return _llentry.pack(self._cmprev << 2, self._target)

    def execute(self, rev, pc, emit):
        if rev >= self._cmprev:
            return self._target
        return pc + 1

class _jump(_llinstruction):
    """Unconditional jumps are expressed as a JGE with op1 set to 0."""

    def __init__(self, op1, op2):
        if op1 != 0:

    def __str__(self):
        return r'JUMP %d' % (self._target)

    def __eq__(self, other):
        return (type(self) == type(other)
                and self._target == other._target)

    def encode(self):
        return _llentry.pack(0, self._target)

    def execute(self, rev, pc, emit):
        return self._target

class _eof(_llinstruction):
    """EOF is expressed as a JGE that always jumps to 0."""

    def __init__(self, op1, op2):

        return _llentry.pack(0, 0)

    def execute(self, rev, pc, emit):
        return None

class _jl(_llinstruction):
    """If the current rev is less than op1, jump to op2."""

    def __init__(self, op1, op2):
        self._cmprev = op1

    def __str__(self):
        return r'JL %d %d' % (self._cmprev, self._target)

    def __eq__(self, other):
        return (type(self) == type(other)
                and self._cmprev == other._cmprev
                and self._target == other._target)

    def encode(self):
        return _llentry.pack(1 | (self._cmprev << 2), self._target)

    def execute(self, rev, pc, emit):
        if rev < self._cmprev:
            return self._target
        return pc + 1

class _line(_llinstruction):
    """Emit a line."""

    def __init__(self, op1, op2):

    def __str__(self):
        return r'LINE %d %d' % (self._rev, self._origlineno)

    def __eq__(self, other):
        return (type(self) == type(other)
                and self._rev == other._rev
                and self._origlineno == other._origlineno)

    def encode(self):
        return _llentry.pack(2 | (self._rev << 2), self._origlineno)

    def execute(self, rev, pc, emit):
        emit(lineinfo(self._rev, self._origlineno, pc))
        return pc + 1

def _decodeone(data, offset):
    """Decode a single linelog instruction from an offset in a buffer."""
    try:
        op1, op2 = _llentry.unpack_from(data, offset)

        return _jl(op1, op2)
    elif opcode == 2:
        return _line(op1, op2)
    raise NotImplementedError('Unimplemented opcode %r' % opcode)

class linelog(object):
    """Efficient cache for per-line history information."""

    def __init__(self, program=None, maxrev=0):
        if program is None:

        self._program = program
        self._lastannotate = None
        self._maxrev = maxrev

    def __eq__(self, other):
        return (type(self) == type(other)
                and self._program == other._program
                and self._maxrev == other._maxrev)

    def __repr__(self):
        return '<linelog at %s: maxrev=%d size=%d>' % (
            hex(id(self)), self._maxrev, len(self._program))

    def debugstr(self):
        fmt = r'%%%dd %%s' % len(str(len(self._program)))
        return pycompat.sysstr('\n').join(
            fmt % (idx, i) for idx, i in enumerate(self._program[1:], 1))

    @classmethod
    def fromdata(cls, buf):
        if len(buf) % _llentry.size != 0:
            raise LineLogError(
                "invalid linelog buffer size %d (must be a multiple of %d)" % (
                    len(buf), _llentry.size))
        expected = len(buf) / _llentry.size
        fakejge = _decodeone(buf, 0)
        if isinstance(fakejge, _jump):
            maxrev = 0
        else:
            maxrev = fakejge._cmprev
        numentries = fakejge._target
        if expected != numentries:
            raise LineLogError("corrupt linelog data: claimed"
                               " %d entries but given data for %d entries" % (
                                   expected, numentries))
        instructions = [_eof(0, 0)]
        for offset in pycompat.xrange(1, numentries):
            instructions.append(_decodeone(buf, offset * _llentry.size))
        return cls(instructions, maxrev=maxrev)

        self._program = []
        self._maxrev = 0
        self._lastannotate = None

    def replacelines_vec(self, rev, a1, a2, blines):
        return self.replacelines(rev, a1, a2, 0, len(blines),
                                 _internal_blines=blines)

    def replacelines(self, rev, a1, a2, b1, b2, _internal_blines=None):
        """Replace lines [a1, a2) with lines [b1, b2)."""
        if self._lastannotate:
            # TODO(augie): make replacelines() accept a revision at

        else:
            ar = self.annotate(rev)
            #        ar = self.annotate(self._maxrev)
        if a1 > len(ar.lines):
            raise LineLogError(
                '%d contains %d lines, tried to access line %d' % (
                    rev, len(ar.lines), a1))
        elif a1 == len(ar.lines):
            # Simulated EOF instruction since we're at EOF, which
            # doesn't have a "real" line.
            a1inst = _eof(0, 0)
            a1info = lineinfo(0, 0, ar._eof)

                    appendinst(_line(newrev, newlinenum))
        # delete
        if a1 < a2:
            if a2 > len(ar.lines):
                raise LineLogError(
                    '%d contains %d lines, tried to access line %d' % (
                        rev, len(ar.lines), a2))
            elif a2 == len(ar.lines):
                endaddr = ar._eof
            else:
                endaddr = ar.lines[a2]._offset
            if a2 > 0 and rev < self._maxrev:

            lastpc = pc
            pc = inst.execute(rev, pc, lines.append)
            executed += 1
        if pc is not None:
            raise LineLogError(
                r'Probably hit an infinite loop in linelog. Program:\n' +
                self.debugstr())
        ar = annotateresult(rev, lines, lastpc)
        self._lastannotate = ar
        return ar

    @property

43076:2372284d9457

from __future__ import absolute_import, print_function

import abc
import struct

from .thirdparty import attr
from . import pycompat

_llentry = struct.Struct('>II')


class LineLogError(Exception):
    """Error raised when something bad happens internally in linelog."""


@attr.s
class lineinfo(object):
    # Introducing revision of this line.
    rev = attr.ib()
    # Line number for this line in its introducing revision.
    linenum = attr.ib()
    # Private. Offset in the linelog program of this line. Used internally.
    _offset = attr.ib()


@attr.s
class annotateresult(object):
    rev = attr.ib()
    lines = attr.ib()
    _eof = attr.ib()

    def __iter__(self):
        return iter(self.lines)


class _llinstruction(object):

    __metaclass__ = abc.ABCMeta

    @abc.abstractmethod

        Returns:
          The new value of pc. Returns None if exeuction should stop
          (that is, we've found the end of the file.)
        """


class _jge(_llinstruction):
    """If the current rev is greater than or equal to op1, jump to op2."""

    def __init__(self, op1, op2):
        self._cmprev = op1

    def __str__(self):
        return r'JGE %d %d' % (self._cmprev, self._target)

    def __eq__(self, other):
        return (
            type(self) == type(other)
            and self._cmprev == other._cmprev
            and self._target == other._target
        )

    def encode(self):
        return _llentry.pack(self._cmprev << 2, self._target)

    def execute(self, rev, pc, emit):
        if rev >= self._cmprev:
            return self._target
        return pc + 1


class _jump(_llinstruction):
    """Unconditional jumps are expressed as a JGE with op1 set to 0."""

    def __init__(self, op1, op2):
        if op1 != 0:

    def __str__(self):
        return r'JUMP %d' % (self._target)

    def __eq__(self, other):
        return type(self) == type(other) and self._target == other._target

    def encode(self):
        return _llentry.pack(0, self._target)

    def execute(self, rev, pc, emit):
        return self._target


class _eof(_llinstruction):
    """EOF is expressed as a JGE that always jumps to 0."""

    def __init__(self, op1, op2):

        return _llentry.pack(0, 0)

    def execute(self, rev, pc, emit):
        return None


class _jl(_llinstruction):
    """If the current rev is less than op1, jump to op2."""

    def __init__(self, op1, op2):
        self._cmprev = op1

    def __str__(self):
        return r'JL %d %d' % (self._cmprev, self._target)

    def __eq__(self, other):
        return (
            type(self) == type(other)
            and self._cmprev == other._cmprev
            and self._target == other._target
        )

    def encode(self):
        return _llentry.pack(1 | (self._cmprev << 2), self._target)

    def execute(self, rev, pc, emit):
        if rev < self._cmprev:
            return self._target
        return pc + 1


class _line(_llinstruction):
    """Emit a line."""

    def __init__(self, op1, op2):

    def __str__(self):
        return r'LINE %d %d' % (self._rev, self._origlineno)

    def __eq__(self, other):
        return (
            type(self) == type(other)
            and self._rev == other._rev
            and self._origlineno == other._origlineno
        )

    def encode(self):
        return _llentry.pack(2 | (self._rev << 2), self._origlineno)

    def execute(self, rev, pc, emit):
        emit(lineinfo(self._rev, self._origlineno, pc))
        return pc + 1


def _decodeone(data, offset):
    """Decode a single linelog instruction from an offset in a buffer."""
    try:
        op1, op2 = _llentry.unpack_from(data, offset)

        return _jl(op1, op2)
    elif opcode == 2:
        return _line(op1, op2)
    raise NotImplementedError('Unimplemented opcode %r' % opcode)


class linelog(object):
    """Efficient cache for per-line history information."""

    def __init__(self, program=None, maxrev=0):
        if program is None:

        self._program = program
        self._lastannotate = None
        self._maxrev = maxrev

    def __eq__(self, other):
        return (
            type(self) == type(other)
            and self._program == other._program
            and self._maxrev == other._maxrev
        )

    def __repr__(self):
        return '<linelog at %s: maxrev=%d size=%d>' % (
            hex(id(self)),
            self._maxrev,
            len(self._program),
        )

    def debugstr(self):
        fmt = r'%%%dd %%s' % len(str(len(self._program)))
        return pycompat.sysstr('\n').join(
            fmt % (idx, i) for idx, i in enumerate(self._program[1:], 1)
        )

    @classmethod
    def fromdata(cls, buf):
        if len(buf) % _llentry.size != 0:
            raise LineLogError(
                "invalid linelog buffer size %d (must be a multiple of %d)"
                % (len(buf), _llentry.size)
            )
        expected = len(buf) / _llentry.size
        fakejge = _decodeone(buf, 0)
        if isinstance(fakejge, _jump):
            maxrev = 0
        else:
            maxrev = fakejge._cmprev
        numentries = fakejge._target
        if expected != numentries:
            raise LineLogError(
                "corrupt linelog data: claimed"
                " %d entries but given data for %d entries"
                % (expected, numentries)
            )
        instructions = [_eof(0, 0)]
        for offset in pycompat.xrange(1, numentries):
            instructions.append(_decodeone(buf, offset * _llentry.size))
        return cls(instructions, maxrev=maxrev)

        self._program = []
        self._maxrev = 0
        self._lastannotate = None

    def replacelines_vec(self, rev, a1, a2, blines):
        return self.replacelines(
            rev, a1, a2, 0, len(blines), _internal_blines=blines
        )

    def replacelines(self, rev, a1, a2, b1, b2, _internal_blines=None):
        """Replace lines [a1, a2) with lines [b1, b2)."""
        if self._lastannotate:
            # TODO(augie): make replacelines() accept a revision at

        else:
            ar = self.annotate(rev)
            #        ar = self.annotate(self._maxrev)
        if a1 > len(ar.lines):
            raise LineLogError(
                '%d contains %d lines, tried to access line %d'
                % (rev, len(ar.lines), a1)
            )
        elif a1 == len(ar.lines):
            # Simulated EOF instruction since we're at EOF, which
            # doesn't have a "real" line.
            a1inst = _eof(0, 0)
            a1info = lineinfo(0, 0, ar._eof)

                    appendinst(_line(newrev, newlinenum))
        # delete
        if a1 < a2:
            if a2 > len(ar.lines):
                raise LineLogError(
                    '%d contains %d lines, tried to access line %d'
                    % (rev, len(ar.lines), a2)
                )
            elif a2 == len(ar.lines):
                endaddr = ar._eof
            else:
                endaddr = ar.lines[a2]._offset
            if a2 > 0 and rev < self._maxrev:

            lastpc = pc
            pc = inst.execute(rev, pc, lines.append)
            executed += 1
        if pc is not None:
            raise LineLogError(
                r'Probably hit an infinite loop in linelog. Program:\n'
                + self.debugstr()
            )
        ar = annotateresult(rev, lines, lastpc)
        self._lastannotate = ar
        return ar

    @property