Coverage for src/pyfplib/functools.py: 100%

1# SPDX-FileCopyrightText: 2026-present Comet11x 

2# SPDX-License-Identifier: MIT 

3""" 

4functools.py 

5 

6This module provides a set of functions. 

7 

8Example usage: 

9 

10 from pyfplib import 

11 result = add(5, 3) 

12 

13Functions: 

14- add(a, b): Returns the sum of a and b. 

15- subtract(a, b): Returns the difference of a and b. 

16- multiply(a, b): Returns the product of a and b. 

17- divide(a, b): Returns the quotient of a and b. Raises ValueError for division by zero. 

18""" 

19 

20import inspect 

21from collections.abc import Callable, Iterable, Sequence, Sized 

22from typing import Any, cast, overload 

23 

24from pyfplib.option import Nothing, Option, Some 

25 

26 

27class _FnWrapper[T, R]: 

28 # Number of parameters accepted by the wrapped callback. 

29 ONE = 1 

30 TWO = 2 

31 THREE = 3 

32 FORE = 4 

33 

34 def __init__(self, fn: Callable[..., R]) -> None: 

35 # Inspect the callback once so calls can dispatch on its arity. 

36 self._fn = fn 

37 self._sig = inspect.signature(fn) 

38 self._params_num = len(self._sig.parameters) 

39 

40 @property 

41 def signature(self) -> inspect.Signature: 

42 return self._sig 

43 

44 @property 

45 def params_number(self) -> int: 

46 return self._params_num 

47 

48 def __call__(self, v: T, i: int = 0, c: Sequence[T] | None = None) -> R: 

49 # Forward only the arguments the wrapped callback declares. 

50 if self._params_num == _FnWrapper.ONE: 

51 return self._fn(v) 

52 elif self._params_num == _FnWrapper.TWO: 

53 return self._fn(v, i) 

54 else: 

55 c = c if c is not None else () 

56 return self._fn(v, i, c) 

57 

58 

59class _FnWrapper2[T, R](_FnWrapper[T, R]): 

60 def __init__(self, fn: Callable[..., R], *, pipeline: Iterable[Callable[..., R]] = []) -> None: 

61 super().__init__(fn) 

62 self._pipeline = pipeline 

63 

64 def __call__(self, v: T, i: int = 0, c: Sequence[T] | None = None) -> R: 

65 # Run the base callback, then pass its result through each pipeline step. 

66 ret_value = super().__call__(v, i, c) 

67 for fn in self._pipeline: 

68 ret_value = fn(ret_value) 

69 return ret_value 

70 

71 

72class _FnWrapper3[T, R]: 

73 # Like _FnWrapper, but keeps the accumulated value as the first argument. 

74 def __init__(self, fn: Callable[..., R]) -> None: 

75 self._fn = fn 

76 self._sig = inspect.signature(fn) 

77 self._params_num = len(self._sig.parameters) 

78 

79 @property 

80 def params_number(self) -> int: 

81 return self._params_num 

82 

83 def __call__(self, p: R | T | None, v: T, i: int = 0, c: Sequence[T] | None = None) -> R: 

84 # Dispatch on arity, always treating `p` as the accumulated value. 

85 if self._params_num == _FnWrapper.ONE: 

86 return self._fn(v) 

87 elif self._params_num == _FnWrapper.TWO: 

88 return self._fn(p, v) 

89 elif self._params_num == _FnWrapper.THREE: 

90 return self._fn(p, v, i) 

91 else: 

92 return self._fn(p, v, i, c) 

93 

94 

95@overload 

96def for_each[T](function: Callable[[T], None], sequence: Sequence[T]) -> None: ... 

97 

98 

99@overload 

100def for_each[T](function: Callable[[T, int], None], sequence: Sequence[T]) -> None: ... 

101 

102 

103def for_each[T](function: Callable[[T, int, Sequence[T]], None], sequence: Sequence[T]) -> None: 

104 """Applies the given function for each elements of the given sequence object 

105 

106 Args: 

107 function: a callback to execute with elements 

108 sequence: provides elements to apply the callback 

109 

110 Returns: 

111 None 

112 """ 

113 if is_empty(sequence): 

114 return 

115 

116 wrapper = _FnWrapper(function) 

117 

118 # Pass the index (and the sequence) only when the callback accepts them. 

119 if wrapper.params_number == _FnWrapper.ONE: 

120 for item in sequence: 

121 wrapper(item) 

122 else: 

123 for idx, item in enumerate(sequence): 

124 wrapper(item, idx, sequence) 

125 

126 

127@overload 

128def any_of[T](predicate: Callable[[T], bool], sequence: Sequence[T]): ... 

129 

130 

131@overload 

132def any_of[T](predicate: Callable[[T, int], bool], sequence: Sequence[T]): ... 

133 

134 

135def any_of[T](predicate: Callable[[T, int, Sequence[T]], bool], sequence: Sequence[T]) -> bool: 

136 """ 

137 Checks if any elements of an iterable object satisfy a given condition. 

138 

139 It returns false if an iterable oject is empty. 

140 

141 Args: 

142 predicate: A function checking elements. 

143 iterable: An iterable object providing elements for checking. 

144 

145 Returns: 

146 bool: The result of the checking. 

147 """ 

148 # Stop as soon as the predicate returns a truthy value. 

149 return _abstract_check(predicate=_FnWrapper(predicate), sequence=sequence, init_value=False, check=lambda v: v) 

150 

151 

152@overload 

153def all_of[T](predicate: Callable[[T], bool], iterable: Iterable[T]): ... 

154 

155 

156@overload 

157def all_of[T](predicate: Callable[[T, int], bool], iterable: Iterable[T]): ... 

158 

159 

160def all_of[T](predicate: Callable[[T, int, Sequence[T]], bool], sequence: Sequence[T]) -> bool: 

161 """ 

162 Checks if every element of an iterable object satisfy a given condition. 

163 

164 It returns true if an iterable object is empty. 

165 

166 Args: 

167 predicate: a function checking elements. 

168 iterable: an iterable object providing elements for checking. 

169 

170 Returns: 

171 bool: The result of the checking. 

172 """ 

173 # Stop as soon as the predicate returns a falsy value. 

174 return _abstract_check(predicate=_FnWrapper(predicate), sequence=sequence, init_value=True, check=lambda v: not v) 

175 

176 

177@overload 

178def none_of[T](predicate: Callable[[T], bool], iterable: Iterable[T]): ... 

179 

180 

181@overload 

182def none_of[T](predicate: Callable[[T, int], bool], iterable: Iterable[T]): ... 

183 

184 

185def none_of[T](callback: Callable[[T], bool], sequence: Sequence[T]) -> bool: 

186 """ 

187 Checks if no one element of an iterable object not satisfy a given condition. 

188 

189 It returns true if an iterable object is empty. 

190 

191 Args: 

192 predicate: a function checking elements. 

193 sequence: a sequence object providing elements for checking. 

194 

195 Returns: 

196 bool: The result of the checking. 

197 """ 

198 # Invert the callback's result via the pipeline, then stop on the first match. 

199 return _abstract_check( 

200 predicate=_FnWrapper2(callback, pipeline=(lambda v: not v,)), 

201 sequence=sequence, 

202 init_value=True, 

203 check=lambda v: not v, 

204 ) 

205 

206 

207# @overload 

208# def fold[T]( 

209# function: Callable[[T, T], T], 

210# sequence: Sequence[T], 

211# initial: T | None = None, 

212# ) -> Option[T]: ... 

213 

214 

215def fold[T, R]( 

216 function: Callable[[Option[R], T], R] | Callable[[Option[R], T, int], R], 

217 sequence: Sequence[T], 

218 initial: R | None = None, 

219) -> Option[R]: 

