#!/usr/bin/env python3 import sys import pprint import os if sys.version_info[0] != 3: print("This script requires Python 3") exit(1) if 'py-json' not in sys.path: sys.path.append(os.path.join(os.path.abspath('..'),'py-json')) import argparse from LANforge.lfcli_base import LFCliBase from LANforge.LFUtils import * from LANforge import LFUtils import realm import time import datetime #Currently, this test can only be applied to UDP connections class TIPStationPowersave(LFCliBase): def __init__(self, host, port, ssid=None, security="open", password="[BLANK]", resource_=1, channel_=0, normal_station_list_=None, normal_station_radio_=None, powersave_station_list_=None, powersave_station_radio_=None, monitor_name_=None, monitor_radio_=None, side_a_min_rate_=56000, side_b_min_rate_=56000, side_a_max_rate_=0, side_b_max_rate_=0, pdu_size_=1000, traffic_duration_="5s", 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.resource = resource_ if (channel_ == 0): raise ValueError("Please set your radio channel") self.channel = channel_ self.monitor_name = monitor_name_ self.monitor_radio = monitor_radio_ self.host = host self.port = port self.ssid = ssid self.security = security self.password = password self.normal_sta_list = normal_station_list_ self.normal_sta_radio = normal_station_radio_ self.powersave_sta_list = powersave_station_list_ self.powersave_sta_radio = powersave_station_radio_ self.debug = debug_on_ self.local_realm = realm.Realm(lfclient_host=self.host, lfclient_port=self.port, debug_=self.debug, halt_on_error_=self.exit_on_error) # background traffic self.cx_prof_bg = self.local_realm.new_l3_cx_profile() self.cx_prof_bg.side_a_min_bps = side_a_min_rate_ self.cx_prof_bg.side_b_min_bps = side_a_min_rate_ self.cx_prof_bg.side_a_max_bps = side_a_max_rate_ self.cx_prof_bg.side_b_max_bps = side_a_min_rate_ #upload self.cx_prof_upload = self.local_realm.new_l3_cx_profile() self.cx_prof_upload.side_a_min_bps = side_a_min_rate_ self.cx_prof_upload.side_b_min_bps = 0 self.cx_prof_upload.side_a_max_bps = side_a_max_rate_ self.cx_prof_upload.side_b_max_bps = 0 self.cx_prof_upload.side_a_min_pdu = pdu_size_ self.cx_prof_upload.side_a_max_pdu = 0 self.cx_prof_upload.side_b_min_pdu = pdu_size_ self.cx_prof_upload.side_b_max_pdu = 0, #download self.cx_prof_download = self.local_realm.new_l3_cx_profile() self.cx_prof_download.side_a_min_bps = 0 self.cx_prof_download.side_b_min_bps = side_b_min_rate_ self.cx_prof_download.side_a_max_bps = 0 self.cx_prof_download.side_b_max_bps = side_b_max_rate_ self.cx_prof_download.side_a_min_pdu = pdu_size_ self.cx_prof_download.side_a_max_pdu = 0 self.cx_prof_download.side_b_min_pdu = pdu_size_ self.cx_prof_download.side_b_max_pdu = 0 self.test_duration = traffic_duration_ if isinstance(self.test_duration, int): self.test_duration = "%s"%traffic_duration_ if isinstance(self.test_duration, str): self.test_duration = self.local_realm.parse_time(self.test_duration) self.pause_duration = pause_duration_ if isinstance(self.pause_duration, int): self.pause_duration = "%s"%pause_duration_ if isinstance(self.pause_duration, str): self.pause_duration = self.local_realm.parse_time(self.pause_duration) self.sta_powersave_enabled_profile = self.local_realm.new_station_profile() self.sta_powersave_disabled_profile = self.local_realm.new_station_profile() self.wifi_monitor_profile = self.local_realm.new_wifi_monitor_profile() def build(self): self.sta_powersave_disabled_profile.use_security("open", ssid=self.ssid, passwd=self.password) self.sta_powersave_disabled_profile.set_command_flag("add_sta", "create_admin_down", 1) self.sta_powersave_disabled_profile.set_command_param("set_port", "report_timer", 5000) self.sta_powersave_disabled_profile.set_command_flag("set_port", "rpt_timer", 1) self.sta_powersave_enabled_profile.use_security("open", ssid=self.ssid, passwd=self.password) self.sta_powersave_enabled_profile.set_command_flag("add_sta", "create_admin_down", 1) self.sta_powersave_enabled_profile.set_command_param("set_port", "report_timer", 5000) self.sta_powersave_enabled_profile.set_command_flag("set_port", "rpt_timer", 1) self.sta_powersave_enabled_profile.set_command_flag("add_sta", "power_save_enable", 1) self.wifi_monitor_profile.create(resource_=self.resource, channel=self.channel, radio_=self.monitor_radio, name_=self.monitor_name) LFUtils.wait_until_ports_appear(resource_id=1, base_url=self.local_realm.lfclient_url, port_list=[self.monitor_name]) time.sleep(0.2) mon_j = self.json_get("/port/1/%s/%s"%(self.resource, self.monitor_name)) if ("interface" not in mon_j): raise ValueError("No monitor found") self.sta_powersave_disabled_profile.create(resource=1, radio=self.normal_sta_radio, sta_names_=self.normal_sta_list, debug=self.debug, suppress_related_commands_=True) self.sta_powersave_enabled_profile.create(resource=1, radio=self.powersave_sta_radio, sta_names_=self.powersave_sta_list, debug=self.debug, suppress_related_commands_=True) temp_sta_map = {} for name in self.powersave_sta_list + self.normal_sta_list: if (name in self.sta_powersave_disabled_profile.station_names) \ or (name in self.sta_powersave_enabled_profile.station_names): temp_sta_map[name]=1 print("Stations we want:") pprint.pprint(temp_sta_map) self.local_realm.wait_until_ports_appear(self.resource, temp_sta_map.keys()) if len(temp_sta_map) == (len(self.sta_powersave_disabled_profile.station_names) + len(self.sta_powersave_enabled_profile.station_names)): self._pass("Stations created", print_=True) else: print("Stations we see created:") pprint.pprint(temp_sta_map) self._fail("Not all stations created", print_=True) bg_side_a_eids = [] for port in self.normal_sta_list: bg_side_a_eids.append( "%s.%s"%(self.resource, port)) ul_side_a_eids = [] for port in self.normal_sta_list: ul_side_a_eids.append( "%s.%s"%(self.resource, port)) dl_side_a_eids = [] for port in self.normal_sta_list: dl_side_a_eids.append( "%s.%s"%(self.resource, port)) print("Creating background cx profile ") self.cx_prof_bg.name_prefix= "udp_bg" self.cx_prof_bg.create(endp_type="lf_udp", side_a=bg_side_a_eids, side_b="1.eth1") print("Creating upload cx profile ") self.cx_prof_upload.name_prefix = "udp_up" self.cx_prof_upload.create(endp_type="lf_udp", side_a=ul_side_a_eids, side_b="1.eth1") print("Creating download cx profile") self.cx_prof_download.name_prefix = "udp_down" self.cx_prof_download.create(endp_type="lf_udp", side_a=ul_side_a_eids, side_b="1.eth1") def __get_rx_values(self): cx_list = self.json_get("/endp/list?fields=name,rx+bytes", debug_=False) #print("==============\n", cx_list, "\n==============") cx_rx_map = {} for cx_name in cx_list['endpoint']: if cx_name != 'uri' and cx_name != 'handler': for item, value in cx_name.items(): for value_name, value_rx in value.items(): if value_name == 'rx bytes': cx_rx_map[item] = value_rx return cx_rx_map def start(self, print_pass=False, print_fail = False): """ 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: """ #admin up on new monitor self.wifi_monitor_profile.admin_up() now = datetime.datetime.now() date_time = now.strftime("%Y-%m-%d-%H%M%S") curr_mon_name = self.wifi_monitor_profile.monitor_name pcap_file = "/home/lanforge/Documents/%s-%s.pcap"%(curr_mon_name, date_time) capture_duration = 2 * ( self.test_duration.total_seconds() + self.pause_duration.total_seconds() + 4) self.wifi_monitor_profile.start_sniff(pcap_file, capture_duration) time.sleep(0.05) self.sta_powersave_disabled_profile.admin_up(resource=1) self.sta_powersave_enabled_profile.admin_up(resource=1) LFUtils.wait_until_ports_admin_up(resource_id=self.resource, base_url=self.local_realm.lfclient_url, port_list=self.sta_powersave_disabled_profile.station_names + self.sta_powersave_enabled_profile.station_names) self.local_realm.wait_for_ip(resource=self.resource, station_list=self.sta_powersave_disabled_profile.station_names + self.sta_powersave_enabled_profile.station_names) self.cx_prof_bg.start_cx() print("Upload starts at: %d"%time.time()) self.cx_prof_upload.start_cx() time.sleep(self.test_duration.total_seconds()) self.cx_prof_upload.stop_cx() print("Upload ends at: %d"%time.time()) time.sleep(float(self.pause_duration.total_seconds())) # here is where we should sleep long enough for station to go to sleep print("Download begins at: %d"%time.time()) self.cx_prof_download.start_cx() time.sleep(float(self.test_duration.total_seconds())) self.cx_prof_download.stop_cx() def stop(self): #switch off new monitor self.wifi_monitor_profile.admin_down() self.cx_prof_download.stop_cx() self.cx_prof_upload.stop_cx() self.sta_powersave_enabled_profile.admin_down(self.resource) self.sta_powersave_disabled_profile.admin_down(self.resource) def cleanup(self): self.wifi_monitor_profile.cleanup(resource_=self.resource, desired_ports=[self.monitor_name]) self.cx_prof_download.cleanup() self.cx_prof_upload.cleanup() self.sta_powersave_enabled_profile.cleanup(resource=self.resource, desired_stations=self.powersave_sta_list) self.sta_powersave_disabled_profile.cleanup(resource=self.resource, 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) normal_station_list = ["sta1000" ] powersave_station_list = ["sta0001"] #,"sta0002","sta0003","sta0004"] ip_powersave_test = TIPStationPowersave(lfjson_host, lfjson_port, ssid="jedway-open-x2048-5-1", password="[BLANK]", channel_=157, normal_station_list_=normal_station_list, normal_station_radio_="wiphy0", powersave_station_list_=powersave_station_list, powersave_station_radio_="wiphy0", monitor_name_="moni0", monitor_radio_="wiphy1", side_a_min_rate_=56000, side_b_min_rate_=56000, traffic_duration_="5s", pause_duration_="2s", debug_on_=True, exit_on_error_=False, exit_on_fail_=True) ip_powersave_test.cleanup() ip_powersave_test.build() ip_powersave_test.start() ip_powersave_test.stop() ip_powersave_test.cleanup() if __name__ == "__main__": main()