uptime-monitor/main.py

import cv2
import numpy as np
import urllib.request
from datetime import datetime
import time
import os
recent_blog=[]
recent_cloud=[]
recent_code=[]
recent_git=[]
blog=[]
cloud=[]
code=[]
git=[]
width=720
height=50
red=(0x00,0x00,0xff)
orange=(0x00,0x66,0xff)
yellow=(0x00,0xcc,0xff)
light_green=(0x33,0xff,0xcc)
green=(0x00,0xcc,0x00)
imgpath="/tmp/uptime-graph/"

def update_history(history,recent):
    average=0
    for i in recent:
        average+=i
    average/=len(recent)
    if len(history)>=144:
        del(history[0])
    history.append(average)
def fill(img,pos,color):
    for i in range(5):
        for j in range(height):
            img[j,i+5*pos,:]=[color[0]%256,color[1]%256,color[2]%256]
def update_graph(filename,history):
    img=np.ones([height,width,3],dtype=np.uint8)
    for i,delay in enumerate(history):
        if delay>=5000:
            fill(img,i,red)
        elif delay>=2000:
            fill(img,i,orange)
        elif delay>=1000:
            fill(img,i,yellow)
        elif delay>=500:
            fill(img,i,light_green)
        else:
            fill(img,i,green)
    cv2.imwrite(imgpath+filename, img)        


    
def testConnect(url):
    try:
        time1=datetime.now().timestamp()
        status_code=urllib.request.urlopen(url,timeout=5).getcode()
        time2=datetime.now().timestamp()
        delay=int((time2-time1)*1000)
    except:
        status_code=404
    if status_code/100>=4:
        delay=10000
    return (status_code,delay)

if __name__=="__main__":
    if not os.path.exists(imgpath):#如果路径不存在
        os.makedirs(imgpath)
    while True:
        time.sleep(0.5)
        if datetime.now().second==0:
            time.sleep(1)
            (status_code,delay)=testConnect("https://blog.clf3.org")
            print(status_code,end=" ")
            if status_code/100<=3:
                print("connected",delay,"ms")
            else:
                print("error")
            if len(recent_blog)<10:
                recent_blog.append(delay)
            else:
                update_history(blog,recent_blog)
                update_graph("blog.png",blog)
                recent_blog=[]
                recent_blog.append(delay)

        elif datetime.now().second==10:
            time.sleep(1)
            (status_code,delay)=testConnect("https://cloud.clf3.org")
            print(status_code,end=" ")
            if status_code/100<=3:
                print("connected",delay,"ms")
            else:
                print("error")
            if len(recent_cloud)<10:
                recent_cloud.append(delay)
            else:
                update_history(cloud,recent_cloud)
                update_graph("cloud.png",cloud)
                recent_cloud=[]
                recent_cloud.append(delay)
 
        elif datetime.now().second==20:
        # if True:
            time.sleep(1)
            (status_code,delay)=testConnect("https://code.clf3.org")
            print(status_code,end=" ")
            if status_code/100<=3:
                print("connected",delay,"ms")
            else:
                print("error")
            if len(recent_code)<10:
                recent_code.append(delay)
            else:
                update_history(code,recent_code)
                update_graph("code.png",code)
                recent_code=[]
                recent_code.append(delay)

        elif datetime.now().second==30:
        # if True:
            time.sleep(1)
            (status_code,delay)=testConnect("https://git.clf3.org")
            print(status_code,end=" ")
            if status_code/100<=3:
                print("connected",delay,"ms")
            else:
                print("error")
            if len(recent_git)<10:
                recent_git.append(delay)
            else:
                update_history(git,recent_git)
                update_graph("git.png",git)
                recent_git=[]
                recent_git.append(delay)