220 """ 

221 Folds the sequence into a single accumulated value. 

222 

223 Args: 

224 function: Callback receiving the accumulator and the current item, and optionally its index. 

225 sequence: Elements to fold over. 

226 initial: Optional starting value for the accumulator. 

227 

228 Returns: 

229 Option[R]: The final accumulator value, or Nothing if the sequence is empty and no initial value is given. 

230 """ 

231 # Seed the accumulator with the optional initial value. 

232 ret_value = Option.from_optional(initial) 

233 if len(sequence) != 0: 

234 wrapper = _FnWrapper3(function) 

235 # Two-parameter callbacks receive the accumulator and the current item. 

236 if wrapper.params_number == _FnWrapper.TWO: 

237 fn = cast(Callable[[Option[R], T], R], function) 

238 for item in sequence: 

239 ret_value.push(fn(ret_value, item)) 

240 # Three-parameter callbacks additionally receive the index. 

241 elif wrapper.params_number == _FnWrapper.THREE: 

242 for idx, item in enumerate(sequence): 

243 ret_value.push(wrapper(ret_value, item, idx)) 

244 return ret_value 

245 

246 

247def head[T](iterable: Iterable[T]) -> Option[T]: 

248 """ 

249 Returns Some[T] contained 1st item of the given iterable object. 

250 If the given iterable is empty, it returns Nothing[T]. 

251 

252 Args: 

253 iterable: an iterable object provided 1st element. 

254 

255 Return: 

256 Option[T]: Returns Some(item) if non-empty, otherwise Nothing(). 

257 """ 

258 # Use a unique sentinel so a legitimate `None` element is not mistaken for "empty". 

259 iterator = iter(iterable) 

260 sentinel = object() 

261 item = next(iterator, sentinel) 

262 return Nothing() if item is sentinel else Some(cast(T, item)) 

263 

264 

265def tail[T](sequence: Sequence[T]) -> Sequence[T]: 

266 """ 

267 Returns Some[T] contained a slice of the given sequence without 1st element, 

268 i.g. it returns sequence[1:]. 

269 If the sequence is empty, this function returns it. 

270 

271 Args: 

272 sequence: a sequence used to get a slice. 

273 

274 Return: 

275 Sequence[T]: A slice of the given sequence. 

276 """ 

277 return sequence[1:] if sequence else sequence 

278 

279 

280def split_first[T](sequence: Sequence[T]) -> tuple[Option[T], Sequence[T]]: 

281 """ 

282 Splits a sequence into its first element (if any) and the remaining tail. 

283 

284 Args: 

285 sequence: Sequence to split. 

286 

287 Returns: 

288 tuple[Option[T], Sequence[T]]: The first element as an Option and the remaining items. 

289 """ 

290 return (Some(sequence[0]), sequence[1:]) if sequence else (Nothing(), sequence) 

291 

292 

293def split_last[T](sequence: Sequence[T]) -> tuple[Sequence[T], Option[T]]: 

294 """ 

295 Splits a sequence into its initial part and its last element (if any). 

296 

297 Args: 

298 sequence: Sequence to split. 

299 

300 Returns: 

301 tuple[Sequence[T], Option[T]]: The initial items and the last element as an Option. 

302 """ 

303 return (sequence[:-1], Some(sequence[-1])) if sequence else (sequence, Nothing()) 

304 

305 

306def first[T](iterable: Iterable[T]) -> Option[T]: 

307 """ 

308 Returns Some[T] contained 1st item of the given iterable object. 

309 If the given iterable is empty, it returns Nothing[T]. 

310 

311 Args: 

312 iterable: an iterable object provided 1st element. 

313 

314 Return: 

315 Option[T]: Returns Some(item) if non-empty, otherwise Nothing(). 

316 """ 

317 iterator = iter(iterable) 

318 sentinel = object() 

319 item = next(iterator, sentinel) 

320 return Nothing() if item is sentinel else Some(cast(T, item)) 

