Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/pyrsistent/_pdeque.py: 28%

1from collections.abc import Sequence, Hashable 

2from itertools import islice, chain 

3from numbers import Integral 

4from pyrsistent._plist import plist 

5 

6 

7class PDeque(object): 

8 """ 

9 Persistent double ended queue (deque). Allows quick appends and pops in both ends. Implemented 

10 using two persistent lists. 

11 

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

13 

14 Fully supports the Sequence and Hashable protocols including indexing and slicing but 

15 if you need fast random access go for the PVector instead. 

16 

17 Do not instantiate directly, instead use the factory functions :py:func:`dq` or :py:func:`pdeque` to 

18 create an instance. 

19 

20 Some examples: 

21 

22 >>> x = pdeque([1, 2, 3]) 

23 >>> x.left 

24 1 

25 >>> x.right 

26 3 

27 >>> x[0] == x.left 

28 True 

29 >>> x[-1] == x.right 

30 True 

31 >>> x.pop() 

32 pdeque([1, 2]) 

33 >>> x.pop() == x[:-1] 

34 True 

35 >>> x.popleft() 

36 pdeque([2, 3]) 

37 >>> x.append(4) 

38 pdeque([1, 2, 3, 4]) 

39 >>> x.appendleft(4) 

40 pdeque([4, 1, 2, 3]) 

41 

42 >>> y = pdeque([1, 2, 3], maxlen=3) 

43 >>> y.append(4) 

44 pdeque([2, 3, 4], maxlen=3) 

45 >>> y.appendleft(4) 

46 pdeque([4, 1, 2], maxlen=3) 

47 """ 

48 __slots__ = ('_left_list', '_right_list', '_length', '_maxlen', '__weakref__') 

49 

50 def __new__(cls, left_list, right_list, length, maxlen=None): 

51 instance = super(PDeque, cls).__new__(cls) 

52 instance._left_list = left_list 

53 instance._right_list = right_list 

54 instance._length = length 

55 

56 if maxlen is not None: 

57 if not isinstance(maxlen, Integral): 

58 raise TypeError('An integer is required as maxlen') 

59 

60 if maxlen < 0: 

61 raise ValueError("maxlen must be non-negative") 

62 

63 instance._maxlen = maxlen 

64 return instance 

65 

66 @property 

67 def right(self): 

68 """ 

69 Rightmost element in dqueue. 

70 """ 

71 return PDeque._tip_from_lists(self._right_list, self._left_list) 

72 

73 @property 

74 def left(self): 

75 """ 

76 Leftmost element in dqueue. 

77 """ 

78 return PDeque._tip_from_lists(self._left_list, self._right_list) 

79 

80 @staticmethod 

81 def _tip_from_lists(primary_list, secondary_list): 

82 if primary_list: 

83 return primary_list.first 

84 

85 if secondary_list: 

86 return secondary_list[-1] 

87 

88 raise IndexError('No elements in empty deque') 

89 

90 def __iter__(self): 

91 return chain(self._left_list, self._right_list.reverse()) 

92 

93 def __repr__(self): 

94 return "pdeque({0}{1})".format(list(self), 

95 ', maxlen={0}'.format(self._maxlen) if self._maxlen is not None else '') 

96 __str__ = __repr__ 

97 

98 @property 

99 def maxlen(self): 

100 """ 

101 Maximum length of the queue. 

102 """ 

103 return self._maxlen 

104 

105 def pop(self, count=1): 

106 """ 

107 Return new deque with rightmost element removed. Popping the empty queue 

108 will return the empty queue. A optional count can be given to indicate the 

109 number of elements to pop. Popping with a negative index is the same as 

110 popleft. Executes in amortized O(k) where k is the number of elements to pop. 

111 

112 >>> pdeque([1, 2]).pop() 

113 pdeque([1]) 

114 >>> pdeque([1, 2]).pop(2) 

115 pdeque([]) 

116 >>> pdeque([1, 2]).pop(-1) 

117 pdeque([2]) 

118 """ 

119 if count < 0: 

120 return self.popleft(-count) 

121 

122 new_right_list, new_left_list = PDeque._pop_lists(self._right_list, self._left_list, count) 

123 return PDeque(new_left_list, new_right_list, max(self._length - count, 0), self._maxlen) 

124 

125 def popleft(self, count=1): 

126 """ 

127 Return new deque with leftmost element removed. Otherwise functionally 

128 equivalent to pop(). 

129 

130 >>> pdeque([1, 2]).popleft() 

131 pdeque([2]) 

132 """ 

