-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathispmon.py
181 lines (162 loc) · 7.07 KB
/
ispmon.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
# PrettyFlyForABeeGuy 12/20/2023.
# Regularly test your ISPs speed and log the results.
# Using the aarch64 Speedtest by Ookla utility: https://www.speedtest.net/apps/cli
# To easily view the .csv file from the command line:
# cat data/st_results.csv | perl -pe 's/((?<=,)|(?<=^)),/ ,/g;' | column -t -s, | less -S
import time, datetime
import logging
import os
import sys
import csv
import re
class ISPMonitor:
def __init__(self):
logging.basicConfig(filename='debug.log', format='%(asctime)s - %(message)s', level=logging.INFO)
self.cmd = "./speedtest"
self.datafile = "data/speedtest.txt"
self.resultsfile = "data//st_results.csv"
self.csvcolumnheaders = ["Timestamp", "Server", "ISP", "Idle Latency in ms", "Idle Latency Details", "Download in Mbps", "Download Details", "Upload in Mbps", "Upload Details", "Packet Loss", "Result URL" ]
def log_output(self,data, method):
logging.info("Writing to text file for speed test results.")
#Store API results in a file
if os.path.isfile(self.datafile):
pass
try:
file = open(self.datafile, method)
file.close()
fd = os.open(self.datafile, os.O_RDWR)
line = str.encode(data)
numBytes = os.write(fd, line)
#print(f"Creating {filepath} bytes:{numBytes}")
os.close(fd)
except:
print("Failed to write to /data")
logging.exception("Exception occurred: Failed to write to /data in log_output function.")
def readText(self):
logging.info("Reading the text file for speed test results.")
#print("Reading speed test results...")
mylines = []
with open (self.datafile, 'rt') as myfile:
for myline in myfile:
myline = myline.strip('\n')
myline = myline.strip('\t')
mylines.append(myline.strip())
mylines.remove("")
mylines.remove('')
#print(mylines)
return mylines
def clean_string(self,p, s):
if p == 1:
pattern = r',| '
return re.sub(pattern, ' ', s)
def resultCleanup(self, rlist):
logging.info("Cleaning up results.")
#print("Preparing .csv payload...")
server = rlist[1]
isp = rlist[2]
latency = rlist[3]
download = rlist[5]
dlatency = rlist[6]
upload = rlist[8]
ulatency = rlist[9]
ploss = rlist[10]
rurl = rlist[11]
server = server.split("Server:",1)[1]
isp = isp.split("ISP:",1)[1]
latency = latency.split("Idle Latency:",1)[1]
download = download.split("Download:",1)[1]
upload = upload.split("Upload:",1)[1]
ploss = ploss.split("Packet Loss:",1)[1]
rurl = rurl.split("Result URL:",1)[1]
server = self.clean_string(1, server)
server = server.strip(" ")
isp = self.clean_string(1, isp)
isp = isp.strip(" ")
# Split the text out so it's easier to plot
latency = self.clean_string(1, latency)
ldetails = latency.split("ms", 1)[1]
latency = latency.split("ms", 1)[0]
latency = latency.strip(" ")
ldetails = ldetails.strip(" ")
# Split the text out so it's easier to plot
download = self.clean_string(1, download)
ddetails = download.split("Mbps", 1)[1]
download = download.split("Mbps", 1)[0]
download = download.strip(" ")
dlatency = self.clean_string(1, dlatency)
dlatency = dlatency.strip(" ")
dlatency = ddetails + dlatency
# Split the text out so it's easier to plot
upload = self.clean_string(1, upload)
udetails = upload.split("Mbps", 1)[1]
upload = upload.split("Mbps", 1)[0]
upload = upload.strip(" ")
ulatency = self.clean_string(1, ulatency)
ulatency = ulatency.strip(" ")
ulatency = udetails + ulatency
ploss = self.clean_string(1, ploss)
ploss = ploss.strip(" ")
rurl = self.clean_string(1, rurl)
rurl = rurl.strip(" ")
payload = [server,isp, latency, ldetails, download, dlatency, upload, ulatency, ploss, rurl]
return payload
def createCSV(self, csvfile):
logging.info("Creating the .csv file if it doesn't exist.")
if os.path.isfile(csvfile):
pass
else:
try:
with open(csvfile, 'w', newline='') as file:
fwriter = csv.writer(file, delimiter=',', quotechar='|', quoting=csv.QUOTE_MINIMAL)
fwriter.writerow(self.csvcolumnheaders)
except IOError:
print(f"Error: Unable to create {csvfile}")
logging.exception("Exception occurred: Failed to create .csv file in createCSV function.")
def writeCSV(self, csvfile, payload):
logging.info("Writing data to the .csv file.")
try:
print("Writing to .csv file...")
with open(csvfile, 'a', newline='') as file:
fwriter = csv.writer(file, delimiter=',', quotechar='|', quoting=csv.QUOTE_MINIMAL)
fwriter.writerow(payload)
except IOError:
print(f"Error: Unable to write to {csvfile}")
logging.exception("Exception occurred: Failed to write to .csv file in writeCSV function.")
def speedTest(self):
logging.info("executing ./speedtest")
try:
command = f"{self.cmd}"
st = os.popen(command)
st_list = st.read()
print(st_list)
self.log_output(st_list, 'w+')
except Exception as e:
print(f"Failed to perform speedtest. {e}")
logging.exception("Exception occurred: Failed to complete speedTest. %s", e)
def main(self):
waittime = 15 # Specify how many minutes to wait between speed tests.
waittime = int(waittime) * 60
while True:
print("Getting speedtest.net results. This will take a few minutes...")
timestamp = datetime.datetime.now()
print(timestamp)
_ispm.speedTest()
rlist = _ispm.readText()
payload =_ispm.resultCleanup(rlist)
payload.insert(0, timestamp)
_ispm.createCSV(self.resultsfile)
_ispm.writeCSV(self.resultsfile, payload)
print(datetime.datetime.now())
print(f"Waiting {(waittime / 60)} minute(s) before checking again.")
time.sleep(int(waittime))
if __name__ == '__main__':
logging.info("Start program.")
_ispm = ISPMonitor()
welcome = f"""***********************************************************************************
* Welcome to the ISP Monitoring tool. This python tool is a wrapper for *
* the Speedtest by Ookla utility: https://www.speedtest.net/apps/cli *
* ISPMON allows you to check your ISPs speeds at a designated interval and *
* log the results in a .csv file. *
***********************************************************************************"""
print(welcome)
_ispm.main()