321 

322 

323def last[T](sequence: Sequence[T]) -> Option[T]: 

324 """ 

325 Returns Some[Any] contained last item of the given sequence. 

326 If the given sequence is empty, it returns Nothing[Any]. 

327 

328 Args: 

329 sequence: a sequence provided last element. 

330 

331 Return: 

332 Option[T]: Returns Some(item) if non-empty, otherwise Nothing(). 

333 """ 

334 return Some(sequence[-1]) if sequence else Nothing() 

335 

336 

337def is_empty(iterable: Sized) -> bool: 

338 """ 

339 Checks if the given iterable object is empty. 

340 

341 Args: 

342 iterable: an iterable object. 

343 

344 Return: 

345 bool: Returns true if the iterable is empty, otherwise returns false. 

346 """ 

347 return len(iterable) == 0 

348 

349 

350def is_not_empty(value: Sized) -> bool: 

351 """ 

352 Checks whether the given object of Sized is not empty. 

353 

354 Args: 

355 Sized: an sized object. 

356 

357 Return: 

358 bool: Returns true if the value is not empty, otherwise returns false. 

359 """ 

360 return len(value) != 0 

361 

362 

363def index[T](predicate: Callable[..., bool | Option[Any]] | T, sequence: Sequence[T]) -> Option[int]: 

364 """ 

365 Finds the position of 1st occurrence of a specified element in the given iterable object. 

366 It returns Some[Any] contained the index of the found element, and if the element is not found, it returns Nothing 

367 

368 Args: 

369 predicate: a function checking elements or a value used for checking. 

370 iterable: an iterable object providing elements for checking. 

371 

372 Return: 

373 Option[int]: A found index 

374 """ 

375 if is_not_empty(sequence): 

376 if callable(predicate): 

377 wrapper = _FnWrapper(predicate) 

378 for idx, value in enumerate(sequence): 

379 if wrapper(value, idx, sequence): 

380 return Some(idx) 

381 else: 

382 expected_value: T = cast(T, predicate) 

383 

384 for idx, value in enumerate(sequence): 

385 if expected_value == value: 

386 return Some(idx) 

387 return Nothing() 

388 

389 

390def find[T](callback: Callable[[T], Option[T]], sequence: Sequence[T]) -> Option[T]: 

391 """ 

392 Finds a specified element in the given iterable object. 

393 It returns Some[Any] contained f of the found element, and if the element is not found, it returns Nothing 

394 

395 Args: 

396 predicate: a function checking elements or a value used for checking. 

397 sequence: a sequence object providing elements for checking. 

398 

399 Return: 

400 Option[int]: A found index 

401 """ 

402 for item in sequence: 

403 ret_value = callback(item) 

404 # Accept either a non-empty Option or any truthy value as a match. 

405 if isinstance(ret_value, Option) and ret_value.is_some(): 

406 return ret_value 

407 elif ret_value: 

408 return Some(item) 

409 

410 return Nothing() 

411 

412 

413type _FnWrapperType[T, R] = _FnWrapper[T, R] | _FnWrapper2[T, R] 

414 

415 

416def _abstract_check[T, bool]( 

417 *, 

418 predicate: _FnWrapperType[T, bool], 

419 sequence: Sequence[T], 

420 init_value: bool, 

421 check: Callable[[bool], bool], 

422): 

423 ret_value = init_value 

424 # Empty input keeps the identity value (False for any_of, True for all_of/none_of). 

425 if is_empty(sequence): 

426 return ret_value 

427 

428 # Stop early when `check` flags the predicate result as decisive. 

429 if predicate.params_number == _FnWrapper.ONE: 

430 for item in sequence: 

431 ret_value = predicate(item) 

432 if check(ret_value): 

433 break 

434 else: 

435 for idx, item in enumerate(sequence): 

436 ret_value = predicate(item, idx, sequence) 

437 if check(ret_value): 

438 break 

439 

440 return ret_value