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

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 

16from collections import OrderedDict 

17from operator import itemgetter 

18from os import system 

19from typing import Dict, Tuple 

20 

21import jsonschema 

22import rich_click as click 

23from duniterpy.api import bma 

24from duniterpy.api.client import WSConnection 

25from pendulum import from_timestamp 

26from websocket._exceptions import WebSocketConnectionClosedException 

27 

28from silkaj import network, tui 

29from silkaj.constants import ALL 

30 

31 

32@click.command( 

33 "difficulty", 

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

35) 

36def difficulties() -> None: 

37 client = network.client_instance() 

38 try: 

39 ws = client(bma.ws.block) 

40 while True: 

41 current = ws.receive_json() 

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

43 diffi = client(bma.blockchain.difficulties) 

44 display_diffi(current, diffi) 

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

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

47 

48 

49def display_diffi(current: WSConnection, diffi: Dict) -> None: 

50 levels = [OrderedDict((i, d[i]) for i in ("uid", "level")) for d in diffi["levels"]] 

51 diffi["levels"] = levels 

52 issuers = 0 

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

54 for d in diffi["levels"]: 

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

56 issuers += 1 

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

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

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

60 system("cls||clear") 

61 block_gen = from_timestamp(current["time"], tz="local").format(ALL) 

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

63 

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

65 table.fill_from_dict_list(sorted_diffi) 

66 

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

68 \n\ 

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

70 \ 

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

72 members\n\ 

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

74 `\n\ 

75{table.draw()}' 

76 print(content) 

77 

78 

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

80 while _pow > 0: 

81 if _pow >= 16: 

82 match += "0" 

83 _pow -= 16 

84 p *= 16 

85 else: 

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

87 p *= _pow 

88 _pow = 0 

89 return f"{match}*", p 

90 

91 

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

93 while nbr >= 10: 

94 nbr /= 10 

95 power += 1 

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