Coverage for silkaj/blockchain/difficulty.py: 28%

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 

16from operator import itemgetter 

17from os import system 

18 

19import arrow 

20import jsonschema 

21import rich_click as click 

22from duniterpy.api import bma 

23from duniterpy.api.client import WSConnection 

24from websocket._exceptions import WebSocketConnectionClosedException 

25 

26from silkaj import network, tui 

27from silkaj.constants import ALL 

28 

29 

30@click.command( 

31 "difficulty", 

32 help="Display the current Proof of Work difficulty level to generate the next block", 

33) 

34def difficulties() -> None: 

35 client = network.client_instance() 

36 try: 

37 ws = client(bma.ws.block) 

38 while True: 

39 current = ws.receive_json() 

40 jsonschema.validate(current, bma.ws.WS_BLOCK_SCHEMA) 

41 diffi = client(bma.blockchain.difficulties) 

42 display_diffi(current, diffi) 

43 except (jsonschema.ValidationError, WebSocketConnectionClosedException) as e: 

44 print(f"{e.__class__.__name__!s}: {e!s}") 

45 

46 

47def display_diffi(current: WSConnection, diffi: dict) -> None: 

48 issuers = 0 

49 sorted_diffi = sorted(diffi["levels"], key=itemgetter("level"), reverse=True) 

50 for d in diffi["levels"]: 

51 if d["level"] / 2 < current["powMin"]: 

52 issuers += 1 

53 d["match"] = match_pattern(d["level"])[0][:20] 

54 d["Π diffi"] = compute_power(match_pattern(d["level"])[1]) 

55 d["Σ diffi"] = d.pop("level") 

56 system("cls||clear") 

57 block_gen = arrow.get(current["time"]).to("local").format(ALL) 

58 match = match_pattern(int(current["powMin"]))[0] 

59 

60 table = tui.Table(style="columns").set_cols_dtype(["t", "t", "t", "i"]) 

61 table.fill_from_dict_list(sorted_diffi) 

62 

63 content = f'Current block: n°{current["number"]}, generated on {block_gen}\n\ 

64Generation of next block n°{diffi["block"]} \ 

65possible by at least {issuers}/{len(diffi["levels"])} members\n\ 

66Common Proof-of-Work difficulty level: {current["powMin"]}, hash starting with `{match}`\n\ 

67{table.draw()}' 

68 print(content) 

69 

70 

71def match_pattern(_pow: int, match: str = "", p: int = 1) -> tuple[str, int]: 

72 while _pow > 0: 

73 if _pow >= 16: 

74 match += "0" 

75 _pow -= 16 

76 p *= 16 

77 else: 

78 match += f"[0-{hex(15 - _pow)[2:].upper()}]" 

79 p *= _pow 

80 _pow = 0 

81 return f"{match}*", p 

82 

83 

84def compute_power(nbr: float, power: int = 0) -> str: 

85 while nbr >= 10: 

86 nbr /= 10 

87 power += 1 

88 return f"{nbr:.1f} x 10^{power}"