-
Notifications
You must be signed in to change notification settings - Fork 2
/
eth_sync_committee.py
87 lines (64 loc) · 2.87 KB
/
eth_sync_committee.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
#!/usr/bin/python
import argparse
from functions import (
get_user_validators, get_epochs, print_all_validators, stringify_list, generate_notification, add_cron_job,
write_log
)
if __name__ == '__main__':
parser = argparse.ArgumentParser()
parser.add_argument('-p', '--print-all', action='store_true')
parser.add_argument('-v', '--validators', nargs='*', action='store', type=str)
parser.add_argument(
'-c', '--cron', action='store_true',
help='run via crontab (adds a job to launch the script 2 epochs after the next sync committee starts)'
)
parser.add_argument(
'-n', '--notify', action='store_true',
help='send email notifications if a validator of yours is in the current or next sync committee'
)
args = parser.parse_args()
my_validators = get_user_validators(user_provided=args.validators)
epochs = get_epochs(my_validators=my_validators)
if my_validators:
check_str = f' checking validator{"s" if len(my_validators) > 1 else ""}: {stringify_list(my_validators)}'
else:
check_str = ' no validators specified.'
print(f'\n{check_str}')
longest_name = len(max([x.name_with_num for x in epochs], key=len))
longest_val_str = len(max([x.validators_str for x in epochs], key=len))
print(
f'\n {" " * longest_name} {"epoch":>7} {"start date & time":^19} validators\n'
f' {"-" * (longest_name + 33 + longest_val_str)}'
)
log_file_str = ''
for e in epochs:
epoch_str = (
f' {e.name_with_num:{longest_name}} {e.epoch_number:>7} '
f'{e.start_time.strftime("%Y/%m/%d %H:%M:%S")} '
f'{e.validators_str if e.is_sync_committee else "n/a"}'
)
if (e.name == 'current' and not e.is_sync_committee) or e.name == 'next':
log_file_str += f'{epoch_str}\n'
print(epoch_str)
current_committee = [e for e in epochs if e.name == 'current'][0]
next_committee = [e for e in epochs if e.name == 'next'][0]
if args.cron:
cron_str = add_cron_job(
next_start_time=next_committee.start_time,
in_next_committee=True if next_committee.validators else False
)
if cron_str:
print(f'\n{cron_str}')
log_file_str += f'{cron_str}\n'
log_file_str += '\n'
write_log(log_file_str=log_file_str)
if args.notify and current_committee.validators or next_committee.validators:
generate_notification(current_committee=current_committee, next_committee=next_committee)
if args.print_all:
for committee in [current_committee, next_committee]:
print(f'\n {committee.name_with_num}:\n')
if committee.all_validators:
print_all_validators(committee.all_validators)
else:
print(' validators in this committee are not yet known')
print()