Admin Control Panel v2.0

✎ Editing: _pdeque.py ← Back

from collections.abc import Sequence, Hashable
from itertools import islice, chain
from numbers import Integral
from pyrsistent._plist import plist

class PDeque(object):
    """
    Persistent double ended queue (deque). Allows quick appends and pops in both ends. Implemented
    using two persistent lists.

A maximum length can be specified to create a bounded queue.

Fully supports the Sequence and Hashable protocols including indexing and slicing but
    if you need fast random access go for the PVector instead.

Do not instantiate directly, instead use the factory functions :py:func:`dq` or :py:func:`pdeque` to
    create an instance.

Some examples:

>>> x = pdeque([1, 2, 3])
    >>> x.left
    1
    >>> x.right
    3
    >>> x[0] == x.left
    True
    >>> x[-1] == x.right
    True
    >>> x.pop()
    pdeque([1, 2])
    >>> x.pop() == x[:-1]
    True
    >>> x.popleft()
    pdeque([2, 3])
    >>> x.append(4)
    pdeque([1, 2, 3, 4])
    >>> x.appendleft(4)
    pdeque([4, 1, 2, 3])

>>> y = pdeque([1, 2, 3], maxlen=3)
    >>> y.append(4)
    pdeque([2, 3, 4], maxlen=3)
    >>> y.appendleft(4)
    pdeque([4, 1, 2], maxlen=3)
    """
    __slots__ = ('_left_list', '_right_list', '_length', '_maxlen', '__weakref__')

def __new__(cls, left_list, right_list, length, maxlen=None):
        instance = super(PDeque, cls).__new__(cls)
        instance._left_list = left_list
        instance._right_list = right_list
        instance._length = length

if maxlen is not None:
            if not isinstance(maxlen, Integral):
                raise TypeError('An integer is required as maxlen')

if maxlen < 0:
                raise ValueError("maxlen must be non-negative")

instance._maxlen = maxlen
        return instance

@property
    def right(self):
        """
        Rightmost element in dqueue.
        """
        return PDeque._tip_from_lists(self._right_list, self._left_list)

@property
    def left(self):
        """
        Leftmost element in dqueue.
        """
        return PDeque._tip_from_lists(self._left_list, self._right_list)

@staticmethod
    def _tip_from_lists(primary_list, secondary_list):
        if primary_list:
            return primary_list.first

if secondary_list:
            return secondary_list[-1]

raise IndexError('No elements in empty deque')

def __iter__(self):
        return chain(self._left_list, self._right_list.reverse())

def __repr__(self):
        return "pdeque({0}{1})".format(list(self),
                                       ', maxlen={0}'.format(self._maxlen) if self._maxlen is not None else '')
    __str__ = __repr__

@property
    def maxlen(self):
        """
        Maximum length of the queue.
        """
        return self._maxlen

def pop(self, count=1):
        """
        Return new deque with rightmost element removed. Popping the empty queue
        will return the empty queue. A optional count can be given to indicate the
        number of elements to pop. Popping with a negative index is the same as
        popleft. Executes in amortized O(k) where k is the number of elements to pop.

>>> pdeque([1, 2]).pop()
        pdeque([1])
        >>> pdeque([1, 2]).pop(2)
        pdeque([])
        >>> pdeque([1, 2]).pop(-1)
        pdeque([2])
        """
        if count < 0:
            return self.popleft(-count)

new_right_list, new_left_list = PDeque._pop_lists(self._right_list, self._left_list, count)
        return PDeque(new_left_list, new_right_list, max(self._length - count, 0), self._maxlen)

def popleft(self, count=1):
        """
        Return new deque with leftmost element removed. Otherwise functionally
        equivalent to pop().

>>> pdeque([1, 2]).popleft()
        pdeque([2])
        """
        if count < 0:
            return self.pop(-count)

new_left_list, new_right_list = PDeque._pop_lists(self._left_list, self._right_list, count)
        return PDeque(new_left_list, new_right_list, max(self._length - count, 0), self._maxlen)

@staticmethod
    def _pop_lists(primary_list, secondary_list, count):
        new_primary_list = primary_list
        new_secondary_list = secondary_list

while count > 0 and (new_primary_list or new_secondary_list):
            count -= 1
            if new_primary_list.rest:
                new_primary_list = new_primary_list.rest
            elif new_primary_list:
                new_primary_list = new_secondary_list.reverse()
                new_secondary_list = plist()
            else:
                new_primary_list = new_secondary_list.reverse().rest
                new_secondary_list = plist()

return new_primary_list, new_secondary_list

def _is_empty(self):
        return not self._left_list and not self._right_list

def __lt__(self, other):
        if not isinstance(other, PDeque):
            return NotImplemented

return tuple(self) < tuple(other)

def __eq__(self, other):
        if not isinstance(other, PDeque):
            return NotImplemented

if tuple(self) == tuple(other):
            # Sanity check of the length value since it is redundant (there for performance)
            assert len(self) == len(other)
            return True

return False

def __hash__(self):
        return  hash(tuple(self))

def __len__(self):
        return self._length

def append(self, elem):
        """
        Return new deque with elem as the rightmost element.

>>> pdeque([1, 2]).append(3)
        pdeque([1, 2, 3])
        """
        new_left_list, new_right_list, new_length = self._append(self._left_list, self._right_list, elem)
        return PDeque(new_left_list, new_right_list, new_length, self._maxlen)

def appendleft(self, elem):
        """
        Return new deque with elem as the leftmost element.

>>> pdeque([1, 2]).appendleft(3)
        pdeque([3, 1, 2])
        """
        new_right_list, new_left_list, new_length = self._append(self._right_list, self._left_list, elem)
        return PDeque(new_left_list, new_right_list, new_length, self._maxlen)

