IPFS lights-based monitoring on self-hosted SBC (blue is for status, green and red for upstream and downstream payloads)

Summary: IPFS is light and simple enough to run from one's home, even on a low-voltage machine, and the code below can be used as a baseline for monitoring IPFS activity 24/7

---

#!/usr/bin/python3
# 2019-04-22
# 2020-11-07





from blinkt import set_pixel, show
from random import randint,random,shuffle,randrange
from time import sleep
import argparse
import signal





def solid(r,g,b,s):
    while True:
        for pixel in range(8):
            set_pixel(pixel, r, g, b)
            show()
            sleep(0.1)




def random_lights3():
    while True:
        for pixel in range(8):
            r = randint(0, 255)
            g = randint(0, 255)
            b = randint(0, 255)
            set_pixel(pixel, r, g, b)
            show()
            sleep(0.1)




def random_lights2():
    while True:
        p=range(8)
        p=sorted(p, key=lambda x: random())
        for pixel in p:
            r = randrange(0, 255, 16)
            g = randrange(0, 255, 16)
            b = randrange(0, 255, 16)
            set_pixel(pixel, r, g, b)
            show()
            sleep(0.1)




def random_lights1():
    while True:
        p=range(8)
        p=sorted(p, key=lambda x: random())
        for pixel in p:
            r = randrange(0, 255, 8)
            g = randrange(0, 255, 8)
            b = randrange(0, 255, 8)
            set_pixel(pixel, r, g, b)
            show()
            sleep(0.1)




def spacer(r,g,b,seconds):
    while True:
        for pixel in range(8):
            set_pixel(pixel, r, g, b)
            next = (pixel+1)%8
            set_pixel(next, 0, 0, 0)
            show()
            sleep(seconds)




def reversed_spacer(r,g,b,seconds):
    while True:
        for pixel in reversed(range(8)):
            set_pixel(pixel, r, g, b)
            prev = (pixel-1)%8
            set_pixel(prev, 0, 0, 0)
            show()
            sleep(seconds)




def cylon(r,g,b,seconds):
    while True:
        for pixel in reversed(range(8)):
            set_pixel(pixel, r, g, b)
            prev = (pixel-1)%8
            if prev < pixel:
                set_pixel(prev, 0, 0, 0)
            show()
            sleep(seconds)
        for pixel in range(8):
            set_pixel(pixel, r, g, b)
            next = (pixel+1)%8
            if next > pixel:
                set_pixel(next, 0, 0, 0)
            show()
            sleep(seconds)





def pulsed_bar(r,g,b,seconds):
    steps=8
    while True:
        for fade in reversed(range(steps)):
            r2=r*(fade+1)/steps
            g2=g*(fade+1)/steps
            b2=b*(fade+1)/steps
            # print (fade)
            for pixel in range(8):
                set_pixel(pixel, r2, g2, b2)





            show()
            sleep(seconds)
        for fade in range(int(steps/1)):
            r2=r*(fade+1)/steps
            g2=g*(fade+1)/steps
            b2=b*(fade+1)/steps
            for pixel in range(8):
                set_pixel(pixel, r2, g2, b2)





            show()
            sleep(seconds*0.5)





def ipfs(r,g,b,seconds):
    steps=4  # how many stages in gradient
    brightness=0.5 # how bright the lights will get
    bluebright=100 # the brightness of the blue light in the middle (0-255), albeit overriden by input
    dim=1 # increase to dim down the lights
    run = 0 # running count for periodic file access
    while True: # run always (until interruption)
      run=run+1    
    # first, open from files the required values, which change over time
      if (int(run) % 50 == 1):
        with open(r'~/RateIn', 'r') as f:   # open from file the IN value
            # print(r)
            lines = f.read().splitlines()
            r=int(lines[-1]) # read the value
    # r=int(map(int, f.readline().split())) # prototype, for multiples (stale)





        with open(r'~/RateOut', 'r') as f: # open from file OUT value
            # print(g) # show values, debugging
            lines = f.read().splitlines()
            g=int(lines[-1]) 





        with open(r'~/Swarm', 'r') as f: # open from file Swarm value
            # print(g) # show values, debugging
            lines = f.read().splitlines()
            bluebright=int(lines[-1])/2
            # print(bluebright) 





      for fade in reversed(range(steps)):  # fade in effect
