]> code.communitydata.science - rises_declines_wikia_code.git/blob - mediawiki_dump_tools/Mediawiki-Utilities/mw/util/ordered.py
add copy of the GPL
[rises_declines_wikia_code.git] / mediawiki_dump_tools / Mediawiki-Utilities / mw / util / ordered.py
1 from . import autovivifying
2
3
4 class Circle(list):
5     def __init__(self, maxsize, iterable=None):
6         self._maxsize = int(maxsize)
7         list.__init__(self, [None] * maxsize)
8         self._size = 0
9         self._pointer = 0
10
11         if iterable is not None:
12             self.extend(iterable)
13
14     def state(self):
15         return list(list.__iter__(self))
16
17     def _internalize(self, index):
18         if self._size < self._maxsize:
19             return index
20         else:
21             return (self._pointer + index) % self._maxsize
22
23     def __iter__(self):
24         for i in range(0, self._size):
25             yield list.__getitem__(self, self._internalize(i))
26
27     def __reversed__(self):
28         for i in range(self._size - 1, -1, -1):
29             yield list.__getitem__(self, self._internalize(i))
30
31     def pop(self, index=None):
32         raise NotImplementedError()
33
34     def __len__(self):
35         return self._size
36
37     def __getitem__(self, index):
38         return list.__getitem__(self, self._internalize(index))
39
40     def append(self, value):
41         # Get the old value
42         old_value = list.__getitem__(self, self._pointer)
43
44         # Update internal list
45         list.__setitem__(self, self._pointer, value)
46
47         # Update state
48         self._pointer = (self._pointer + 1) % self._maxsize
49         self._size = min(self._maxsize, self._size + 1)
50
51         # If we overwrote a value, yield it.
52         return old_value
53
54     def extend(self, values):
55         for value in values:
56             expectorate = self.append(value)
57             if expectorate is not None or self._size == self._maxsize:
58                 yield expectorate
59
60
61 class HistoricalMap(autovivifying.Dict):
62     '''
63     A datastructure for efficiently storing and retrieving a
64     limited number of historical records.
65
66     TODO: Rename this to FIFOCache
67     '''
68
69     def __init__(self, *args, maxlen, **kwargs):
70         '''Maxlen specifies the maximum amount of history to keep'''
71         super().__init__(self, *args, vivifier=lambda k: [], **kwargs)
72
73         self._circle = Circle(maxlen)  # List to preserve order for history
74
75     def __iter__(self):
76         return iter(self._circle)
77
78     def __setitem__(self, key, value):
79         '''Adds a new key-value pair. Returns any discarded values.'''
80
81         # Add to history circle and catch expectorate
82         expectorate = self._circle.append((key, value))
83
84         autovivifying.Dict.__getitem__(self, key).append(value)
85
86         if expectorate is not None:
87             old_key, old_value = expectorate
88             autovivifying.Dict.__getitem__(self, old_key).pop(0)
89             if len(autovivifying.Dict.__getitem__(self, old_key)) == 0:
90                 autovivifying.Dict.__delitem__(self, old_key)
91
92             return (old_key, old_value)
93
94     def insert(self, key, value):
95         return self.__setitem__(key, value)
96
97     def __getitem__(self, key):
98         if key in self:
99             return autovivifying.Dict.__getitem__(self, key)[-1]
100         else:
101             raise KeyError(key)
102
103     def get(self, key):
104         '''Gets the most recently added value for a key'''
105         return self.__getitem__(key)
106
107     def up_to(self, key):
108         '''Gets the recently inserted values up to a key'''
109         for okey, ovalue in reversed(self._circle):
110             if okey == key:
111                 break
112             else:
113                 yield ovalue
114
115     def last(self):
116         return self.circle[-1]

Community Data Science Collective || Want to submit a patch?