133 if count < 0: 

134 return self.pop(-count) 

135 

136 new_left_list, new_right_list = PDeque._pop_lists(self._left_list, self._right_list, count) 

137 return PDeque(new_left_list, new_right_list, max(self._length - count, 0), self._maxlen) 

138 

139 @staticmethod 

140 def _pop_lists(primary_list, secondary_list, count): 

141 new_primary_list = primary_list 

142 new_secondary_list = secondary_list 

143 

144 while count > 0 and (new_primary_list or new_secondary_list): 

145 count -= 1 

146 if new_primary_list.rest: 

147 new_primary_list = new_primary_list.rest 

148 elif new_primary_list: 

149 new_primary_list = new_secondary_list.reverse() 

150 new_secondary_list = plist() 

151 else: 

152 new_primary_list = new_secondary_list.reverse().rest 

153 new_secondary_list = plist() 

154 

155 return new_primary_list, new_secondary_list 

156 

157 def _is_empty(self): 

158 return not self._left_list and not self._right_list 

159 

160 def __lt__(self, other): 

161 if not isinstance(other, PDeque): 

162 return NotImplemented 

163 

164 return tuple(self) < tuple(other) 

165 

166 def __eq__(self, other): 

167 if not isinstance(other, PDeque): 

168 return NotImplemented 

169 

170 if tuple(self) == tuple(other): 

171 # Sanity check of the length value since it is redundant (there for performance) 

172 assert len(self) == len(other) 

173 return True 

174 

175 return False 

176 

177 def __hash__(self): 

178 return hash(tuple(self)) 

179 

180 def __len__(self): 

181 return self._length 

182 

183 def append(self, elem): 

184 """ 

185 Return new deque with elem as the rightmost element. 

186 

187 >>> pdeque([1, 2]).append(3) 

188 pdeque([1, 2, 3]) 

189 """ 

190 new_left_list, new_right_list, new_length = self._append(self._left_list, self._right_list, elem) 

191 return PDeque(new_left_list, new_right_list, new_length, self._maxlen) 

192 

193 def appendleft(self, elem): 

194 """ 

195 Return new deque with elem as the leftmost element. 

196 

197 >>> pdeque([1, 2]).appendleft(3) 

198 pdeque([3, 1, 2]) 

199 """ 

200 new_right_list, new_left_list, new_length = self._append(self._right_list, self._left_list, elem) 

201 return PDeque(new_left_list, new_right_list, new_length, self._maxlen) 

202 

203 def _append(self, primary_list, secondary_list, elem): 

204 if self._maxlen is not None and self._length == self._maxlen: 

205 if self._maxlen == 0: 

206 return primary_list, secondary_list, 0 

207 new_primary_list, new_secondary_list = PDeque._pop_lists(primary_list, secondary_list, 1) 

208 return new_primary_list, new_secondary_list.cons(elem), self._length 

209 

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

211 

212 @staticmethod 

213 def _extend_list(the_list, iterable): 

214 count = 0 

215 for elem in iterable: 

216 the_list = the_list.cons(elem) 

217 count += 1 

218 

219 return the_list, count 

220 

221 def _extend(self, primary_list, secondary_list, iterable): 

222 new_primary_list, extend_count = PDeque._extend_list(primary_list, iterable) 

223 new_secondary_list = secondary_list 

224 current_len = self._length + extend_count 

225 if self._maxlen is not None and current_len > self._maxlen: 

226 pop_len = current_len - self._maxlen 

227 new_secondary_list, new_primary_list = PDeque._pop_lists(new_secondary_list, new_primary_list, pop_len) 

228 extend_count -= pop_len 

229 

230 return new_primary_list, new_secondary_list, extend_count 

231 

232 def extend(self, iterable): 

233 """ 

234 Return new deque with all elements of iterable appended to the right. 

235 

236 >>> pdeque([1, 2]).extend([3, 4]) 

237 pdeque([1, 2, 3, 4]) 

238 """ 

239 new_right_list, new_left_list, extend_count = self._extend(self._right_list, self._left_list, iterable) 

240 return PDeque(new_left_list, new_right_list, self._length + extend_count, self._maxlen) 

241 

242 def extendleft(self, iterable): 

243 """ 

244 Return new deque with all elements of iterable appended to the left. 

245 

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

247 

248 >>> pdeque([1, 2]).extendleft([3, 4]) 

249 pdeque([4, 3, 1, 2]) 

250 """ 