#            print(g2) # show values again, debugging
#            print(r2)
            r2=r*(fade+1)/steps/dim
            g2=g*(fade+1)/steps/dim
            b2=b*(fade+1)/steps/dim





            # print(g2) # show values again, debugging
            # print(r2)





            # print (fade)
            for pixel in range(3):  # first 3 LED lights
                set_pixel(pixel, r2/20, (g2*brightness)+(pixel*1), b2/20)





            for pixel in range(5,8): # the other/last 3 lights
                set_pixel(pixel, (r2*brightness)+(pixel*1), g2/20, b2/20)
            if  (bluebright==0):
                set_pixel(3, 255, 255, 255)
                set_pixel(4, 255, 255, 255)
            else:
                set_pixel(3, 0, 0, 0)
                set_pixel(4, 0, 0, bluebright)





            show()
            sleep(seconds/r*r+0.1)
      for fade in range(int(steps/1)): # fade out effect
            r2=r*(fade+1)/steps/dim
            g2=g*(fade+1)/steps/dim
            b2=b*(fade+1)/steps/dim




            for pixel in range(3):    
                set_pixel(pixel, r2/20, (g2*brightness)+(pixel*1), b2/20)





            for pixel in range(5,8):  
                set_pixel(pixel, (r2*brightness)+(pixel*1), g2/20, b2/20)
            set_pixel(3, 0, 0, bluebright)
            set_pixel(4, 0, 0, 0)
            show()
            sleep(seconds/g*g+0.1)





def flashed_bar(r,g,b,seconds):
    while True:
        for half in range(4):
            set_pixel(half,r,g,b)
        for half in range(4,8):
            set_pixel(half,0,0,0)
        show()
        sleep(seconds)
        for half in range(4,8):
            set_pixel(half,r,g,b)
        for half in range(4):
            set_pixel(half,0,0,0)
        show()
        sleep(seconds)





def handler(signum, frame):
    print("\nSignal handler called with signal", signum)
    exit(0)




signal.signal(signal.SIGTERM, handler)
signal.signal(signal.SIGINT,  handler)





# read run-time options





parser = argparse.ArgumentParser(description="Drive 'blinkt' 8-pixel display.")
parser.add_argument("pattern", help="name of light pattern: \
random[1-3], spacer, reversed_spacer, cylon, pulsed_bar, flashed_bar")
parser.add_argument("r", metavar="r", type=int, help="red channel, 0-255")
parser.add_argument("g", metavar="g", type=int, help="green channel, 0-255")
parser.add_argument("b", metavar="b", type=int, help="blue channel, 0-255")
parser.add_argument("timing", metavar="s", type=float, \
                    help="rate of binking in seconds")
options = parser.parse_args()





pattern = options.pattern.lower()
r = options.r
g = options.g
b = options.b
s = options.timing





if pattern == "solid":
    solid(r,b,g,s)
elif pattern == "random3":
    random_lights3()
elif pattern == "random2":
    random_lights2()
elif pattern == "random1" or pattern == "random":
    random_lights1()
elif pattern == "spacer":
    spacer(r,g,b,s)
elif pattern == "reversed_spacer":
    reversed_spacer(r,g,b,s)
elif pattern == "cylon":
    cylon(r,g,b,s)
elif pattern == "pulsed_bar":
    pulsed_bar(r,g,b,s)
elif pattern == "ipfs":
    ipfs(r,g,b,s)
elif pattern == "flashed_bar":
    flashed_bar(r,g,b,s)
else:
    print("Unknown pattern")
    exit(1)





exit(0)

---

Example runtime: run-blinkt-ipfs.py ipfs 0 0 0 0.00

Based on or derived from baseline blinkt scripts; requires the hardware and accompanying libraries being installed on the system.

For the code to run properly in the above form, given that it takes input from files, the IPFS values need to be periodically written to disk/card, e.g. for every minute of the day:

* * * * * ipfs stats bw | grep RateIn | cut -d ' ' -f 2 | cut -d '.' -f 1 >> ~/RateIn * * * * * ipfs stats bw | grep RateOut | cut -d ' ' -f 2 | cut -d '.' -f 1 >> ~/RateOut * * * * * ipfs swarm peers | wc -l >> ~/Swarm

These lists of numbers can, in turn, also produce status reports to be shown in IRC channels. When our git repository becomes public it'll be included (AGPLv3).