mirror of
https://github.com/Telecominfraproject/wlan-lanforge-scripts.git
synced 2025-10-31 18:58:01 +00:00
tip_station_powersave.py: WIP with Syama's requests in mind
This commit is contained in:
Jed Reynolds
@@ -20,39 +20,72 @@ import datetime
|
||||
|
||||
#Currently, this test can only be applied to UDP connections
|
||||
class TIPStationPowersave(LFCliBase):
|
||||
def __init__(self, host, port, ssid, security, password, station_list, side_a_min_rate=56, side_b_min_rate=56, side_a_max_rate=0,
|
||||
side_b_max_rate=0, pdu_size = 1000, prefix="00000", test_duration="5m",
|
||||
_debug_on=False, _exit_on_error=False, _exit_on_fail=False):
|
||||
def __init__(self, host, port, ssid, security, password,
|
||||
normal_station_list_=None,
|
||||
powersave_station_list_=None,
|
||||
side_a_min_rate=56000,
|
||||
side_b_min_rate=56000,
|
||||
side_a_max_rate=0,
|
||||
side_b_max_rate=0,
|
||||
pdu_size = 1000,
|
||||
prefix="00000",
|
||||
traffic_duration_="5m",
|
||||
pause_duration_="2s",
|
||||
_debug_on=False,
|
||||
_exit_on_error=False,
|
||||
_exit_on_fail=False):
|
||||
super().__init__(host, port, _debug=_debug_on, _halt_on_error=_exit_on_error, _exit_on_fail=_exit_on_fail)
|
||||
self.host = host
|
||||
self.port = port
|
||||
self.ssid = ssid
|
||||
self.security = security
|
||||
self.password = password
|
||||
self.sta_list = station_list
|
||||
self.normal_sta_list = normal_station_list_
|
||||
self.powersave_sta_list = powersave_station_list_
|
||||
self.prefix = prefix
|
||||
self.debug = _debug_on
|
||||
self.local_realm = realm.Realm(lfclient_host=self.host, lfclient_port=self.port, debug_=False, halt_on_error_=True)
|
||||
#upload
|
||||
self.cx_prof_upload = realm.L3CXProfile(self.host, self.port, self.local_realm,
|
||||
side_a_min_bps=side_a_min_rate,side_b_min_bps=0,
|
||||
side_a_max_bps=side_a_max_rate,side_b_max_bps=0,
|
||||
side_a_min_pdu=pdu_size, side_a_max_pdu=pdu_size,
|
||||
side_b_min_pdu=0, side_b_max_pdu=0, debug_=False)
|
||||
side_a_min_bps=side_a_min_rate,
|
||||
side_b_min_bps=0,
|
||||
side_a_max_bps=side_a_max_rate,
|
||||
side_b_max_bps=0,
|
||||
side_a_min_pdu=pdu_size,
|
||||
side_a_max_pdu=pdu_size,
|
||||
side_b_min_pdu=pdu_size,
|
||||
side_b_max_pdu=pdu_size,
|
||||
debug_=False)
|
||||
|
||||
#download
|
||||
self.cx_prof_download = realm.L3CXProfile(self.host, self.port, self.local_realm,
|
||||
side_a_min_bps=0, side_b_min_bps=side_b_min_rate,
|
||||
side_a_max_bps=0,side_b_max_bps=side_b_max_rate,
|
||||
side_a_min_pdu=0, side_a_max_pdu=0,
|
||||
side_b_min_pdu=pdu_size,side_b_max_pdu=pdu_size, debug_=False)
|
||||
self.test_duration = test_duration
|
||||
self.sta_powersave_enabled_profile = realm.StationProfile(self.lfclient_url, self.local_realm, ssid=self.ssid, ssid_pass=self.password,
|
||||
security=self.security, number_template_=self.prefix, mode=0, up=True,
|
||||
dhcp=True,
|
||||
debug_=False)
|
||||
self.sta_powersave_disabled_profile = realm.StationProfile(self.lfclient_url, self.local_realm, ssid=self.ssid, ssid_pass=self.password,
|
||||
security=self.security, number_template_=self.prefix, mode=0, up=True,
|
||||
self.cx_prof_download = realm.L3CXProfile(self.host, self.port, self.local_realm,
|
||||
side_a_min_bps=0,
|
||||
side_b_min_bps=side_b_min_rate,
|
||||
side_a_max_bps=0,
|
||||
side_b_max_bps=side_b_max_rate,
|
||||
side_a_min_pdu=pdu_size,
|
||||
side_a_max_pdu=pdu_size,
|
||||
side_b_min_pdu=pdu_size,
|
||||
side_b_max_pdu=pdu_size,
|
||||
debug_=False)
|
||||
self.test_duration = traffic_duration_
|
||||
self.pause_duration = pause_duration_
|
||||
self.sta_powersave_enabled_profile = realm.StationProfile(self.lfclient_url, self.local_realm,
|
||||
ssid=self.ssid,
|
||||
ssid_pass=self.password,
|
||||
security=self.security,
|
||||
number_template_=self.prefix,
|
||||
mode=0,
|
||||
up=True,
|
||||
dhcp=True,
|
||||
debug_=False)
|
||||
self.sta_powersave_disabled_profile = realm.StationProfile(self.lfclient_url, self.local_realm,
|
||||
ssid=self.ssid,
|
||||
ssid_pass=self.password,
|
||||
security=self.security,
|
||||
number_template_=self.prefix,
|
||||
mode=0,
|
||||
up=True,
|
||||
dhcp=True,
|
||||
debug_=False)
|
||||
self.new_monitor = realm.WifiMonitor(self.lfclient_url, self.local_realm,debug_= _debug_on)
|
||||
@@ -73,12 +106,11 @@ class TIPStationPowersave(LFCliBase):
|
||||
self.sta_powersave_enabled_profile.set_command_flag("add_sta", "power_save_enable", 1)
|
||||
|
||||
self.new_monitor.create(resource_=1, channel=157, radio_= "wiphy1", name_="moni0")
|
||||
self.sta_powersave_disabled_profile.create(resource=1, radio="wiphy0", sta_names_=self.sta_list, debug=False)
|
||||
self.sta_powersave_disabled_profile.create(resource=1, radio="wiphy0", sta_names_=self.normal_sta_list, debug=False)
|
||||
# station_channel = self.json_get("/port/1/%s/%s")
|
||||
# pprint.pprint(station_channel)
|
||||
|
||||
|
||||
self._pass("PASS: Station builds finished")
|
||||
|
||||
self._pass("PASS: Stations created")
|
||||
temp_sta_list = []
|
||||
for name in list(self.local_realm.station_list()):
|
||||
if "sta" in list(name)[0]:
|
||||
@@ -92,8 +124,8 @@ class TIPStationPowersave(LFCliBase):
|
||||
print("Creating download cx profile")
|
||||
self.cx_prof_download.create(endp_type="lf_udp", side_a=temp_sta_list, side_b="1.eth1", sleep_time=.05)
|
||||
|
||||
|
||||
def __set_all_cx_state(self, state, sleep_time=5):
|
||||
|
||||
print("Setting CX States to %s" % state)
|
||||
cx_list = list(self.local_realm.cx_list())
|
||||
for cx_name in cx_list:
|
||||
@@ -121,8 +153,16 @@ class TIPStationPowersave(LFCliBase):
|
||||
|
||||
|
||||
def start(self, print_pass=False, print_fail = False):
|
||||
#start one test, measure
|
||||
#start second test, measure
|
||||
"""
|
||||
This method is intended to start the monitor, the normal station (without powersave),
|
||||
and the remaining power save stations. The powersave stations will transmit for tx duration,
|
||||
pause, then the AP will pass along upstream traffic. This upstream traffic (download) should
|
||||
express a beacon before actually delivering a buffer full of traffic in order to alert the
|
||||
station it should wake up for incomming traffic.
|
||||
:param print_pass:
|
||||
:param print_fail:
|
||||
:return:
|
||||
"""
|
||||
cur_time = datetime.datetime.now()
|
||||
end_time = self.local_realm.parse_time(self.test_duration) + cur_time
|
||||
#admin up on new monitor
|
||||
@@ -132,25 +172,25 @@ class TIPStationPowersave(LFCliBase):
|
||||
curr_mon_name = self.new_monitor.monitor_name
|
||||
#("date and time: ",date_time)
|
||||
self.new_monitor.start_sniff("/home/lanforge/Documents/"+curr_mon_name+"-"+date_time+".cap")
|
||||
#admin up on station
|
||||
|
||||
time.sleep(0.05)
|
||||
self.sta_powersave_disabled_profile.admin_up(resource=1)
|
||||
#self.new_monitor.set_flag()
|
||||
self.__set_all_cx_state("RUNNING")
|
||||
self.sta_powersave_enabled_profile.admin_up(resource=1)
|
||||
|
||||
while cur_time < end_time:
|
||||
#DOUBLE CHECK
|
||||
interval_time = cur_time + datetime.timedelta(minutes=1)
|
||||
while cur_time < interval_time:
|
||||
cur_time = datetime.datetime.now()
|
||||
time.sleep(1)
|
||||
self.cx_prof_download.
|
||||
# self.__set_all_cx_state("RUNNING")
|
||||
# while cur_time < end_time:
|
||||
# #DOUBLE CHECK
|
||||
# interval_time = cur_time + datetime.timedelta(minutes=1)
|
||||
# while cur_time < interval_time:
|
||||
# cur_time = datetime.datetime.now()
|
||||
# time.sleep(1)
|
||||
|
||||
|
||||
def stop(self):
|
||||
#switch off new monitor
|
||||
self.new_monitor.admin_down()
|
||||
self.__set_all_cx_state("STOPPED")
|
||||
for sta_name in self.sta_list:
|
||||
for sta_name in self.normal_sta_list:
|
||||
data = LFUtils.portDownRequest(1, sta_name)
|
||||
url = "json-cli/set_port"
|
||||
self.json_post(url, data)
|
||||
@@ -160,19 +200,30 @@ class TIPStationPowersave(LFCliBase):
|
||||
self.new_monitor.cleanup()
|
||||
self.cx_prof_download.cleanup()
|
||||
self.cx_prof_upload.cleanup()
|
||||
self.sta_powersave_disabled_profile.cleanup(resource=1, desired_stations=self.sta_list)
|
||||
self.sta_powersave_disabled_profile.cleanup(resource=1, desired_stations=self.normal_sta_list)
|
||||
|
||||
|
||||
def main():
|
||||
|
||||
lfjson_host = "localhost"
|
||||
lfjson_port = 8080
|
||||
#station_list = LFUtils.portNameSeries(prefix_="sta", start_id_=0, end_id_=4, padding_number_=10000)
|
||||
station_list = ["sta0000","sta0001"]
|
||||
ip_powersave_test = TIPStationPowersave(lfjson_host, lfjson_port, ssid = "jedway-open" , security = "open",
|
||||
password ="[BLANK]", station_list = station_list , side_a_min_rate=2000, side_b_min_rate=2000, side_a_max_rate=0,
|
||||
side_b_max_rate=0, prefix="00000", test_duration="30s",
|
||||
_debug_on=False, _exit_on_error=True, _exit_on_fail=True)
|
||||
normal_station_list = ["sta0000" ]
|
||||
powersave_station_list = ["sta0001","sta0002","sta0003","sta0004"]
|
||||
ip_powersave_test = TIPStationPowersave(lfjson_host, lfjson_port,
|
||||
ssid="jedway-open",
|
||||
security="open",
|
||||
password="[BLANK]",
|
||||
normal_station_list_=normal_station_list,
|
||||
powersave_station_list_=powersave_station_list,
|
||||
side_a_min_rate=20000,
|
||||
side_b_min_rate=20000,
|
||||
side_a_max_rate=0,
|
||||
side_b_max_rate=0,
|
||||
prefix="00000",
|
||||
traffic_duration_="5s",
|
||||
_debug_on=False,
|
||||
_exit_on_error=True,
|
||||
_exit_on_fail=True)
|
||||
ip_powersave_test.cleanup()
|
||||
ip_powersave_test.build()
|
||||
ip_powersave_test.start()
|
||||
|
||||
Reference in New Issue
Block a user