def _append(self, primary_list, secondary_list, elem):
        if self._maxlen is not None and self._length == self._maxlen:
            if self._maxlen == 0:
                return primary_list, secondary_list, 0
            new_primary_list, new_secondary_list = PDeque._pop_lists(primary_list, secondary_list, 1)
            return new_primary_list, new_secondary_list.cons(elem), self._length

return primary_list, secondary_list.cons(elem), self._length + 1

@staticmethod
    def _extend_list(the_list, iterable):
        count = 0
        for elem in iterable:
            the_list = the_list.cons(elem)
            count += 1

return the_list, count

def _extend(self, primary_list, secondary_list, iterable):
        new_primary_list, extend_count = PDeque._extend_list(primary_list, iterable)
        new_secondary_list = secondary_list
        current_len = self._length + extend_count
        if self._maxlen is not None and current_len > self._maxlen:
            pop_len = current_len - self._maxlen
            new_secondary_list, new_primary_list = PDeque._pop_lists(new_secondary_list, new_primary_list, pop_len)
            extend_count -= pop_len

return new_primary_list, new_secondary_list, extend_count

def extend(self, iterable):
        """
        Return new deque with all elements of iterable appended to the right.

>>> pdeque([1, 2]).extend([3, 4])
        pdeque([1, 2, 3, 4])
        """
        new_right_list, new_left_list, extend_count = self._extend(self._right_list, self._left_list, iterable)
        return PDeque(new_left_list, new_right_list, self._length + extend_count, self._maxlen)

def extendleft(self, iterable):
        """
        Return new deque with all elements of iterable appended to the left.

NB! The elements will be inserted in reverse order compared to the order in the iterable.

>>> pdeque([1, 2]).extendleft([3, 4])
        pdeque([4, 3, 1, 2])
        """
        new_left_list, new_right_list, extend_count = self._extend(self._left_list, self._right_list, iterable)
        return PDeque(new_left_list, new_right_list, self._length + extend_count, self._maxlen)

def count(self, elem):
        """
        Return the number of elements equal to elem present in the queue

>>> pdeque([1, 2, 1]).count(1)
        2
        """
        return self._left_list.count(elem) + self._right_list.count(elem)

def remove(self, elem):
        """
        Return new deque with first element from left equal to elem removed. If no such element is found
        a ValueError is raised.

>>> pdeque([2, 1, 2]).remove(2)
        pdeque([1, 2])
        """
        try:
            return PDeque(self._left_list.remove(elem), self._right_list, self._length - 1)
        except ValueError:
            # Value not found in left list, try the right list
            try:
                # This is severely inefficient with a double reverse, should perhaps implement a remove_last()?
                return PDeque(self._left_list,
                               self._right_list.reverse().remove(elem).reverse(), self._length - 1)
            except ValueError as e:
                raise ValueError('{0} not found in PDeque'.format(elem)) from e

def reverse(self):
        """
        Return reversed deque.

>>> pdeque([1, 2, 3]).reverse()
        pdeque([3, 2, 1])

Also supports the standard python reverse function.

>>> reversed(pdeque([1, 2, 3]))
        pdeque([3, 2, 1])
        """
        return PDeque(self._right_list, self._left_list, self._length)
    __reversed__ = reverse

def rotate(self, steps):
        """
        Return deque with elements rotated steps steps.

>>> x = pdeque([1, 2, 3])
        >>> x.rotate(1)
        pdeque([3, 1, 2])
        >>> x.rotate(-2)
        pdeque([3, 1, 2])
        """
        popped_deque = self.pop(steps)
        if steps >= 0:
            return popped_deque.extendleft(islice(self.reverse(), steps))

return popped_deque.extend(islice(self, -steps))

def __reduce__(self):
        # Pickling support
        return pdeque, (list(self), self._maxlen)

def __getitem__(self, index):
        if isinstance(index, slice):
            if index.step is not None and index.step != 1:
                # Too difficult, no structural sharing possible
                return pdeque(tuple(self)[index], maxlen=self._maxlen)

result = self
            if index.start is not None:
                result = result.popleft(index.start % self._length)
            if index.stop is not None:
                result = result.pop(self._length - (index.stop % self._length))

return result

if not isinstance(index, Integral):
            raise TypeError("'%s' object cannot be interpreted as an index" % type(index).__name__)

if index >= 0:
            return self.popleft(index).left

shifted = len(self) + index
        if shifted < 0:
            raise IndexError(
                "pdeque index {0} out of range {1}".format(index, len(self)),
            )
        return self.popleft(shifted).left

index = Sequence.index

Sequence.register(PDeque)
Hashable.register(PDeque)

def pdeque(iterable=(), maxlen=None):
    """
    Return deque containing the elements of iterable. If maxlen is specified then
    len(iterable) - maxlen elements are discarded from the left to if len(iterable) > maxlen.

>>> pdeque([1, 2, 3])
    pdeque([1, 2, 3])
    >>> pdeque([1, 2, 3, 4], maxlen=2)
    pdeque([3, 4], maxlen=2)
    """
    t = tuple(iterable)
    if maxlen is not None:
        t = t[-maxlen:]
    length = len(t)
    pivot = int(length / 2)
    left = plist(t[:pivot])
    right = plist(t[pivot:], reverse=True)
    return PDeque(left, right, length, maxlen)

def dq(*elements):
    """
    Return deque containing all arguments.

>>> dq(1, 2, 3)
    pdeque([1, 2, 3])
    """
    return pdeque(elements)

Cancel