scottk/wlan-lanforge-scripts
1
0
Fork 0
mirror of https://github.com/Telecominfraproject/wlan-lanforge-scripts.git synced 2025-11-03 20:27:54 +00:00
Code Issues Packages Projects Releases Wiki Activity

Improvements recommended by Jed on Realm

Browse Source
This commit is contained in:
Matthew Stidham
2020-12-22 09:13:07 -08:00
parent f9958f01d1
commit 6ffee44895
1 changed files with 52 additions and 64 deletions
Download Patch File Download Diff File Expand all files Collapse all files

116
py-json/realm.py
View File

@@ -1073,59 +1073,44 @@ class L3CXProfile(LFCliBase):
else: else:
return False return False
def monitor(self,duration_sec=60, def monitor(self,
interval_sec=1, duration_sec=60,
monitor_interval=1,
col_names=None, col_names=None,
created_cx=None, created_cx=None,
show=False, show=False,
report_file=None, report_file=None,
excel=None): output_format=None):
if (duration_sec is None) or (duration_sec <= 1): try:
raise ValueError("L3CXProfile::monitor wants duration_sec > 1 second") duration_sec=int(duration_sec)
if (interval_sec is None) or (interval_sec < 1): except:
raise ValueError("L3CXProfile::monitor wants interval_sec >= 1 second") if (duration_sec is None) or (duration_sec <= 1):
if (duration_sec <= interval_sec ): raise ValueError("L3CXProfile::monitor wants duration_sec > 1 second")
raise ValueError("L3CXProfile::monitor wants duration_sec > interval_sec") if (duration_sec <= monitor_interval ):
raise ValueError("L3CXProfile::monitor wants duration_sec > monitor_interval")
if report_file == None:
raise ValueError("Monitor requires an output file to be defined")
if created_cx == None:
raise ValueError("")
if (monitor_interval is None) or (monitor_interval < 1):
raise ValueError("L3CXProfile::monitor wants monitor_interval >= 1 second")
if col_names is None: if col_names is None:
raise ValueError("L3CXProfile::monitor wants a list of column names to monitor") raise ValueError("L3CXProfile::monitor wants a list of column names to monitor")
#Step 1, get a list of Layer 3 columns
lfcli=LFCliBase('localhost',8080) #Step 1, column names
if created_cx == None: #No user defined endpoints
try:
print('Loading Layer 3 Connections')
endps = ','.join([[*x.keys()][0] for x in lfcli.json_get('endp')['endpoint']])
except:
print('No layer 3 connections found')
else: #User defined Layer 3 columns
try:
print('Loading user defined Layer 3 Connections')
endps=created_cx
except:
print('Please format your col_names variable like the following:')
#Step 2, column names
fields=",".join(col_names) fields=",".join(col_names)
print('fields') #Step 2, create report file
print(fields) report_fh = open(report_file, "w+")
#Step 3, create report file #Step 3, monitor columns
if (report_file is not None) and (report_file != ""):
report_fh = open(report_file, "w+")
else:
pass
#report_fh = open(report_file, "w")
#Step 4, monitor columns
start_time = datetime.datetime.now() start_time = datetime.datetime.now()
end_time = start_time + datetime.timedelta(seconds=duration_sec) end_time = start_time + datetime.timedelta(seconds=duration_sec)
print('endpoints')
print(endps)
value_map = dict() value_map = dict()
passes = 0 passes = 0
expected_passes = 0 expected_passes = 0
old_cx_rx_values = self.__get_rx_values() old_cx_rx_values = self.__get_rx_values()
while datetime.datetime.now() < end_time: while datetime.datetime.now() < end_time:
response = lfcli.json_get("/endp/%s?fields=%s" % (endps, fields), debug_=self.debug) response = self.json_get("/endp/%s?fields=%s" % (created_cx, fields), debug_=self.debug)
# lfcli.json_get("/endp/VTsta0000-0-B,VTsta0001-1")
if "endpoint" not in response: if "endpoint" not in response:
print(response) print(response)
raise ValueError("no endpoint?") raise ValueError("no endpoint?")
@@ -1147,36 +1132,39 @@ class L3CXProfile(LFCliBase):
self._fail("FAIL: Not all stations increased traffic") self._fail("FAIL: Not all stations increased traffic")
self.exit_fail() self.exit_fail()
old_cx_rx_values = new_cx_rx_values old_cx_rx_values = new_cx_rx_values
time.sleep(interval_sec) time.sleep(monitor_interval)
#print(value_map) #print(value_map)
if passes == expected_passes: if passes == expected_passes:
self._pass("PASS: All tests passed") self._pass("PASS: All tests passed")
#Step 5, close and save #Step 4, close and save
endpoints=list() if output_format.lower() == 'excel':
for endpoint in value_map.values(): endpoints=list()
endpoints.append(endpoint['endpoint']) for endpoint in value_map.values():
endpoints2=[] endpoints.append(endpoint['endpoint'])
for y in range(0, len(endpoints)): endpoints2=[]
for x in range(0, len(endpoints[0])): for y in range(0, len(endpoints)):
endpoints2.append(list(list(endpoints[y][x].values())[0].values())) for x in range(0, len(endpoints[0])):
timestamps=[] endpoints2.append(list(list(endpoints[y][x].values())[0].values()))
for timestamp in [*value_map.keys()]: timestamps=[]
timestamps.extend([str(timestamp)]*2*len(endps)) for timestamp in [*value_map.keys()]:
for point in range(0, len(endpoints2)): timestamps.extend([str(timestamp)]*2*len(created_cx))
endpoints2[point].insert(0, timestamps[point]) for point in range(0, len(endpoints2)):
workbook = xlsxwriter.Workbook(report_file) endpoints2[point].insert(0, timestamps[point])
worksheet = workbook.add_worksheet() workbook = xlsxwriter.Workbook(report_file)
header_row=col_names worksheet = workbook.add_worksheet()
header_row.insert(0,'Timestamp') header_row=col_names
for col_num,data in enumerate(header_row): header_row.insert(0,'Timestamp')
worksheet.write(0, col_num,data) for col_num,data in enumerate(header_row):
row_num = 1 worksheet.write(0, col_num,data)
for x in endpoints2: row_num = 1
for col_num, data in enumerate(x): for x in endpoints2:
worksheet.write(row_num, col_num, str(data)) for col_num, data in enumerate(x):
row_num+=1 worksheet.write(row_num, col_num, str(data))
workbook.close() row_num+=1
workbook.close()
else:
pass