251 new_left_list, new_right_list, extend_count = self._extend(self._left_list, self._right_list, iterable) 

252 return PDeque(new_left_list, new_right_list, self._length + extend_count, self._maxlen) 

253 

254 def count(self, elem): 

255 """ 

256 Return the number of elements equal to elem present in the queue 

257 

258 >>> pdeque([1, 2, 1]).count(1) 

259 2 

260 """ 

261 return self._left_list.count(elem) + self._right_list.count(elem) 

262 

263 def remove(self, elem): 

264 """ 

265 Return new deque with first element from left equal to elem removed. If no such element is found 

266 a ValueError is raised. 

267 

268 >>> pdeque([2, 1, 2]).remove(2) 

269 pdeque([1, 2]) 

270 """ 

271 try: 

272 return PDeque(self._left_list.remove(elem), self._right_list, self._length - 1) 

273 except ValueError: 

274 # Value not found in left list, try the right list 

275 try: 

276 # This is severely inefficient with a double reverse, should perhaps implement a remove_last()? 

277 return PDeque(self._left_list, 

278 self._right_list.reverse().remove(elem).reverse(), self._length - 1) 

279 except ValueError as e: 

280 raise ValueError('{0} not found in PDeque'.format(elem)) from e 

281 

282 def reverse(self): 

283 """ 

284 Return reversed deque. 

285 

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

287 pdeque([3, 2, 1]) 

288 

289 Also supports the standard python reverse function. 

290 

291 >>> reversed(pdeque([1, 2, 3])) 

292 pdeque([3, 2, 1]) 

293 """ 

294 return PDeque(self._right_list, self._left_list, self._length) 

295 __reversed__ = reverse 

296 

297 def rotate(self, steps): 

298 """ 

299 Return deque with elements rotated steps steps. 

300 

301 >>> x = pdeque([1, 2, 3]) 

302 >>> x.rotate(1) 

303 pdeque([3, 1, 2]) 

304 >>> x.rotate(-2) 

305 pdeque([3, 1, 2]) 

306 """ 

307 popped_deque = self.pop(steps) 

308 if steps >= 0: 

309 return popped_deque.extendleft(islice(self.reverse(), steps)) 

310 

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

312 

313 def __reduce__(self): 

314 # Pickling support 

315 return pdeque, (list(self), self._maxlen) 

316 

317 def __getitem__(self, index): 

318 if isinstance(index, slice): 

319 if index.step is not None and index.step != 1: 

320 # Too difficult, no structural sharing possible 

321 return pdeque(tuple(self)[index], maxlen=self._maxlen) 

322 

323 result = self 

324 if index.start is not None: 

325 result = result.popleft(index.start % self._length) 

326 if index.stop is not None: 

327 result = result.pop(self._length - (index.stop % self._length)) 

328 

329 return result 

330 

331 if not isinstance(index, Integral): 

332 raise TypeError("'%s' object cannot be interpreted as an index" % type(index).__name__) 

333 

334 if index >= 0: 

335 return self.popleft(index).left 

336 

337 shifted = len(self) + index 

338 if shifted < 0: 

339 raise IndexError( 

340 "pdeque index {0} out of range {1}".format(index, len(self)), 

341 ) 

342 return self.popleft(shifted).left 

343 

344 index = Sequence.index 

345 

346Sequence.register(PDeque) 

347Hashable.register(PDeque) 

348 

349 

350def pdeque(iterable=(), maxlen=None): 

351 """ 

352 Return deque containing the elements of iterable. If maxlen is specified then 

353 len(iterable) - maxlen elements are discarded from the left to if len(iterable) > maxlen. 

354 

355 >>> pdeque([1, 2, 3]) 

356 pdeque([1, 2, 3]) 

357 >>> pdeque([1, 2, 3, 4], maxlen=2) 

358 pdeque([3, 4], maxlen=2) 

359 """ 

360 t = tuple(iterable) 

361 if maxlen is not None: 

362 t = t[-maxlen:] 

363 length = len(t) 

364 pivot = int(length / 2) 

365 left = plist(t[:pivot]) 

366 right = plist(t[pivot:], reverse=True) 

367 return PDeque(left, right, length, maxlen) 

368 

369def dq(*elements): 

370 """ 

371 Return deque containing all arguments. 

372 

373 >>> dq(1, 2, 3) 

374 pdeque([1, 2, 3]) 

375 """ 

376 return pdeque(elements)