Coverage for silkaj/blockchain/blocks.py: 21%

1# Copyright 2016-2024 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 time 

17from collections import OrderedDict 

18from operator import itemgetter 

19from urllib.error import HTTPError 

20 

21import rich_click as click 

22from duniterpy.api import bma 

23from pendulum import from_timestamp 

24 

25from silkaj import tui 

26from silkaj.blockchain.tools import get_head_block 

27from silkaj.constants import ALL, BMA_SLEEP 

28from silkaj.network import client_instance 

29from silkaj.wot.tools import identity_of 

30 

31 

32@click.command("blocks", help="Display blocks: default: 0 for current window size") 

33@click.argument("number", default=0, type=click.IntRange(0, 5000)) 

34@click.option( 

35 "--detailed", 

36 "-d", 

37 is_flag=True, 

38 help="Force detailed view. Compact view happen over 30 blocks", 

39) 

40def list_blocks(number: int, detailed: bool) -> None: 

41 head_block = get_head_block() 

42 current_nbr = head_block["number"] 

43 if number == 0: 

44 number = head_block["issuersFrame"] 

45 client = client_instance() 

46 blocks = client(bma.blockchain.blocks, number, current_nbr - number + 1) 

47 issuers = [] 

48 issuers_dict = {} 

49 for block in blocks: 

50 issuer = OrderedDict() 

51 issuer["pubkey"] = block["issuer"] 

52 if detailed or number <= 30: 

53 gentime = from_timestamp(block["time"], tz="local").format(ALL) 

54 mediantime = from_timestamp(block["medianTime"], tz="local").format(ALL) 

55 issuer["block"] = block["number"] 

56 issuer["gentime"] = gentime 

57 issuer["mediantime"] = mediantime 

58 issuer["hash"] = block["hash"][:10] 

59 issuer["powMin"] = block["powMin"] 

60 issuers_dict[issuer["pubkey"]] = issuer 

61 issuers.append(issuer) 

62 for pubkey in issuers_dict: 

63 issuer = issuers_dict[pubkey] 

64 time.sleep(BMA_SLEEP) 

65 try: 

66 idty = identity_of(issuer["pubkey"]) 

67 except HTTPError: 

68 idty = None 

69 for issuer2 in issuers: 

70 if ( 

71 issuer2.get("pubkey") is not None 

72 and issuer.get("pubkey") is not None 

73 and issuer2["pubkey"] == issuer["pubkey"] 

74 ): 

75 issuer2["uid"] = idty["uid"] if idty else None 

76 issuer2.pop("pubkey") 

77 print_blocks_views(issuers, current_nbr, number, detailed) 

78 

79 

80def print_blocks_views(issuers, current_nbr, number, detailed): 

81 header = ( 

82 f"Last {number} blocks from n°{current_nbr - number + 1} to n°{current_nbr}" 

83 ) 

84 print(header, end=" ") 

85 if detailed or number <= 30: 

86 sorted_list = sorted(issuers, key=itemgetter("block"), reverse=True) 

87 

88 table = tui.Table(style="columns") 

89 table.set_cols_align(["r", "r", "r", "r", "r", "l"]) 

90 table.set_cols_dtype(["i", "t", "t", "t", "i", "t"]) 

91 table.fill_from_dict_list(sorted_list) 

92 table.set_cols_align(["r", "r", "r", "r", "r", "l"]) 

93 table.set_cols_dtype(["i", "t", "t", "t", "i", "t"]) 

94 print(f"\n{table.draw()}") 

95 

96 else: 

97 list_issued = [] 

98 for issuer in issuers: 

99 found = False 

100 for issued in list_issued: 

101 if issued.get("uid") is not None and issued["uid"] == issuer["uid"]: 

102 issued["blocks"] += 1 

103 found = True 

104 break 

105 if not found: 

106 issued = OrderedDict() 

107 issued["uid"] = issuer["uid"] 

108 issued["blocks"] = 1 

109 list_issued.append(issued) 

110 for issued in list_issued: 

111 issued["percent"] = round(issued["blocks"] / number * 100) 

112 sorted_list = sorted(list_issued, key=itemgetter("blocks"), reverse=True) 

113 table = tui.Table(style="columns") 

114 table.fill_from_dict_list(sorted_list) 

115 table.set_cols_align(["l", "r", "r"]) 

116 table.set_cols_dtype(["t", "i", "i"]) 

117 print(f"from {len(list_issued)} issuers\n{table.draw()}")