PVA node checker

pva-node-checker.py

#!/usr/bin/env python
# coding: utf-8

import json
import requests
from requests.exceptions import HTTPError
import re
import argparse

def get_information_rpc(rpc_endpoint, method, params):
    url = rpc_endpoint
    headers = {'Content-Type': 'application/json'}
    data = {"jsonrpc":"2.0", "method": method, "params": params, "id":1}
    try:
        r = requests.post(url, headers=headers, data = json.dumps(data))
        r.raise_for_status()
    except HTTPError as http_err:
            print(f'HTTP error occurred: {http_err}')  # Python 3.6
    except Exception as err:
            print(f'Other error occurred: {err}')  # Python 3.6
    else:
        content = json.loads(r.content)
        return content

def getNodemetadata_rpcurl(rpcurl):
    method = 'hmyv2_getNodeMetadata'
    params = []
    return get_information_rpc(rpcurl, method, params)['result']


def GetLatestChainHeaders_rpcurl(rpcurl):
    method = "hmy_getLatestChainHeaders"
    params = []
    return get_information_rpc(rpcurl, method, params)['result']

def IsInSynced(nodeurl = "http://127.0.0.1:9500"):

    nodeinfo = getNodemetadata_rpcurl(nodeurl)
    shardid = nodeinfo['shard-id']
    network = nodeinfo['network']

    #print (shardid)

    if (network == "mainnet"):
        endpoint=f"https://api.s{shardid}.t.hmny.io"
    if (network == "testnet"):
        endpoint=f"https://api.s{shardid}.b.hmny.io"
    
    latest_headers = GetLatestChainHeaders_rpcurl(nodeurl)
    node_bc_shard_block_height = latest_headers['beacon-chain-header']['block-number']
    node_non_bc_shard_block_height = latest_headers['shard-chain-header']['block-number']

    network_latest_header = GetLatestChainHeaders_rpcurl(endpoint)
    network_bc_shard_block_height = network_latest_header['beacon-chain-header']['block-number']
    network_non_bc_shard_block_height = network_latest_header['shard-chain-header']['block-number']

    print (f"  Node (s{shardid}) block info [{node_bc_shard_block_height}, {node_non_bc_shard_block_height}]")
    print (f"  Network block info [{network_bc_shard_block_height}, {network_non_bc_shard_block_height}]")

    bc_diff = abs(node_bc_shard_block_height - network_bc_shard_block_height) 
    non_bc_diff = abs(node_non_bc_shard_block_height - network_non_bc_shard_block_height)

    bc_sync_status = True if bc_diff < 100 else False
    non_bc_sync_status = True if non_bc_diff < 100 else False

    print (f"  {bc_sync_status}({bc_diff}) {non_bc_sync_status}({non_bc_diff})")

    if (bc_sync_status and non_bc_sync_status):
        return True
    else:
        return False

def CheckVersion(nodeurl = "http://127.0.0.1:9500", version = "v6268-v2.3.5"):
    nodeversion = getNodemetadata_rpcurl(nodeurl)['version']
    if re.search(version, nodeversion) != None:
        return True
    return False

def argsparse():
   parser = argparse.ArgumentParser(description="POPS PVA Checker")
   parser.add_argument( "-vc", "--versioncheck", 
                        help="harmony binary version to check",
                        default="v6268-v2.3.4")
   parser.add_argument( "-n", "--node", help="Validator Node URL (eg: http://1.1.1.1:9500)",
                        default="http://127.0.0.1:9500")
#    parser.add_argument( "-vla", "--validator_addr", 
#                         help="validator address the node is helping")                  
   args = parser.parse_args()
   return args

if __name__ == "__main__":

    args = argsparse()
    version = args.versioncheck
    node_rpc = args.node
    #validators = args.validator_addr

    print("PVA node Checker")

    version_result = CheckVersion(node_rpc, version)
    print(f"Version expected ({version}) : {version_result}")

    nodesync_result = IsInSynced(node_rpc)
    print(f"Node synced: {nodesync_result}")