Get validator duties (find largest gap in which to update for that $0.04 it will save/benefit you)

get_validator_duties.py

#!/usr/bin/env python3

""" This is a modified version of a script by pietjepuk2, forked by beemeeupnow to add:
* input parameters beyond validator indices
* fixes to work with more Beacon Node clients
* automatic fetch of genesis timestamp
* Gnosis compatibility

Original script by pietjepuk2 at: https://gist.github.com/pietjepuk2/eb021db978ad20bfd94dce485be63150
"""

import enum
import math
import sys

from datetime import datetime, timedelta

import requests

SLOTS_PER_EPOCH = "slots_per_epoch"
SECONDS_PER_SLOT = "seconds_per_slot"

SEPARATOR_WIDTH = 80


@enum.unique
class GenesisForkVersion(enum.IntEnum):
    MAINNET = 0
    PRATER = 4128
    GNOSIS = 100


BLOCK_TIMES = {
    GenesisForkVersion.MAINNET: {
        SLOTS_PER_EPOCH: 32,
        SECONDS_PER_SLOT: 12
    },
    GenesisForkVersion.GNOSIS: {
        SLOTS_PER_EPOCH: 16,
        SECONDS_PER_SLOT: 5
    },
    GenesisForkVersion.PRATER: {
        SLOTS_PER_EPOCH: 32,
        SECONDS_PER_SLOT: 12
    }
}

assert all(item in BLOCK_TIMES for item in GenesisForkVersion)
assert len(GenesisForkVersion) == len(BLOCK_TIMES)
assert all(key in item for key in (SLOTS_PER_EPOCH, SECONDS_PER_SLOT) for item in BLOCK_TIMES.values())


def main(validators_indices, eth2_api_scheme, eth2_api_host, eth2_api_port):
    eth2_api_url = f"{eth2_api_scheme}://{eth2_api_host}:{eth2_api_port}/eth/v1/"

    def api_get(endpoint: str):
        """send GET request"""
        response = requests.get(f"{eth2_api_url}{endpoint}")
        return response.json()

    def api_post(endpoint: str, data=None, json_data=None):
        """send POST request

        Provide non-JSON data for data, OR
        Provide a JSON-serializable Python object for json_data (automatically sets the Content-Type header properly)
        """
        if data and json_data:
            raise ValueError("provide only data OR json_data")

        if data:
            response = requests.post(f"{eth2_api_url}{endpoint}", data)
        elif json_data:
            response = requests.post(f"{eth2_api_url}{endpoint}", json=json_data)
        else:
            raise ValueError("No data provided")

        return response.json()

    genesis_data = api_get("beacon/genesis")

    genesis_timestamp = int(genesis_data["data"]["genesis_time"])
    genesis_fork_version = GenesisForkVersion(int(genesis_data["data"]["genesis_fork_version"], 16))

    if genesis_fork_version in BLOCK_TIMES:
        slots_per_epoch = BLOCK_TIMES[genesis_fork_version][SLOTS_PER_EPOCH]
        seconds_per_slot = BLOCK_TIMES[genesis_fork_version][SECONDS_PER_SLOT]
    else:
        print(f"block times not found for chain. genesis_fork_version: {genesis_fork_version}")
        sys.exit(1)

    head_slot = int(api_get("beacon/headers/head")["data"]["header"]["message"]["slot"])
    epoch = head_slot // slots_per_epoch

    cur_epoch_data = api_post(f"validator/duties/attester/{epoch}", json_data=validators_indices)["data"]
    next_epoch_data = api_post(f"validator/duties/attester/{epoch + 1}", json_data=validators_indices)["data"]

    attestation_duties = {}
    for d in (*cur_epoch_data, *next_epoch_data):
        attestation_duties.setdefault(int(d["slot"]), []).append(d["validator_index"])
    attestation_duties = {k: v for k, v in sorted(attestation_duties.items()) if k > head_slot}

    # Also insert (still unknown) attestation duties at epoch after next,
    # assuming worst case of having to attest at its first slot
    first_slot_epoch_p2 = (epoch + 2) * slots_per_epoch
    attestation_duties[first_slot_epoch_p2] = []

    print(f"Calculating attestation slots and gaps for validators:")
    print(f"  {validators_indices}")

    print("\nUpcoming voting slots and gaps")
    print("(Gap in seconds)")
    print("(slot/epoch - time range - validators)")
    print("*" * SEPARATOR_WIDTH)

    prev_end_time = datetime.now()
    longest_gap = timedelta(seconds=0)
    gap_time_range = ()

    for slot, validators in attestation_duties.items():
        slot_start = datetime.fromtimestamp(genesis_timestamp + slot * seconds_per_slot)
        slot_end = slot_start + timedelta(seconds=seconds_per_slot)

        gap = slot_start - prev_end_time
        print(f"Gap - {math.floor((slot_start - prev_end_time).total_seconds())} seconds")

        if validators:
            print(
                f"  {slot}/{slot // slots_per_epoch}"
                f" - {slot_start.strftime('%H:%M:%S')} until {slot_end.strftime('%H:%M:%S')}"
                f" - [{', '.join(validators)}]"
            )
        else:
            assert slot % slots_per_epoch == 0

        if gap > longest_gap:
            longest_gap = gap
            gap_time_range = (prev_end_time, slot_start)

        prev_end_time = slot_end

    print("\nLongest gap (first):")
    print("*" * SEPARATOR_WIDTH)
    print(
        f"{longest_gap.total_seconds()} seconds"
        f" ({int(longest_gap.total_seconds()) // seconds_per_slot} slots),"
        f" from {gap_time_range[0].strftime('%H:%M:%S') if gap_time_range else 'ERROR'}"
        f" until {gap_time_range[1].strftime('%H:%M:%S') if gap_time_range else 'ERROR'}"
    )


if __name__ == "__main__":
    import argparse

    parser = argparse.ArgumentParser(
        description="Show validator duties of current and next epoch to find largest gap."
    )
    parser.add_argument("--host", metavar="HOST", type=str, default="localhost", help="Beacon Node hostname or IP")
    parser.add_argument("--port", metavar="PORT", type=int, default=5052, help="Beacon Node port")
    parser.add_argument("--scheme", type=str, default="http", choices=("http", "https"), help="protocol")
    parser.add_argument("indices", metavar="index", type=int, nargs="+", help="validator indices")

    args = parser.parse_args()
    # strings are required for the API request, only using int in argparse for input validation purposes
    index_strings = [str(index) for index in args.indices]

    main(index_strings, eth2_api_scheme=args.scheme, eth2_api_host=args.host, eth2_api_port=args.port)