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

Update port_probe and files which depend on it. Add vht and ht calculators

Browse Source 
Signed-off-by: Matthew Stidham <stidmatt@gmail.com>
This commit is contained in:
Matthew Stidham
2021-10-19 10:36:24 -07:00
parent 5bc581eacd
commit c996c90845
3 changed files with 140 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
py-json/l3_cxprofile.py
View File

@@ -16,6 +16,7 @@ pandas_extensions = importlib.import_module("py-json.LANforge.pandas_extensions"
port_probe = importlib.import_module("py-json.port_probe")
ProbePort = port_probe.ProbePort
class L3CXProfile(LFCliBase):
def __init__(self,
lfclient_host,
@@ -269,6 +270,20 @@ class L3CXProfile(LFCliBase):
probe_results['Signal Combined'] = probe_port.getSignalCombined()
probe_results['Signal per Chain'] = probe_port.getSignalPerChain()
probe_results['Beacon Avg Signal'] = probe_port.getBeaconSignalAvg()
probe_results['HE status'] = probe_port.he
probe_results['TX Bitrate'] = probe_port.tx_bitrate
probe_results['TX MCS'] = probe_port.tx_mcs
probe_results['TX NSS'] = probe_port.tx_nss
probe_results['TX MHz'] = probe_port.tx_mhz
probe_results['TX guard interval'] = probe_port.tx_ns
probe_results['RX Bitrate'] = probe_port.rx_bitrate
probe_results['RX MCS'] = probe_port.rx_mcs
probe_results['RX NSS'] = probe_port.rx_nss
probe_results['RX MHz'] = probe_port.rx_mhz
probe_results['RX guard interval'] = probe_port.rx_ns
probe_df_initial = pd.DataFrame(probe_results.values()).transpose()
probe_df_initial.columns = probe_results.keys()
probe_df_initial.columns = ['probe '+x for x in probe_df_initial.columns]

121
py-json/port_probe.py
View File

@@ -1,8 +1,16 @@
#!/usr/bin/env python3
import importlib
from time import sleep
import pandas as pd
import sys
import os
sys.path.append(os.path.join(os.path.abspath(__file__ + "../../../")))
lfcli_base = importlib.import_module("py-json.LANforge.lfcli_base")
LFCliBase = lfcli_base.LFCliBase
# Probe data can change frequently. It is recommended to update
class ProbePort(LFCliBase):
@@ -19,6 +27,22 @@ class ProbePort(LFCliBase):
self.probepath = "/probe/1/%s/%s" % (hunks[-2], hunks[-1])
self.response = None
self.signals = None
self.he = None
self.tx_bitrate = None
self.tx_mcs = None
self.tx_nss = None
self.tx_mbit = None
self.tx_mhz = None
self.tx_ns = None
self.rx_bitrate = None
self.rx_actual_mcs = None
self.rx_mcs = None
self.rx_nss = None
self.rx_mbit = None
self.rx_mhz = None
self.rx_ns = None
def refreshProbe(self):
self.json_post(self.probepath, {})
@@ -33,6 +57,61 @@ class ProbePort(LFCliBase):
keys = [x[0].strip(' ').strip(':') for x in signals]
values = [x[1].strip('dBm').strip(' ') for x in signals]
self.signals = dict(zip(keys, values))
tx_bitrate = [x for x in text if 'tx bitrate' in x][0].replace('\t', ' ')
self.tx_bitrate = tx_bitrate.split(':')[-1].strip(' ')
rx_bitrate = [x for x in text if 'rx bitrate' in x][0].replace('\t', ' ')
self.rx_bitrate = rx_bitrate.split(':')[-1].strip(' ')
try:
tx_mcs = [x.strip('\t') for x in text if 'tx bitrate' in x][0].split(':')[1].strip('\t')
self.tx_mcs = int(tx_mcs.split('MCS')[1].strip(' ').split(' ')[0])
self.tx_mbit = self.tx_bitrate.split(' ')[0]
if 'VHT' in self.tx_bitrate:
tx_mhz_ns = self.vht_calculator(self.tx_mbit, self.tx_mcs)
try:
self.tx_mhz = tx_mhz_ns['MHz']
self.tx_nss = [x.strip('\t') for x in text if 'tx bitrate' in x][0].split('VHT-NSS')[1].strip(' ')
self.tx_ns = tx_mhz_ns['ns']
except TypeError as error:
print('%s, %s' % (error, tx_mhz_ns))
else:
tx_mhz_ns = self.ht_calculator(self.tx_mbit, self.tx_mcs)
try:
self.tx_mhz = tx_mhz_ns['MHz']
self.tx_nss = int(self.tx_mcs / 8) + 1
self.tx_ns = tx_mhz_ns['ns']
except TypeError:
print('tx_mhz_ns is None')
except IndexError as error:
print(error)
try:
rx_mcs = [x.strip('\t') for x in text if 'rx bitrate' in x][0].split(':')[1].strip('\t')
self.rx_mcs = int(rx_mcs.split('MCS')[1].strip(' ').split(' ')[0])
self.rx_actual_mcs = self.rx_mcs & 8
self.rx_mbit = self.rx_bitrate.split(' ')[0]
if 'HE not supported' in [x.strip('\t') for x in text if 'HE' in x]:
self.he = False
else:
self.he = True
if 'VHT' in self.rx_bitrate:
rx_mhz_ns = self.vht_calculator(self.rx_mbit, self.rx_mcs)
try:
self.rx_mhz = rx_mhz_ns['MHz']
self.rx_nss = [x.strip('\t') for x in text if 'rx bitrate' in x][0].split('VHT-NSS')[1].strip(' ')
self.rx_ns = rx_mhz_ns['ns']
except TypeError as error:
print('%s, %s' % (error, rx_mhz_ns))
else:
rx_mhz_ns = self.ht_calculator(self.rx_mbit, self.rx_mcs)
try:
self.rx_mhz = rx_mhz_ns['MHz']
self.rx_nss = int(self.rx_mcs / 8) + 1
self.rx_ns = rx_mhz_ns['ns']
except TypeError:
print('rx_mhz_nz not found')
except IndexError as error:
print(error)
def getSignalAvgCombined(self):
return self.signals['signal avg'].split(' ')[0]
@@ -47,4 +126,44 @@ class ProbePort(LFCliBase):
return ' '.join(self.signals['signal'].split(' ')[1:])
def getBeaconSignalAvg(self):
return ' '.join(self.signals['beacon signal avg'])
return ' '.join(self.signals['beacon signal avg']).replace(' ', '')
def ht_calculator(self, mbit, mcs):
df = pd.read_csv('../py-json/mcs_snr_rssi_calculator.csv')
df = df.dropna(subset=['HT'])
df.index = [int(x) for x in df['HT']]
df = df[df['HT'] == mcs].transpose()
df['values'] = df.index
df.index = df[mcs]
df = df.drop(float(mcs), 1)
values = df.to_dict()['values']
response = dict()
try:
for value in values[float(mbit)].split(' '):
if 'MHz' in value:
response['MHz'] = value.strip('MHz')
if 'ns' in value:
response['ns'] = value.strip('ns')
return response
except KeyError:
self.vht_calculator(mbit, mcs)
@staticmethod
def vht_calculator(mbit, mcs):
df = pd.read_csv('../py-json/mcs_snr_rssi_calculator.csv')
df = df[df['VHT'] == mcs]
results = pd.concat([pd.DataFrame(dictionary.items()) for dictionary in df.transpose().to_dict().values()]).dropna()
results[1] = [round(x) for x in results[1]]
results.index = results[0]
data = dict(zip(results[1], results[0]))
response = dict()
try:
for value in data[round(float(mbit))].split(' '):
if 'MHz' in value:
response['MHz'] = value.strip('MHz')
if 'ns' in value:
response['ns'] = value.strip('ns')
return response
except KeyError as error:
print(error)
print('mbit: %s, mcs: %s' % (mbit, mcs))

19
py-scripts/test_ip_variable_time.py
View File

@@ -259,17 +259,11 @@ class IPVariableTime(Realm):
try:
monitor_interval = Realm.parse_time(self.monitor_interval).total_seconds()
except ValueError as error:
print(str(error))
print(ValueError(
"The time string provided for monitor_interval argument is invalid. Please see supported time stamp increments and inputs for monitor_interval in --help. "))
exit(1)
print(error)
return ValueError(
"The time string provided for monitor_interval argument is invalid. Please see supported time stamp increments and inputs for monitor_interval in --help. ")
self.start(False, False)
# if self.influx_mgr is None:
# manager = self.mgr
# else:
# manager = self.influx_mgr
if self.influx_org is not None:
grapher = RecordInflux(_influx_host=self.influx_host,
_influx_port=self.influx_port,
@@ -539,7 +533,7 @@ python3 ./test_ip_variable_time.py
parser.add_argument('--layer3_cols', help='Columns wished to be monitored from layer 3 endpoint tab',
default=['name', 'tx bytes', 'rx bytes', 'tx rate', 'rx rate'])
parser.add_argument('--port_mgr_cols', help='Columns wished to be monitored from port manager tab',
default=['alias', 'ap', 'ip', 'parent dev', 'rx-rate', 'beacon'])
default=['alias', 'ap', 'ip', 'parent dev', 'rx-rate'])
parser.add_argument('--compared_report', help='report path and file which is wished to be compared with new report',
default=None)
parser.add_argument('--monitor_interval',
@@ -559,9 +553,6 @@ python3 ./test_ip_variable_time.py
parser.add_argument('--create_sta', help='Used to force a connection to a particular AP', default=True)
parser.add_argument('--sta_names', help='Used to force a connection to a particular AP', default="sta0000")
args = parser.parse_args()
create_sta = True
if args.create_sta == "False":
create_sta = False
num_sta = 2
if (args.num_stations is not None) and (int(args.num_stations) > 0):
@@ -599,7 +590,7 @@ python3 ./test_ip_variable_time.py
port=args.mgr_port,
number_template="0000",
sta_list=station_list,
create_sta=create_sta,
create_sta=args.create_sta,
name_prefix="VT",
upstream=args.upstream_port,
ssid=args.ssid,