Coverage for silkaj/money/tools.py: 98%

1# Copyright 2016-2025 Maël Azimi <m.a@moul.re> 

2# 

3# Silkaj is free software: you can redistribute it and/or modify 

4# it under the terms of the GNU Affero General Public License as published by 

5# the Free Software Foundation, either version 3 of the License, or 

6# (at your option) any later version. 

7# 

8# Silkaj is distributed in the hope that it will be useful, 

9# but WITHOUT ANY WARRANTY; without even the implied warranty of 

10# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 

11# GNU Affero General Public License for more details. 

12# 

13# You should have received a copy of the GNU Affero General Public License 

14# along with Silkaj. If not, see <https://www.gnu.org/licenses/>. 

15 

16import functools 

17from typing import Union 

18 

19from duniterpy.api import bma 

20from duniterpy.documents.transaction import InputSource, OutputSource 

21 

22from silkaj.network import client_instance 

23from silkaj.public_key import gen_pubkey_checksum 

24from silkaj.wot import tools as wt 

25 

26 

27def display_amount( 

28 tx: list, 

29 message: str, 

30 amount: float, 

31 ud_value: float, 

32 currency_symbol: str, 

33) -> None: 

34 """ 

35 Displays an amount in unit and relative reference. 

36 """ 

37 UD_amount = str(round((amount / ud_value), 2)) 

38 unit_amount = str(amount / 100) 

39 tx.append( 

40 [ 

41 f"{message} (unit|relative)", 

42 f"{unit_amount} {currency_symbol} | {UD_amount} UD {currency_symbol}", 

43 ], 

44 ) 

45 

46 

47def display_pubkey(tx: list, message: str, pubkey: str) -> None: 

48 """ 

49 Displays a pubkey and the eventually associated identity 

50 """ 

51 tx.append([f"{message} (pubkey:checksum)", gen_pubkey_checksum(pubkey)]) 

52 idty = wt.is_member(pubkey) 

53 if idty: 

54 tx.append([f"{message} (id)", idty["uid"]]) 

55 

56 

57def get_amount_from_pubkey(pubkey: str) -> list[int]: 

58 listinput, amount = get_sources(pubkey) 

59 

60 totalAmountInput = 0 

61 for _input in listinput: 

62 totalAmountInput += amount_in_current_base(_input) 

63 return [totalAmountInput, amount] 

64 

65 

66def get_sources(pubkey: str) -> tuple[list[InputSource], int]: 

67 client = client_instance() 

68 # Sources written into the blockchain 

69 sources = client(bma.tx.sources, pubkey) 

70 

71 listinput = [] 

72 amount = 0 

73 for source in sources["sources"]: 

74 if source["conditions"] == f"SIG({pubkey})": 

75 listinput.append( 

76 InputSource( 

77 amount=source["amount"], 

78 base=source["base"], 

79 source=source["type"], 

80 origin_id=source["identifier"], 

81 index=source["noffset"], 

82 ), 

83 ) 

84 amount += amount_in_current_base(listinput[-1]) 

85 

86 # pending source 

87 history = (client(bma.tx.pending, pubkey))["history"] 

88 pendings = history["sending"] + history["pending"] 

89 

90 # add pending output 

91 pending_sources = [] 

92 for pending in pendings: 

93 for i, output in enumerate(pending["outputs"]): 

94 # duniterpy#80 

95 outputsplited = output.split(":") 

96 if outputsplited[2] == f"SIG({pubkey})": 

97 inputgenerated = InputSource( 

98 amount=int(outputsplited[0]), 

99 base=int(outputsplited[1]), 

100 source="T", 

101 origin_id=pending["hash"], 

102 index=i, 

103 ) 

104 if inputgenerated not in listinput: 

105 # add pendings before blockchain sources for change txs 

106 listinput.insert(0, inputgenerated) 

107 

108 for _input in pending["inputs"]: 

109 pending_sources.append(InputSource.from_inline(_input)) 

110 

111 # remove input already used 

112 for _input in pending_sources: 

113 if _input in listinput: 

114 listinput.remove(_input) 

115 

116 return listinput, amount 

117 

118 

119@functools.lru_cache(maxsize=1) 

120def get_ud_value() -> int: 

121 client = client_instance() 

122 blockswithud = client(bma.blockchain.ud) 

123 NBlastUDblock = blockswithud["result"]["blocks"][-1] 

124 lastUDblock = client(bma.blockchain.block, NBlastUDblock) 

125 return lastUDblock["dividend"] * 10 ** lastUDblock["unitbase"] 

126 

127 

128def amount_in_current_base(source: Union[InputSource, OutputSource]) -> int: 

129 """ 

130 Get amount in current base from input or output source 

131 """ 

132 return source.amount * 10**source.base