Skip to content

lf_interop_video_streaming.py: Fix incremental-capacity device tracking and cumulative batch restarts. #344

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
goyalsaurabh06 merged 1 commit into from
Aug 7, 2026
Merged

lf_interop_video_streaming.py: Fix incremental-capacity device tracking and cumulative batch restarts. #344

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
102 changes: 68 additions & 34 deletions py-scripts/lf_interop_video_streaming.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 "
Expand All @@ -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),
Expand Down Expand Up @@ -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))
Expand Down Expand Up @@ -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 = []
Expand Down Expand Up @@ -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):
Expand All @@ -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]
Expand All @@ -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:
Expand Down Expand Up @@ -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):
Expand Down Expand Up @@ -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':
Expand Down Expand Up @@ -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))
Expand Down Expand Up @@ -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)
Expand All @@ -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)
Expand Down
Loading