mirror of
https://github.com/Mikaela/Limnoria.git
synced 2024-11-23 19:19:32 +01:00
377 lines
11 KiB
Python
377 lines
11 KiB
Python
#!/usr/bin/env python
|
|
|
|
###
|
|
# Copyright (c) 2002, Jeremiah Fincher
|
|
# All rights reserved.
|
|
#
|
|
# Redistribution and use in source and binary forms, with or without
|
|
# modification, are permitted provided that the following conditions are met:
|
|
#
|
|
# * Redistributions of source code must retain the above copyright notice,
|
|
# this list of conditions, and the following disclaimer.
|
|
# * Redistributions in binary form must reproduce the above copyright notice,
|
|
# this list of conditions, and the following disclaimer in the
|
|
# documentation and/or other materials provided with the distribution.
|
|
# * Neither the name of the author of this software nor the name of
|
|
# contributors to this software may be used to endorse or promote products
|
|
# derived from this software without specific prior written consent.
|
|
#
|
|
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
|
|
# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
|
|
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
|
|
# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
|
|
# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
|
|
# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
|
|
# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
|
|
# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
|
|
# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
|
|
# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
|
|
# POSSIBILITY OF SUCH DAMAGE.
|
|
###
|
|
|
|
"""
|
|
Data structures for Python.
|
|
"""
|
|
|
|
## from __future__ import generators
|
|
|
|
__revision__ = "$Id$"
|
|
|
|
import fix
|
|
|
|
import types
|
|
import pprint
|
|
import string
|
|
from itertools import imap
|
|
|
|
class RingBuffer(object):
|
|
"""Class to represent a fixed-size ring buffer."""
|
|
__slots__ = ('L', 'i', 'full', 'maxSize')
|
|
def __init__(self, maxSize, seq=()):
|
|
if maxSize <= 0:
|
|
raise ValueError, 'maxSize must be > 0.'
|
|
self.maxSize = maxSize
|
|
self.reset()
|
|
for elt in seq:
|
|
self.append(elt)
|
|
|
|
def reset(self):
|
|
self.full = False
|
|
self.L = []
|
|
self.i = 0
|
|
|
|
def __len__(self):
|
|
return len(self.L)
|
|
|
|
def __eq__(self, other):
|
|
if self.__class__ == other.__class__ and \
|
|
self.maxSize == other.maxSize and len(self) == len(other):
|
|
iterator = iter(other)
|
|
for elt in self:
|
|
otherelt = iterator.next()
|
|
if not elt == otherelt:
|
|
return False
|
|
return True
|
|
return False
|
|
|
|
def __nonzero__(self):
|
|
return len(self) > 0
|
|
|
|
def __contains__(self, elt):
|
|
return elt in self.L
|
|
|
|
def append(self, elt):
|
|
if self.full:
|
|
self.L[self.i] = elt
|
|
self.i += 1
|
|
self.i %= len(self.L)
|
|
else:
|
|
if len(self) >= self.maxSize:
|
|
self.full = True
|
|
self.append(elt)
|
|
else:
|
|
self.L.append(elt)
|
|
|
|
def extend(self, seq):
|
|
for elt in seq:
|
|
self.append(elt)
|
|
|
|
def __getitem__(self, idx):
|
|
if self.full:
|
|
oidx = idx
|
|
if type(oidx) == types.SliceType:
|
|
L = []
|
|
for i in xrange(*slice.indices(oidx, len(self))):
|
|
L.append(self[i])
|
|
return L
|
|
else:
|
|
(m, idx) = divmod(oidx, len(self.L))
|
|
if m and m != -1:
|
|
raise IndexError, oidx
|
|
idx = (idx + self.i) % len(self.L)
|
|
return self.L[idx]
|
|
else:
|
|
if type(idx) == types.SliceType:
|
|
L = []
|
|
for i in xrange(*slice.indices(idx, len(self))):
|
|
L.append(self[i])
|
|
return L
|
|
else:
|
|
return self.L[idx]
|
|
|
|
def __setitem__(self, idx, elt):
|
|
if self.full:
|
|
oidx = idx
|
|
if type(oidx) == types.SliceType:
|
|
range = xrange(*slice.indices(oidx, len(self)))
|
|
if len(range) != len(elt):
|
|
raise ValueError, 'seq must be the same length as slice.'
|
|
else:
|
|
for (i, x) in zip(range, elt):
|
|
self[i] = x
|
|
else:
|
|
(m, idx) = divmod(oidx, len(self.L))
|
|
if m and m != -1:
|
|
raise IndexError, oidx
|
|
idx = (idx + self.i) % len(self.L)
|
|
self.L[idx] = elt
|
|
else:
|
|
if type(idx) == types.SliceType:
|
|
range = xrange(*slice.indices(idx, len(self)))
|
|
if len(range) != len(elt):
|
|
raise ValueError, 'seq must be the same length as slice.'
|
|
else:
|
|
for (i, x) in zip(range, elt):
|
|
self[i] = x
|
|
else:
|
|
self.L[idx] = elt
|
|
|
|
def __repr__(self):
|
|
if self.full:
|
|
return 'RingBuffer(%r, %r)' % (self.maxSize, list(self))
|
|
else:
|
|
return 'RingBuffer(%r, %r)' % (self.maxSize, list(self))
|
|
|
|
def __getstate__(self):
|
|
return (self.maxSize, self.full, self.i, self.L)
|
|
|
|
def __setstate__(self, (maxSize, full, i, L)):
|
|
self.maxSize = maxSize
|
|
self.full = full
|
|
self.i = i
|
|
self.L = L
|
|
|
|
|
|
class queue(object):
|
|
"""Queue class for handling large queues. Queues smaller than 1,000 or so
|
|
elements are probably better served by the smallqueue class.
|
|
"""
|
|
__slots__ = ('front', 'back')
|
|
def __init__(self, seq=()):
|
|
self.back = []
|
|
self.front = []
|
|
for elt in seq:
|
|
self.enqueue(elt)
|
|
|
|
def reset(self):
|
|
self.back[:] = []
|
|
self.front[:] = []
|
|
|
|
def enqueue(self, elt):
|
|
self.back.append(elt)
|
|
|
|
def dequeue(self):
|
|
try:
|
|
return self.front.pop()
|
|
except IndexError:
|
|
self.back.reverse()
|
|
self.front = self.back
|
|
self.back = []
|
|
return self.front.pop()
|
|
|
|
def peek(self):
|
|
if self.front:
|
|
return self.front[-1]
|
|
else:
|
|
return self.back[0]
|
|
|
|
def __len__(self):
|
|
return len(self.front) + len(self.back)
|
|
|
|
def __nonzero__(self):
|
|
return bool(self.back or self.front)
|
|
|
|
def __contains__(self, elt):
|
|
return elt in self.front or elt in self.back
|
|
|
|
def __iter__(self):
|
|
for elt in reviter(self.front):
|
|
yield elt
|
|
for elt in self.back:
|
|
yield elt
|
|
|
|
def __eq__(self, other):
|
|
if len(self) == len(other):
|
|
otheriter = iter(other)
|
|
for elt in self:
|
|
otherelt = otheriter.next()
|
|
if not (elt == otherelt):
|
|
return False
|
|
return True
|
|
else:
|
|
return False
|
|
|
|
def __repr__(self):
|
|
return 'queue([%s])' % ', '.join(imap(repr, self))
|
|
|
|
def __getitem__(self, oidx):
|
|
if len(self) == 0:
|
|
raise IndexError, 'queue index out of range'
|
|
if type(oidx) == types.SliceType:
|
|
L = []
|
|
for i in xrange(*slice.indices(oidx, len(self))):
|
|
L.append(self[i])
|
|
return L
|
|
else:
|
|
(m, idx) = divmod(oidx, len(self))
|
|
if m and m != -1:
|
|
raise IndexError, oidx
|
|
if len(self.front) > idx:
|
|
return self.front[-(idx+1)]
|
|
else:
|
|
return self.back[(idx-len(self.front))]
|
|
|
|
def __setitem__(self, oidx, value):
|
|
if len(self) == 0:
|
|
raise IndexError, 'queue index out of range'
|
|
if type(oidx) == types.SliceType:
|
|
range = xrange(*slice.indices(oidx, len(self)))
|
|
if len(range) != len(value):
|
|
raise ValueError, 'seq must be the same length as slice.'
|
|
else:
|
|
for i in range:
|
|
(m, idx) = divmod(oidx, len(self))
|
|
if m and m != -1:
|
|
raise IndexError, oidx
|
|
for (i, x) in zip(range, value):
|
|
self[i] = x
|
|
else:
|
|
(m, idx) = divmod(oidx, len(self))
|
|
if m and m != -1:
|
|
raise IndexError, oidx
|
|
if len(self.front) > idx:
|
|
self.front[-(idx+1)] = value
|
|
else:
|
|
self.back[idx-len(self.front)] = value
|
|
|
|
def __delitem__(self, oidx):
|
|
if type(oidx) == types.SliceType:
|
|
range = xrange(*slice.indices(oidx, len(self)))
|
|
for i in range:
|
|
del self[i]
|
|
else:
|
|
(m, idx) = divmod(oidx, len(self))
|
|
if m and m != -1:
|
|
raise IndexError, oidx
|
|
if len(self.front) > idx:
|
|
del self.front[-(idx+1)]
|
|
else:
|
|
del self.back[idx-len(self.front)]
|
|
|
|
def __getstate__(self):
|
|
return (list(self),)
|
|
|
|
def __setstate__(self, (L,)):
|
|
L.reverse()
|
|
self.front = L
|
|
self.back = []
|
|
|
|
class smallqueue(list):
|
|
__slots__ = ()
|
|
def enqueue(self, elt):
|
|
self.append(elt)
|
|
|
|
def dequeue(self):
|
|
return self.pop(0)
|
|
|
|
def peek(self):
|
|
return self[0]
|
|
|
|
def __repr__(self):
|
|
return 'smallqueue([%s])' % ', '.join(imap(repr, self))
|
|
|
|
def reset(self):
|
|
self[:] = []
|
|
|
|
|
|
class MaxLengthQueue(queue):
|
|
__slots__ = ('length',)
|
|
def __init__(self, length, seq=()):
|
|
self.length = length
|
|
queue.__init__(self, seq)
|
|
|
|
def __getstate__(self):
|
|
return (self.length, queue.__getstate__(self))
|
|
|
|
def __setstate__(self, (length, q)):
|
|
self.length = length
|
|
queue.__setstate__(self, q)
|
|
|
|
def enqueue(self, elt):
|
|
queue.enqueue(self, elt)
|
|
if len(self) > self.length:
|
|
self.dequeue()
|
|
|
|
|
|
class TwoWayDictionary(dict):
|
|
__slots__ = ()
|
|
def __init__(self, seq=(), **kwargs):
|
|
if hasattr(seq, 'iteritems'):
|
|
seq = seq.iteritems()
|
|
elif hasattr(seq, 'items'):
|
|
seq = seq.items()
|
|
for (key, value) in seq:
|
|
self[key] = value
|
|
self[value] = key
|
|
for (key, value) in kwargs.iteritems():
|
|
self[key] = value
|
|
self[value] = key
|
|
|
|
def __setitem__(self, key, value):
|
|
dict.__setitem__(self, key, value)
|
|
dict.__setitem__(self, value, key)
|
|
|
|
def __delitem__(self, key):
|
|
value = self[key]
|
|
dict.__delitem__(self, key)
|
|
dict.__delitem__(self, value)
|
|
|
|
|
|
class PersistentDictionary(dict):
|
|
_trans = string.maketrans('\r\n', ' ')
|
|
_locals = locals
|
|
_globals = globals
|
|
def __init__(self, filename, globals=None, locals=None):
|
|
mylocals = locals
|
|
myglobals = globals
|
|
if mylocals is None:
|
|
mylocals = self._locals()
|
|
if myglobals is None:
|
|
myglobals = self._globals()
|
|
self.filename = filename
|
|
try:
|
|
fd = file(filename, 'r')
|
|
s = fd.read()
|
|
fd.close()
|
|
s.translate(self._trans)
|
|
super(self.__class__, self).__init__(eval(s, myglobals, mylocals))
|
|
except IOError:
|
|
pass
|
|
|
|
def close(self):
|
|
fd = file(self.filename, 'w')
|
|
fd.write(pprint.pformat(self))
|
|
fd.close()
|
|
|
|
flush = close
|