diff --git a/py-scripts/lf_interop_video_streaming.py b/py-scripts/lf_interop_video_streaming.py index ba5fb4937..86ee723e5 100755 --- a/py-scripts/lf_interop_video_streaming.py +++ b/py-scripts/lf_interop_video_streaming.py @@ -792,8 +792,10 @@ def my_monitor_runtime(self): for cx_name in self.created_cx.keys(): value = cx_metrics.get(cx_name) was_missing = value is None + # devices not yet in all_cx_list belong to a future incremental stage, not a failure + started = cx_name in self.all_cx_list if was_missing: - if cx_name not in self.missing_cx_logged: + if started and cx_name not in self.missing_cx_logged: logger.warning( "CX '{}' is missing from the monitoring data, the device may have disconnected " "or its connection was not created. Continuing the test with the remaining " @@ -817,7 +819,7 @@ def my_monitor_runtime(self): # 10s grace period after monitor start, since CXs may still be starting up past_grace_period = (self.monitor_start_time is None or (datetime.now() - self.monitor_start_time).total_seconds() >= 10) if status != 'Run': - if past_grace_period and cx_name not in self.cx_not_running_logged: + if started and past_grace_period and cx_name not in self.cx_not_running_logged: logger.warning("CX '{}' status is '{}', not running.".format(cx_name, status)) self.cx_not_running_logged.add(cx_name) self.record_device_issue(self.port_label_from_cx_name(cx_name), @@ -878,7 +880,8 @@ def wait_for_any_cx_recovery(self, timeout=40, poll_interval=5): return True self.my_monitor_runtime() elapsed = (datetime.now() - wait_start).total_seconds() - if len(self.missing_cx_logged) < len(self.created_cx): + # compare against currently-started devices only, not future incremental stages + if len(self.missing_cx_logged) < len(set(self.all_cx_list)): logger.info("Device(s) responded again after {:.0f}s, resuming.".format(elapsed)) return True logger.warning("Still no devices responding after {:.0f}s, retrying...".format(elapsed)) @@ -1055,10 +1058,6 @@ def monitor_for_runtime_csv(self, duration, file_path, individual_df, iteration, self.monitor_start_time = starttime self.data["name"] = self.my_monitor('name') current_time = datetime.now() - endtime = "" - endtime = starttime + timedelta(minutes=duration) - endtime = endtime.isoformat()[0:19] - endtime_check = datetime.strptime(endtime, "%Y-%m-%dT%H:%M:%S") self.data['status'] = self.my_monitor('status') device_type = [] username = [] @@ -1104,13 +1103,10 @@ def monitor_for_runtime_csv(self, duration, file_path, individual_df, iteration, incremental_capacity_list = self.get_incremental_capacity_list() video_rate_dict = {i: [] for i in range(len(device_type))} - # Verify CX endpoints are responding before monitoring starts. A device with no CX - # data is left in self.created_cx (my_monitor_runtime already reports it with default - # 'Stopped' metrics so the per-device arrays built above stay in sync) but is logged - # here so it's clear from the start which devices the test will continue without. if self.created_cx: self.my_monitor_runtime() - if len(self.missing_cx_logged) == len(self.created_cx): + # compare against currently-started devices only, not future incremental stages + if self.all_cx_list and len(self.missing_cx_logged) == len(set(self.all_cx_list)): logger.warning("No devices are responding before monitoring starts, retrying for " "up to 40 seconds before failing the test.") if not self.wait_for_any_cx_recovery(timeout=40, poll_interval=5): @@ -1120,9 +1116,7 @@ def monitor_for_runtime_csv(self, duration, file_path, individual_df, iteration, "Video streaming test failed: no devices are available to run the test " "(all CX endpoints missing after 40s retry).") if self.missing_cx_logged and not self.pre_monitoring_missing_logged: - # This pre-check runs on every call to monitor_for_runtime_csv, which in - # bandsteering/robot mode is invoked repeatedly as the monitor_function tick. - # Only print this summary once per test instead of on every tick. + # Print this summary once per test, since bandsteering/robot mode calls monitor_for_runtime_csv repeatedly as the monitor tick. last_response_endpoint = self.last_monitor_response.get('endpoint') if self.last_monitor_response else None if isinstance(last_response_endpoint, list): response_keys = [key for endpoint in last_response_endpoint for key in endpoint] @@ -1134,11 +1128,15 @@ def monitor_for_runtime_csv(self, duration, file_path, individual_df, iteration, "continuing the test with the remaining {} device(s): {}\n" "URL : {}\n" "Response keys: {}".format( - len(self.created_cx) - len(self.missing_cx_logged), + len(set(self.all_cx_list)) - len(self.missing_cx_logged), sorted(self.port_label_from_cx_name(cx) for cx in self.missing_cx_logged), self.last_monitor_url, response_keys)) self.pre_monitoring_missing_logged = True + # start the duration window here, after setup/pre-checks above, not at function entry + current_time = datetime.now() + endtime_check = current_time + timedelta(minutes=duration) + # Loop until the current time is less than the end time while current_time < endtime_check or self.background_run: if self.test_stopped: @@ -1207,7 +1205,8 @@ def monitor_for_runtime_csv(self, duration, file_path, individual_df, iteration, # giving up on this monitor loop. Unlike the pre-monitoring check, this does not # fail the test: the loop just ends gracefully and execution continues with # whatever data was already collected. - if self.created_cx and len(self.missing_cx_logged) == len(self.created_cx): + # compare against currently-started devices only, not future incremental stages + if self.all_cx_list and len(self.missing_cx_logged) == len(set(self.all_cx_list)): logger.warning("All devices have stopped responding during monitoring, retrying " "for up to 40 seconds before ending the monitor loop.") if not self.wait_for_any_cx_recovery(timeout=40, poll_interval=5): @@ -1317,6 +1316,10 @@ def monitor_for_runtime_csv(self, duration, file_path, individual_df, iteration, individual_df_data = [] overall_video_rate = [] + # test_stopped can break the loop before its first tick, leaving signal data unset + if 'rssi_data' not in locals(): + rssi_data, link_speed_data, bssid_data, channel_data = self.get_signal_data(resource_order=resource_order) + # Collecting data when test is stopped for i in range(len(self.data["total_wait_time"])): if self.data['status'][i] != 'Run': @@ -3662,10 +3665,9 @@ def main(): if obj.android_list: resource_ids = ",".join([item.split(".")[1] for item in obj.android_list]) - num_list = list(map(int, resource_ids.split(','))) + num_list = sorted(map(int, resource_ids.split(','))) # Sort the list - num_list.sort() # Join the sorted list back into a string sorted_string = ','.join(map(str, num_list)) @@ -3815,17 +3817,27 @@ def main(): # Case 3: Incremental list has multiple values and length of keys is greater than 1 elif len(obj.incremental) != 1 and len(keys) > 1: - index = 0 + # each stage runs the cumulative set of devices, not just the newly added ones for num in obj.incremental: + cx_order_list.append(keys[0:num]) - cx_order_list.append(keys[index: num]) - index = num + if obj.incremental[-1] < len(keys): + cx_order_list.append(keys[:]) - if index < len(keys): - cx_order_list.append(keys[index:]) + # cumulative stages need the previous batch stopped before the next one starts + cumulative_incremental_batches = len(obj.incremental) != 1 and len(keys) > 1 # Iterate over cx_order_list to start tests incrementally for i in range(len(cx_order_list)): + # a prior stage timing out shouldn't stop the next stage's devices from starting + obj.test_stopped = False + if i > 0 and cumulative_incremental_batches: + obj.stop() + # stale missing/not-running state from the stopped batch would otherwise falsely fail the new stage + obj.missing_cx_logged.clear() + obj.cx_not_running_logged.clear() + obj.pre_monitoring_missing_logged = False + time.sleep(10) if i == 0: if args.robot_test: obj.perform_robo(args, individual_dataframe_columns, cx_order_list, i, actual_start_time, iterations_before_test_stopped_by_user) @@ -3849,16 +3861,38 @@ def main(): date_time = datetime.now().strftime('%Y-%m-%d %H:%M:%S') obj.data['remaining_time_webGUI'] = [datetime.strptime(end_time_webGUI, "%Y-%m-%d %H:%M:%S") - datetime.strptime(date_time, "%Y-%m-%d %H:%M:%S")] - if args.dowebgui: - file_path = os.path.join(obj.result_dir, "../../Running_instances/{}_{}_running.json".format(obj.host, obj.test_name)) - if os.path.exists(file_path): - with open(file_path, 'r') as file: - data = json.load(file) - if data["status"] != "Running": - break - test_stopped_by_user = obj.monitor_for_runtime_csv(args.duration, file_path, individual_df, i, actual_start_time, cx_order_list[i]) - else: - test_stopped_by_user = obj.monitor_for_runtime_csv(args.duration, file_path, individual_df, i, actual_start_time, cx_order_list[i]) + try: + if args.dowebgui: + file_path = os.path.join(obj.result_dir, "../../Running_instances/{}_{}_running.json".format(obj.host, obj.test_name)) + if os.path.exists(file_path): + with open(file_path, 'r') as file: + data = json.load(file) + if data["status"] != "Running": + break + test_stopped_by_user = obj.monitor_for_runtime_csv(args.duration, file_path, individual_df, i, actual_start_time, cx_order_list[i]) + else: + test_stopped_by_user = obj.monitor_for_runtime_csv(args.duration, file_path, individual_df, i, actual_start_time, cx_order_list[i]) + except RuntimeError: + # no devices responded for this stage; stop and report with the data collected so far + obj.test_stopped = True + # this stage never ran a monitoring tick, so skip it in the report instead of feeding empty data + if not individual_df[individual_df['iteration'] == i + 1].empty: + iterations_before_test_stopped_by_user.append(i) + elif i == len(cx_order_list) - 1: + # last stage got no data at all; webGUI still needs a terminal Stopped row to detect completion + terminal_row = [] + for _ in keys: + terminal_row.extend([0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, '', '']) + terminal_row.extend([0, datetime.now().strftime("%H:%M:%S"), i + 1, + actual_start_time.strftime('%Y-%m-%d %H:%M:%S'), + obj.data['end_time_webGUI'][0], 0, "Stopped"]) + individual_df.loc[len(individual_df)] = terminal_row + final_row_df = individual_df.tail(1) + if args.dowebgui: + individual_df.to_csv('{}/video_streaming_realtime_data.csv'.format(obj.result_dir), index=False) + else: + final_row_df.to_csv('video_streaming_realtime_data.csv', index=False, mode='a', header=False) + break if not test_stopped_by_user: # Append current iteration index to iterations_before_test_stopped_by_user iterations_before_test_stopped_by_user.append(i)