Skip to content

Instantly share code, notes, and snippets.

Created September 29, 2019 15:18
Show Gist options
  • Save utdrmac/b3234921d2376dc550798e122d43f931 to your computer and use it in GitHub Desktop.
Save utdrmac/b3234921d2376dc550798e122d43f931 to your computer and use it in GitHub Desktop.
check current block head level from multiple sources
#!/bin/env python36
import requests
import json
from time import sleep
from datetime import timedelta, datetime as dt
from requests.exceptions import Timeout, HTTPError, ConnectionError
from concurrent.futures import TimeoutError, ThreadPoolExecutor
nodes = [
["Local", ''],
["TZBeta", ''],
## ["Obsidian", ''],
def getLevel(node):
name = node[0]
url = node[1]
resp = requests.get(url, timeout=10)
except Timeout:
return [name, 0,, "(T)"]
except HTTPError as e:
return [name, 0,, "(E) {}".format(e.response.status_code)]
except ConnectionError as e:
return [name, 0,, "(CE)"]
js = resp.json()
#print("JSON: %s", js)
#print("Level: {}".format(js['level']))
# Make datetime object from Tezos timestamp, 2019-07-02T19:06:30Z
mdt = dt.strptime(js['timestamp'], "%Y-%m-%dT%H:%M:%SZ")
# returning tuple
return [name, js['level'], mdt]
def getLevels():
global nodes
levels = []
with ThreadPoolExecutor(max_workers=3) as exec:
results =, nodes, timeout=15)
# Get node levels
for i in results:
except TimeoutError as te:
print("Timeout Error: {}".format(te))
print("DEBUG: {}".format(levels))
return levels
def getStatus(lvls):
# "my level" should always be index 0
myLevel = lvls.pop(0)
# Highest level, of remaining data points
highLevel = max(lvls, key=lambda x: x[1])
# Logic
status = ""
if (myLevel[1] == highLevel[1]):
status = "Sync"
elif (myLevel[1] < highLevel[1]):
# Check if timestamps within 10s
tenSeconds = timedelta(seconds=10)
if ((myLevel[2] + tenSeconds) >= highLevel[2]):
# This is ok. Local node within 10s
status = "S-10"
# Probably bad, need to alarm
# Ideally would inform baker/endorser to RPC to remote
status = "NonS"
status = "High"
return status
def main():
global nodes
# print table header
titles = ['status'] + list(map(lambda x: x[0], nodes))
print('| '.join(str(x).ljust(12) for x in titles))
while True:
levelStats = getLevels()
if (len(levelStats) < 2):
print("-- Unable to fetch enough stats")
status = getStatus(levelStats.copy())
# Copy/Append potential timeout/error message for pretty print
for i, d in enumerate(levelStats):
if (levelStats[i][1] == 0):
levelStats[i][1] = "{} {}".format(str(levelStats[i][1]), levelStats[i][3])
# combine status plus levels
data = [status] + list(map(lambda x: x[1], levelStats))
# print table data
print('| '.join(str(x).ljust(12) for x in data))
except KeyboardInterrupt:
if __name__ == "__main__":
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment