Bandwidth Report Worker for Legion Torrent Server

bandwidth_monitor.py

#!/usr/bin/env python3

# Bandwidth Report Worker for Legion Torrent Server
# copyright 2021 Stephanie Sunshine
# License: AGPLv3
# License can be found at: https://www.gnu.org/licenses/agpl-3.0.en.html

from os.path import exists
from sys import exit
from time import sleep

import argparse
import redis

NETSTAT_PATH_PREFIX = "/sys/class/net/"
VERSION = "0.1"

class Monitor:
    def __init__(self):
        parser = argparse.ArgumentParser(description=f"Bandwidth Report Worker v{VERSION}")
        parser.add_argument('INTERFACE', help="Network interface to watch")
        parser.add_argument('-d', '--delay', type=int, default=30, help="Polling delay (default: 30 seconds)")
        parser.add_argument('-r', '--redis-server', default="localhost", help="Redis server address (default: localhost)")
        parser.add_argument('-p', '--redis-port', default=6379, help="Redis server port (default: 6379)")
        parser.add_argument('-s', '--secret', default=None, help="Redis password (default: none)")
        self.args = parser.parse_args()

        if not exists(NETSTAT_PATH_PREFIX+self.args.INTERFACE):
            print(f"Interface {self.args.INTERFACE} cannot be found")
            exit()

        self.RX_FH = open(f"/sys/class/net/{self.args.INTERFACE}/statistics/rx_bytes")
        self.TX_FH = open(f"/sys/class/net/{self.args.INTERFACE}/statistics/tx_bytes")

        self.redis = redis.Redis(host=self.args.redis_server, port=self.args.redis_port, password=self.args.secret)

        print(f"Starting Worker for {self.args.INTERFACE}")

    def get(self):
        self.RX_FH.seek(0)
        self.TX_FH.seek(0)
        return(int(self.RX_FH.read()), int(self.TX_FH.read()))

    def run(self):
        while True:
            (RX1, TX1) = self.get()
            sleep(self.args.delay)
            (RX2, TX2) = self.get()
            RX = (RX2-RX1)/self.args.delay
            TX = (TX2-TX1)/self.args.delay
            self.redis.set(f"system:net:{self.args.INTERFACE}:current_dl", RX)
            self.redis.set(f"system:net:{self.args.INTERFACE}:current_ul", TX)

Monitor().run()