lf_interop_ping.py

#!/usr/bin/env python3

'''
    NAME: lf_interop_ping.py

    PURPOSE: lf_interop_ping.py will let the user select real devices, virtual devices or both and then allows them to run
    ping test for user given duration and packet interval on the given target IP or domain name.

    EXAMPLE-1:
    Command Line Interface to run ping test with only virtual clients
    python3 lf_interop_ping.py --mgr 192.168.200.103  --target 192.168.1.3 --virtual --num_sta 1 --radio 1.1.wiphy2 --ssid RDT_wpa2 --security wpa2
    --passwd OpenWifi --ping_interval 1 --ping_duration 1 --server_ip 192.168.1.61 --debug

    EXAMPLE-2:
    Command Line Interface to run ping test with only real clients
    python3 lf_interop_ping.py --mgr 192.168.200.103 --real --target 192.168.1.3 --ping_interval 1 --ping_duration 1 --server_ip 192.168.1.61 --ssid RDT_wpa2 --security wpa2_personal
    --passwd OpenWifi

    EXAMPLE-3:
    Command Line Interface to run ping test with both real and virtual clients
    python3 lf_interop_ping.py --mgr 192.168.200.103 --target 192.168.1.3 --real --virtual --num_sta 1 --radio 1.1.wiphy2 --ssid RDT_wpa2 --security wpa2
    --passwd OpenWifi --ping_interval 1 --ping_duration 1 --server_ip 192.168.1.61

    EXAMPLE-4:
    Command Line Interface to run ping test with existing Wi-Fi configuration on the real devices
    python3 lf_interop_ping.py --mgr 192.168.200.63 --real --target 192.168.1.61 --ping_interval 5 --ping_duration 1 --passwd OpenWifi --use_default_config

    EXAMPLE-5:
    Command Line Interface to run ping test by setting device specific Pass/Fail values in the csv file
    python3 lf_interop_ping.py --mgr 192.168.244.97 --real --target 192.168.1.3 --ping_interval 1 --ping_duration 1 --device_csv_name device.csv
     --use_default_config

    EXAMPLE-6:
    Command Line Interface to run ping test by setting the same expected Pass/Fail value for all devices
    python3 lf_interop_ping.py --mgr 192.168.244.97 --real --target 192.168.1.3 --ping_interval 1 --ping_duration 1 --expected_passfail_value 3
     --use_default_config

    EXAMPLE-7:
    Command Line Interface to run ping test by configuring Real Devices with SSID, Password, and Security
    python3 lf_interop_ping.py --mgr 192.168.244.97 --real --target 192.168.1.3 --ping_interval 1 --ping_duration 1  --ssid RDT_wpa2 --security wpa2
    --passwd OpenWifi --server_ip 192.168.244.97 --wait_time 30

    EXAMPLE-8:
    Command Line Interface to run ping test by Configuring Devices in Groups with Specific Profiles
    python3 lf_interop_ping.py --mgr 192.168.244.97 --real --target 192.168.1.3 --ping_interval 1 --ping_duration 1   --group_name grp3 --file_name g219 --profile_name Open5
     --server_ip 192.168.204.60

    EXAMPLE-9:
    Command Line Interface to run ping test by Configuring Devices in Groups with Specific Profiles with expected Pass/Fail values
    python3 lf_interop_ping.py --mgr 192.168.244.97 --real --target 192.168.1.3 --ping_interval 1 --ping_duration 1   --group_name grp3 --file_name g219 --profile_name Open5
     --expected_passfail_value 3 --server_ip 192.168.204.60

    EXAMPLE-10:
    Command Line Interface for Configuring Devices in Groups with Specific Profiles with device_csv_name
    python3 lf_interop_ping.py --mgr 192.168.244.97 --real --target 192.168.1.3 --ping_interval 1 --ping_duration 1   --group_name grp3 --file_name g219 --profile_name Open5
     --device_csv_name device.csv --server_ip 192.168.204.60

    SCRIPT_CLASSIFICATION : Test

    SCRIPT_CATEGORIES: Performance, Functional, Report Generation

    NOTES:
    1.Use './lf_interop_ping.py --help' to see command line usage and options
    2.Please pass ping_duration in minutes
    3.Please pass ping_interval in seconds
    4.After passing the cli, if --real flag is selected, then a list of available real devices will be displayed on the terminal.
    5.Enter the real device resource numbers seperated by commas (,)

    STATUS: BETA RELEASE

    VERIFIED_ON:
    Working date    - 20/09/2023
    Build version   - 5.4.7
    kernel version  - 6.2.16+

    License: Free to distribute and modify. LANforge systems must be licensed.
    Copyright (C) 2020-2026 Candela Technologies Inc.
'''

import argparse
import time
import sys
import os
import datetime
import json
import pandas as pd
import importlib
import logging
import asyncio
import csv
import re
import threading

if 'py-json' not in sys.path:
    sys.path.append(os.path.join(os.path.abspath('..'), 'py-json'))

if 'py-scripts' not in sys.path:
    sys.path.append('/home/lanforge/lanforge-scripts/py-scripts')

from lf_base_interop_profile import RealDevice
from lf_graph import lf_bar_graph_horizontal
from lf_report import lf_report
from station_profile import StationProfile
from typing import List, Optional
from LANforge import LFUtils
# Importing DeviceConfig to apply device configurations for ADB devices and laptops
DeviceConfig = importlib.import_module("py-scripts.DeviceConfig")

logger = logging.getLogger(__name__)
lf_logger_config = importlib.import_module("py-scripts.lf_logger_config")

if sys.version_info[0] != 3:
    print("This script requires Python 3")
    exit(1)

realm = importlib.import_module("py-json.realm")
Realm = realm.Realm


class Ping(Realm):
    BG_NAME_PREFIX = 'bgping'

    def __init__(self,
                 host=None,
                 port=None,
                 ssid=None,
                 security=None,
                 password=None,
                 radio=None,
                 target=None,
                 interval=None,
                 lanforge_password='lanforge',
                 sta_list=None,
                 virtual=None,
                 duration=1,
                 real=None,
                 debug=False, file_name=None,
                 profile_name=None,
                 group_name=None,
                 eap_method=None,
                 eap_identity=None,
                 ieee80211=None,
                 ieee80211u=None,
                 ieee80211w=None,
                 enable_pkc=None,
                 bss_transition=None,
                 power_save=None,
                 disable_ofdma=None,
                 roam_ft_ds=None,
                 key_management=None,
                 pairwise=None,
                 private_key=None,
                 ca_cert=None,
                 client_cert=None,
                 pk_passwd=None,
                 pac_file=None,
                 server_ip=None,
                 expected_passfail_val=None,
                 csv_name=None,
                 wait_time=60,
                 total_floors: int = None,
                 get_live_view: bool = None,
                 result_dir: str = None,
                 dowebgui: bool = False,
                 test_name: str = None):
        super().__init__(lfclient_host=host,
                         lfclient_port=port)
        self.ssid_list = []
        self.host = host
        self.lanforge_password = lanforge_password
        self.port = port
        self.lfclient_host = host
        self.lfclient_port = port
        self.ssid = ssid
        self.security = security
        self.password = password
        self.radio = radio
        self.target = target
        self.interval = interval
        self.debug = debug
        self.sta_list = sta_list
        self.real_sta_list = []
        self.real_sta_data_dict = {}
        self.enable_virtual = virtual
        self.enable_real = real
        self.duration = duration
        self.android = 0
        self.virtual = 0
        self.linux = 0
        self.windows = 0
        self.mac = 0
        self.result_json = {}
        self.generic_endps_profile = self.new_generic_endp_profile()
        self.generic_endps_profile.type = 'lfping'
        self.generic_endps_profile.dest = self.target
        self.generic_endps_profile.interval = self.interval
        self.Devices = None
        self.total_floors = total_floors
        self.get_live_view = get_live_view
        self.result_dir = result_dir
        self.dowebgui = dowebgui
        self.test_name = test_name
        self.missing_endp_logged = set()
        self.not_running_endp_logged = set()
        self.client_issue_csv_name = "client_issue.csv"
        self.client_issue_data = []
        self.latest_results = {}
        self.eap_method = eap_method
        self.eap_identity = eap_identity
        self.ieee80211 = ieee80211
        self.ieee80211u = ieee80211u
        self.ieee80211w = ieee80211w
        self.enable_pkc = enable_pkc
        self.bss_transition = bss_transition
        self.power_save = power_save
        self.disable_ofdma = disable_ofdma
        self.roam_ft_ds = roam_ft_ds
        self.key_management = key_management
        self.pairwise = pairwise
        self.private_key = private_key
        self.ca_cert = ca_cert
        self.client_cert = client_cert
        self.pk_passwd = pk_passwd
        self.pac_file = pac_file
        self.profile_name = profile_name
        self.file_name = file_name
        self.group_name = group_name
        self.server_ip = server_ip
        self.real = real
        self.expected_passfail_val = expected_passfail_val
        self.csv_name = csv_name
        self.pass_fail_list = []
        self.test_input_list = []
        self.percent_pac_loss = []
        self.wait_time = wait_time
        self.bg_stats = {}
        self.bg_packet_results = {}
        self.bg_start_time = None
        self.bg_stop_time = None
        self.bg_poll_interval = 20
        self.bg_poll_stop = None
        self.bg_poll_thread = None

    def change_target_to_ip(self):

        # checking if target is an IP or a port
        target_parts = self.target.split('.')
        if (self.target.count('.') != 3 and (len(target_parts) == 1 or target_parts[-2].isnumeric())):
            # checking if target is eth1 or 1.1.eth1
            target_port_list = self.name_to_eid(self.target)
            shelf, resource, port, _ = target_port_list
            port_url = '/port/{}/{}/{}?fields=ip'.format(shelf, resource, port)
            response = self.json_get(port_url)
            if not response:
                logger.error("Failed to fetch port info. Received empty response.\nRequested URL: '{}'\nResponse: {}".format(port_url, response))
                raise RuntimeError('The target port {} not found on the LANforge. Please change the target.'.format(self.target))
            if 'interface' not in response:
                logger.error("'interface' key not found in response.\nRequested URL: '{}'\nResponse: {}".format(port_url, response))
                raise RuntimeError('The target port {} not found on the LANforge. Please change the target.'.format(self.target))
            if 'ip' not in response['interface']:
                logger.error("'ip' key not found in response.\nRequested URL: '{}'\nResponse: {}".format(port_url, response))
                raise RuntimeError('The target port {} not found on the LANforge. Please change the target.'.format(self.target))
            self.target = response['interface']['ip']
            logger.info(self.target)
        else:
            logger.info(self.target)

    def cleanup(self):
        expected_endp_names = []
        # setting the created_cx and created_endp to empty list and adding the existing endpoints to the list for cleanup
        self.generic_endps_profile.created_cx = []
        self.generic_endps_profile.created_endp = []
        if (self.enable_virtual):
            for station in self.sta_list:
                expected_endp_names.append('{}-{}'.format(self.generic_endps_profile.name_prefix, station.split('.')[2]))
        if (self.enable_real):
            for station in self.real_sta_list:
                expected_endp_names.append('{}-{}'.format(self.generic_endps_profile.name_prefix, station))

        if expected_endp_names:
            endp_url = "/generic/{}".format(','.join(expected_endp_names))
            response = self.json_get(endp_url)
            if not response:
                logger.error("Failed to fetch endpoints for cleanup.\nRequested URL: '{}'\nResponse: {}".format(endp_url, response))
                endpoint = []
            else:
                endpoint = response.get('endpoints', response.get('endpoint', []))
                if isinstance(endpoint, dict):
                    endpoint = [{endpoint['name']: endpoint}]

            present_endp_names = set()
            for endp_entry in endpoint:
                present_endp_names.update(endp_entry.keys())

            for endp_name in expected_endp_names:
                if endp_name in present_endp_names:
                    self.generic_endps_profile.created_endp.append(endp_name)
                    self.generic_endps_profile.created_cx.append('CX_{}'.format(endp_name))

        if (self.enable_virtual):
            # removing virtual stations if existing
            for station in self.sta_list:
                logger.info('Removing the station {} if exists'.format(station))
                self.rm_port(station, check_exists=True)

            if (not LFUtils.wait_until_ports_disappear(base_url=self.host, port_list=self.sta_list, debug=self.debug)):
                logger.info('All stations are not removed or a timeout occured.')
                logger.error('Aborting the test.')
                exit(0)

        logger.info("Cleaning up generic endpoints if exists: cx={}, endp={}".format(self.generic_endps_profile.created_cx, self.generic_endps_profile.created_endp))
        self.generic_endps_profile.cleanup()
        self.generic_endps_profile.created_cx = []
        self.generic_endps_profile.created_endp = []
        logger.info('Cleanup Successful')

    # Args:
    #   devices: Connected RealDevice object which has already populated tracked real device
    #            resources through call to get_devices()
    def select_real_devices(self, real_devices, real_sta_list=None, base_interop_obj=None, device_list=None):
        if real_sta_list is None:
            self.real_sta_list, _, _ = real_devices.query_user(device_list=device_list)
        else:
            self.real_sta_list = real_sta_list
        if base_interop_obj is not None:
            self.Devices = base_interop_obj

        # Need real stations to run interop test
        if (len(self.real_sta_list) == 0):
            raise RuntimeError('There are no real devices in this testbed. Aborting test')

        logger.info(self.real_sta_list)

        for sta_name in self.real_sta_list:
            if sta_name not in real_devices.devices_data:
                logger.error('Real station {} not in devices data, ignoring it from testing'.format(sta_name))
                continue
                # raise ValueError('Real station not in devices data')

            self.real_sta_data_dict[sta_name] = real_devices.devices_data[sta_name]

        # Track number of selected devices
        self.android = self.Devices.android
        self.windows = self.Devices.windows
        self.mac = self.Devices.mac
        self.linux = self.Devices.linux
        d_list = []
        for i in self.real_sta_list:
            device = i.split('.')
            d_list.append(device[0] + '.' + device[1])
        return d_list

    def buildstation(self):
        logger.info('Creating Stations {}'.format(self.sta_list))
        for station_index in range(len(self.sta_list)):
            shelf, resource, port = self.sta_list[station_index].split('.')
            logger.info('{} {} {}'.format(shelf, resource, port))
            station_object = StationProfile(lfclient_url='http://{}:{}'.format(self.host, self.port), local_realm=self, ssid=self.ssid,
                                            ssid_pass=self.password, security=self.security, number_template_='00', up=True, resource=resource, shelf=shelf)
            station_object.use_security(
                security_type=self.security, ssid=self.ssid, passwd=self.password)

            station_object.create(radio=self.radio, sta_names_=[
                self.sta_list[station_index]])
        station_object.admin_up()
        if self.wait_for_ip([self.sta_list[station_index]]):
            self._pass("All stations got IPs", print_=True)
        else:
            self._fail(
                "Stations failed to get IPs", print_=True)

    def check_tab_exists(self):
        generic_url = "generic"
        response = self.json_get(generic_url)
        if not response:
            logger.error("Failed to fetch generic tab data.\nRequested URL: '{}'\nResponse: {}".format(generic_url, response))
            return False
        return True

    def create_generic_endp(self):
        # Virtual stations are tracked in same list as real stations, so need to separate them
        # in order to create generic endpoints for just the virtual stations
        virtual_stations = list(set(self.sta_list).difference(set(self.real_sta_list)))

        if (self.enable_virtual):
            if (self.generic_endps_profile.create(ports=virtual_stations, sleep_time=.5)):
                logger.info('Virtual client generic endpoint creation completed.')
            else:
                raise RuntimeError('Virtual client generic endpoint creation failed.')

        if (self.enable_real):
            real_sta_os_types = [self.real_sta_data_dict[real_sta_name]['ostype'] for real_sta_name in self.real_sta_data_dict]

            if (self.generic_endps_profile.create(ports=self.real_sta_list, sleep_time=.5, real_client_os_types=real_sta_os_types)):
                logger.info('Real client generic endpoint creation completed.')
            else:
                raise RuntimeError('Real client generic endpoint creation failed.')

    def start_generic(self):
        self.generic_endps_profile.start_cx()

    def stop_generic(self):
        if self.missing_endp_logged:
            missing_cx_names = {'CX_{}'.format(endp) for endp in self.missing_endp_logged}
            self.generic_endps_profile.created_endp = [
                endp for endp in self.generic_endps_profile.created_endp
                if endp not in self.missing_endp_logged
            ]
            self.generic_endps_profile.created_cx = [
                cx for cx in self.generic_endps_profile.created_cx
                if cx not in missing_cx_names
            ]
        self.generic_endps_profile.stop_cx()

    def get_results(self):
        logger.debug(self.generic_endps_profile.created_endp)
        endp_url = "/generic/{}".format(','.join(self.generic_endps_profile.created_endp))
        response = self.json_get(endp_url)
        if not response:
            logger.error("Failed to fetch endpoint results.\nRequested URL: '{}'\nResponse: {}".format(endp_url, response))
        else:
            endpoint = response.get('endpoints', response.get('endpoint', []))
            if isinstance(endpoint, dict):
                endpoint = [{endpoint['name']: endpoint}]
            present_endps = {}
            for endp_entry in endpoint:
                for endp_name, endp_response in endp_entry.items():
                    # Use the second-last result when more than one is available.
                    last_results = endp_response.get('last results')
                    if last_results:
                        self.record_packet_results(endp_name, last_results)
                        result_lines = last_results.split('\n')
                        if len(result_lines) > 1:
                            endp_response['last results'] = result_lines[-2]
                    present_endps[endp_name] = endp_response
            self.latest_results.update(present_endps)

        if not self.latest_results:
            logger.error('No ping results were collected for any endpoint during the entire test.')
            return []

        return [{name: data} for name, data in self.latest_results.items()]

    def append_endp_data(self, name, state, response):
        """Save an endpoint state change for the report."""
        self.client_issue_data.append([
            datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
            name,
            state,
            response
        ])

    def write_client_issue_csv(self, report_path):
        """Create the endpoint issue CSV in the report folder."""
        if not self.client_issue_data:
            return

        csv_path = os.path.join(report_path, self.client_issue_csv_name)
        with open(csv_path, "w", newline="") as file:
            writer = csv.writer(file)
            writer.writerow(["TIMESTAMP", "ENDP_NAME", "STATE", "API RESPONSE"])
            writer.writerows(self.client_issue_data)

    def check_endpoint_availability(self, expected_endps, present_endps):
        """Log endpoint missing/not-running/recovered transitions and update the tracking sets."""
        endp_url = "/generic/{}".format(','.join(expected_endps))
        for endp_name in expected_endps:
            if endp_name not in present_endps.keys():
                if endp_name not in self.missing_endp_logged:
                    present_endp_names = list(present_endps)
                    logger.warning(
                        "Endpoint '{}' is missing from the monitoring data.\n"
                        "Requested URL: '{}'\n"
                        "Response: {}".format(endp_name, endp_url, present_endp_names))
                    self.append_endp_data(name=endp_name, state="MISSING", response=present_endp_names)
                    self.missing_endp_logged.add(endp_name)
                self.not_running_endp_logged.discard(endp_name)
            elif present_endps[endp_name].get('status').lower() != "run":
                if endp_name not in self.not_running_endp_logged:
                    logger.warning(
                        "Endpoint '{}' is not running in the monitoring data.\n"
                        "Requested URL: '{}'\n"
                        "Response: {}".format(endp_name, endp_url, present_endps[endp_name]))
                    self.append_endp_data(name=endp_name, state=present_endps[endp_name].get('status'), response=present_endps[endp_name])
                    self.not_running_endp_logged.add(endp_name)
                self.missing_endp_logged.discard(endp_name)
            else:
                if endp_name in self.missing_endp_logged or endp_name in self.not_running_endp_logged:
                    logger.info(
                        "Endpoint '{}' is running in the monitoring data\n"
                        "Requested URL: '{}'\n"
                        "Response: {}".format(endp_name, endp_url, present_endps[endp_name]))
                    self.append_endp_data(name=endp_name, state=present_endps[endp_name].get('status'), response=present_endps[endp_name])
                self.missing_endp_logged.discard(endp_name)
                self.not_running_endp_logged.discard(endp_name)

    def is_test_stopped_by_webgui(self):
        """Check the webgui running-instance file to see if the user stopped the test."""
        if not self.dowebgui:
            return False
        running_file = f"{self.result_dir}/../../Running_instances/{self.host}_{self.test_name}_running.json"
        try:
            with open(running_file, "r") as file:
                data = json.load(file)
            if data.get("status") != "Running":
                logger.warning("Test is stopped by the user")
                return True
        except FileNotFoundError:
            logger.warning(f"Running instance file not found: {running_file}")
            return True
        except json.JSONDecodeError:
            logger.warning(f"Running instance file corrupted or empty: {running_file}")
            return True
        except Exception as e:
            logger.error(f"Unexpected error reading running.json: {e}")
            return True
        return False

    def monitor_endp_availability(self, expected_endps, duration=40, interval=5):
        """Retry for up to duration seconds until at least one expected endpoint is present and running."""
        start_time = time.time()
        end_time = start_time + duration
        no_of_attempts = duration // interval
        count = 0
        while (start_time <= end_time):
            count += 1
            if self.is_test_stopped_by_webgui():
                logger.info("Test stopped by user via WebGUI. Exiting monitoring.")
                return False
            if count > 1:
                logger.info("Attempt {} of {} to check endpoints availability".format(count, no_of_attempts))
            endp_url = "/generic/{}".format(','.join(expected_endps))
            response = self.json_get(endp_url)
            if not response:
                logger.error(
                    "Failed to fetch endpoints. Received empty response.\n"
                    f"Requested URL: '{endp_url}'\n"
                    f"Response: {response}")
                response = {}
            endpoint = response.get('endpoints', response.get('endpoint', []))
            if isinstance(endpoint, dict):
                endpoint = [{endpoint['name']: endpoint}]
            present_endps = {}
            for endp_entry in endpoint:
                for endp_name, endp_response in endp_entry.items():
                    # Use the second-last result when more than one is available.
                    last_results = endp_response.get('last results')
                    if last_results:
                        self.record_packet_results(endp_name, last_results)
                        result_lines = last_results.split('\n')
                        if len(result_lines) > 1:
                            endp_response['last results'] = result_lines[-2]
                    present_endps[endp_name] = endp_response
            self.latest_results.update(present_endps)
            self.check_endpoint_availability(expected_endps, present_endps)
            missed = len(self.missing_endp_logged)
            not_run = len(self.not_running_endp_logged)
            if missed == len(expected_endps):
                logger.error("All expected endpoints ({}) are missing from the monitoring data. Retrying... Before giving up and stopping test".format(
                    ", ".join(self.missing_endp_logged)))
            elif not_run == len(expected_endps):
                logger.error("All expected endpoints ({}) are present but not running. Retrying... Before giving up and stopping test".format(
                    ", ".join(self.not_running_endp_logged)))
            elif missed + not_run == len(expected_endps):
                logger.error("Some expected endpoints ({}) are missing and some endpoints ({}) are not running. Retrying... Before giving up and stopping test".format(
                    ", ".join(self.missing_endp_logged), ", ".join(self.not_running_endp_logged)))
            else:
                return True
            time.sleep(interval)
            start_time = time.time()
        return False

    @staticmethod
    def validate_rtt(min_rtt, avg_rtt, max_rtt):
        """Check whether min/avg/max RTT values represent a real ping result.

        A ping that never got a reply shows up differently per OS: Linux/Mac
        report a negative min and/or max (e.g. min=1000000, max=-1000000,
        avg=0), while Windows reports a flat 0.000/0.000/0.000. Either
        signature means the ping did not succeed, so all three values are
        normalized to 'NA'.
        """
        try:
            min_val = float(str(min_rtt).replace(',', ''))
            avg_val = float(str(avg_rtt).replace(',', ''))
            max_val = float(str(max_rtt).replace(',', ''))
        except (TypeError, ValueError):
            return 'NA', 'NA', 'NA'
        if min_val < 0 or avg_val < 0 or max_val < 0 or (min_val == 0 and avg_val == 0 and max_val == 0):
            return 'NA', 'NA', 'NA'
        return min_rtt, avg_rtt, max_rtt

    def generate_remarks(self, station_ping_data):
        remarks = []

        # NOTE if there are any more ping failure cases that are missed, add them here.

        # checking if ping output is not empty
        if (station_ping_data['last_result'] == ""):
            remarks.append('No output for ping')

        # illegal division by zero error. Issue with arguments.
        if ('Illegal division by zero' in station_ping_data['last_result']):
            remarks.append('Illegal division by zero error. Please re-check the arguments passed.')

        # unknown host
        if ('Totals:  *** dropped: 0  received: 0  failed: 0  bytes: 0' in station_ping_data['last_result'] or 'unknown host' in station_ping_data['last_result']):
            remarks.append('Unknown host. Please re-check the target')

        # checking if IP is existing in the ping command or not for Windows device
        if (station_ping_data['os'] == 'Windows'):
            if ('None' in station_ping_data['command'] or station_ping_data['command'].split('-n')[0].split('-S')[-1] == "  "):
                remarks.append('Station has no IP')

        # network buffer overflow
        if ('ping: sendmsg: No buffer space available' in station_ping_data['last_result']):
            remarks.append('Network buffer overlow')

        # checking for no ping states (min/avg/max were normalized to 'NA' by validate_rtt)
        if (station_ping_data['min_rtt'] == 'NA'):

            # Destination Host Unreachable state
            if ('Destination Host Unreachable' in station_ping_data['last_result']):
                remarks.append('Destination Host Unrechable')

            # Destination Net Unreachable state
            if ('Destination Net Unreachable' in station_ping_data['last_result']):
                remarks.append('Destination Net Unreachable')

            # Name or service not known state
            if ('Name or service not known' in station_ping_data['last_result']):
                remarks.append('Name or service not known')

            # Temporary failure in name resolution (e.g. no network/DNS reachable)
            if ('Temporary failure in name resolution' in station_ping_data['last_result']):
                remarks.append('Temporary failure in name resolution')

            # fall back so an invalid/NA ping result is never left unexplained
            if (not remarks):
                remarks.append('Ping failed - invalid RTT statistics (min/avg/max: NA)')

        return (remarks)

    # Converts an upstream port name to its corresponding IP address if it's not already in IP format.
    def change_port_to_ip(self, upstream_port):
        if upstream_port.count('.') != 3:
            target_port_list = self.name_to_eid(upstream_port)
            shelf, resource, port, _ = target_port_list
            try:
                target_port_ip = self.json_get(f'/port/{shelf}/{resource}/{port}?fields=ip')['interface']['ip']
                upstream_port = target_port_ip
            except Exception:
                logger.warning(f'The upstream port is not an ethernet port. Proceeding with the given upstream_port {upstream_port}.')
            logger.info(f"Upstream port IP {upstream_port}")
        else:
            logger.info(f"Upstream port IP {upstream_port}")
        return upstream_port

    # Calculates pass/fail status for each client based on their result compared to the expected value.
    def get_pass_fail_list(self, os_type):
        # When csv_name is provided, for pass/fail criteria, respective values for each client will be used
        if not self.expected_passfail_val:
            res_list = []
            test_input_list = []
            pass_fail_list = []
            adb_url = '/adb/'
            response = self.json_get(adb_url)
            if not response:
                logger.error("Failed to fetch adb data.\nRequested URL: '{}'\nResponse: {}".format(adb_url, response))
                interop_tab_data = None
            else:
                interop_tab_data = response['devices']
                if isinstance(interop_tab_data, dict):
                    interop_tab_data = [{interop_tab_data['name']: interop_tab_data}]
            for client in range(len(os_type)):
                device_name = self.device_names[client].split(' ')[0:-1][0]
                if os_type[client] != 'Android':
                    res_list.append(device_name)
                else:
                    matched_name = None
                    if interop_tab_data is not None:
                        for dev in interop_tab_data:
                            for item in dev.values():
                                if item['user-name'] == device_name:
                                    matched_name = item['name'].split('.')[2]
                                    break
                            if matched_name is not None:
                                break
                    # Always append one entry so res_list stays aligned with os_type/packets_sent.
                    res_list.append(matched_name if matched_name is not None else device_name)
            with open(self.csv_name, mode='r') as file:
                reader = csv.DictReader(file)
                rows = list(reader)
                # fieldnames = reader.fieldnames
            for device in res_list:
                found = False
                for row in rows:
                    if row['DeviceList'] == device and row['PingPacketLoss %'].strip() != '':
                        test_input_list.append(row['PingPacketLoss %'])
                        found = True
                        break
                if not found:
                    logger.info(f"Ping result for device {device} not found in CSV. Using default packet loss = 10%")
                    test_input_list.append(10)
            self.percent_pac_loss = []
            for i in range(len(self.packets_sent)):
                if self.packets_sent[i] != 0:
                    self.percent_pac_loss.append(((self.packets_sent[i] - self.packets_received[i]) / self.packets_sent[i]) * 100)
                else:
                    self.percent_pac_loss.append(0)
            for i in range(len(test_input_list)):
                if self.packets_sent[i] == 0:
                    pass_fail_list.append('FAIL')
                elif float(test_input_list[i]) >= self.percent_pac_loss[i]:
                    pass_fail_list.append('PASS')
                else:
                    pass_fail_list.append('FAIL')
            self.pass_fail_list = pass_fail_list
            self.test_input_list = test_input_list
        # When expected_passfail_val is provided, for pass/fail criteria, the same value will be used for all clients
        else:
            self.test_input_list = [self.expected_passfail_val for val in range(len(self.device_names))]
            self.percent_pac_loss = []
            for i in range(len(self.packets_sent)):
                if self.packets_sent[i] != 0:
                    self.percent_pac_loss.append(((self.packets_sent[i] - self.packets_received[i]) / self.packets_sent[i]) * 100)
                else:
                    self.percent_pac_loss.append(0)
            pass_fail_list = []
            for i in range(len(self.test_input_list)):
                if self.packets_sent[i] == 0:
                    pass_fail_list.append('FAIL')
                elif float(self.expected_passfail_val) >= self.percent_pac_loss[i]:
                    pass_fail_list.append("PASS")
                else:
                    pass_fail_list.append("FAIL")
            self.pass_fail_list = pass_fail_list

    def add_live_view_images_to_report(self, report: lf_report, report_path: str):
        """
        This function looks for throughput and RSSI images for each floor
        in the 'live_view_images' folder within `self.result_dir`.
        It waits up to **60 seconds** for each image. If an image is found,
        it's added to the `report` on a new page; otherwise, it's skipped.
        """
        test_name = os.path.basename(report_path)
        for floor in range(int(self.total_floors)):
            # Construct expected image paths
            packet_sent_image = os.path.join(self.result_dir, "heatmap_images", f"{test_name}_ping_packet_sent_{floor + 1}.png")
            packet_recv_image = os.path.join(self.result_dir, "heatmap_images", f"{test_name}_ping_packet_recv_{floor + 1}.png")
            packet_loss_image = os.path.join(self.result_dir, "heatmap_images", f"{test_name}_ping_packet_loss_{floor + 1}.png")

            # Wait for all required images to be generated (up to timeout)
            timeout = 60  # seconds
            start_time = time.time()

            while not (os.path.exists(packet_sent_image) and os.path.exists(packet_recv_image) and os.path.exists(packet_loss_image)):
                if time.time() - start_time > timeout:
                    print(f"Timeout: Heatmap images for floor {floor + 1} not found within {timeout} seconds.")
                    break
                time.sleep(1)

            report.set_custom_html("<h2>Ping Packet Sent vs Recevied vs Lost: </h2>")
            report.build_custom()

            # Generate report sections for each image if it exists
            for image_path in [packet_sent_image, packet_recv_image, packet_loss_image]:
                if os.path.exists(image_path):
                    report.set_custom_html(f'<img src="file://{image_path}"  style="width:1200px; height:800px;"></img>')
                    report.build_custom()

    def generate_report(self, result_json=None, result_dir='Ping_Test_Report', report_path='', config_devices='', group_device_map=None):
        if result_json is not None:
            self.result_json = result_json
        logger.info('Generating Report')

        report = lf_report(_output_pdf='interop_ping.pdf',
                           _output_html='interop_ping.html',
                           _results_dir_name=result_dir,
                           _path=report_path)
        report_path = report.get_path()
        report_path_date_time = report.get_path_date_time()

        self.write_client_issue_csv(report_path_date_time)

        logger.info('path: {}'.format(report_path))
        logger.info('path_date_time: {}'.format(report_path_date_time))

        # setting report title
        report.set_title('Ping Test Report')
        report.build_banner()

        # Test setup information table for devices in device list
        if config_devices == '':
            test_setup_info = {
                'SSID': self.ssid,
                'Security': self.security,
                'Website / IP': self.target,
                'No of Devices': '{} (V:{}, A:{}, W:{}, L:{}, M:{})'.format(len(self.sta_list), len(self.sta_list) - len(self.real_sta_list), self.android, self.windows, self.linux, self.mac),
                'Duration (in minutes)': self.duration
            }
        # Test setup information table for devices in groups
        else:
            group_names = ', '.join(config_devices.keys())
            profile_names = ', '.join(config_devices.values())
            configmap = "Groups:" + group_names + " -> Profiles:" + profile_names
            test_setup_info = {
                'Configuration': configmap,
                'Website / IP': self.target,
                'No of Devices': '{} (V:{}, A:{}, W:{}, L:{}, M:{})'.format(len(self.sta_list), len(self.sta_list) - len(self.real_sta_list), self.android, self.windows, self.linux, self.mac),
                'Duration (in minutes)': self.duration
            }
        report.test_setup_table(
            test_setup_data=test_setup_info, value='Test Setup Information')

        # objective and description
        report.set_obj_html(_obj_title='Objective',
                            _obj='''The objective of the ping test is to evaluate network connectivity and measure the round-trip time taken for
                            data packets to travel from the source to the destination and back. It helps assess the reliability and latency of the network,
                            identifying any packet loss, delays, or variations in response times. The test aims to ensure that devices can communicate
                            effectively over the network and pinpoint potential issues affecting connectivity.
                            ''')
        report.build_objective()

        # packets sent vs received vs dropped
        report.set_table_title(
            'Packets sent vs packets received vs packets dropped')
        report.build_table_title()
        # graph for the above
        self.packets_sent = []
        self.packets_received = []
        self.packets_dropped = []
        self.device_names = []
        self.device_modes = []
        self.device_channels = []
        self.device_min = []
        self.device_max = []
        self.device_avg = []
        self.device_mac = []
        self.device_names_with_errors = []
        self.devices_with_errors = []
        self.report_names = []
        self.remarks = []
        self.device_ssid = []
        # packet_count_data = {}
        os_type = []
        for device, device_data in self.result_json.items():
            logger.debug('Device data: {} {}'.format(device, device_data))
            os_type.append(device_data['os'])
            self.packets_sent.append(int(device_data['sent']))
            self.packets_received.append(int(device_data['recv']))
            self.packets_dropped.append(int(device_data['dropped']))
            self.device_names.append(device_data['name'] + ' ' + device_data['os'])
            self.device_modes.append(device_data['mode'])
            self.device_channels.append(device_data['channel'])
            self.device_mac.append(device_data['mac'])
            self.device_ssid.append(device_data['ssid'])
            # NA (failed ping) is kept as text here for the Latency table; the bar
            # graph below converts NA to 0 right before it's plotted.
            self.device_min.append(device_data['min_rtt'] if device_data['min_rtt'] == 'NA' else float(device_data['min_rtt'].replace(',', '')))
            self.device_max.append(device_data['max_rtt'] if device_data['max_rtt'] == 'NA' else float(device_data['max_rtt'].replace(',', '')))
            self.device_avg.append(device_data['avg_rtt'] if device_data['avg_rtt'] == 'NA' else float(device_data['avg_rtt'].replace(',', '')))
            if (device_data['os'] == 'Virtual'):
                self.report_names.append('{} {}'.format(device, device_data['os'])[0:25])
            else:
                self.report_names.append('{} {} {}'.format(device, device_data['os'], device_data['name']))
            if (device_data['remarks'] != []):
                self.device_names_with_errors.append(device_data['name'])
                self.devices_with_errors.append(device)
                self.remarks.append(','.join(device_data['remarks']))
            # logging.info(self.packets_sent,
            #       self.packets_received,
            #       self.packets_dropped)
            # logging.info(self.device_min,
            #       self.device_max,
            #       self.device_avg)

            # packet_count_data[device] = {
            #     'MAC': device_data['mac'],
            #     'Channel': device_data['channel'],
            #     'Mode': device_data['mode'],
            #     'Packets Sent': device_data['sent'],
            #     'Packets Received': device_data['recv'],
            #     'Packets Loss': device_data['dropped'],
            # }
        x_fig_size = 15
        y_fig_size = len(self.device_names) * .5 + 4
        graph = lf_bar_graph_horizontal(_data_set=[self.packets_dropped, self.packets_received, self.packets_sent],
                                        _xaxis_name='Packets Count',
                                        _yaxis_name='Wireless Clients',
                                        _label=[
                                            'Packets Loss', 'Packets Received', 'Packets Sent'],
                                        _graph_image_name='Packets sent vs received vs dropped',
                                        _yaxis_label=self.report_names,
                                        _yaxis_categories=self.report_names,
                                        _yaxis_step=1,
                                        _yticks_font=8,
                                        _graph_title='Packets sent vs received vs dropped',
                                        _title_size=16,
                                        _color=['lightgrey',
                                                'orange', 'steelblue'],
                                        _color_edge=['black'],
                                        _bar_height=0.15,
                                        _figsize=(x_fig_size, y_fig_size),
                                        _legend_loc="best",
                                        _legend_box=(1.0, 1.0),
                                        _dpi=96,
                                        _show_bar_value=False,
                                        _enable_csv=True,
                                        _color_name=['lightgrey', 'orange', 'steelblue'])

        graph_png = graph.build_bar_graph_horizontal()
        logger.info('graph name {}'.format(graph_png))
        report.set_graph_image(graph_png)
        # need to move the graph image to the results directory
        report.move_graph_image()
        report.set_csv_filename(graph_png)
        report.move_csv_file()
        report.build_graph()

        if self.real:
            # Calculating the pass/fail criteria when either expected_passfail_val or csv_name is provided
            if self.expected_passfail_val or self.csv_name:
                self.get_pass_fail_list(os_type)
            # When groups are provided a seperate table will be generated for each group using generate_dataframe
            if self.group_name:
                for key, val in group_device_map.items():
                    if self.expected_passfail_val or self.csv_name:
                        dataframe = self.generate_dataframe(
                            val,
                            self.device_names,
                            self.device_mac,
                            self.device_channels,
                            self.device_ssid,
                            self.device_modes,
                            self.packets_sent,
                            self.packets_received,
                            self.packets_dropped,
                            self.percent_pac_loss,
                            self.test_input_list,
                            self.pass_fail_list)
                    else:
                        dataframe = self.generate_dataframe(val, self.device_names, self.device_mac, self.device_channels, self.device_ssid,
                                                            self.device_modes, self.packets_sent, self.packets_received, self.packets_dropped, [], [], [])
                    if dataframe:
                        report.set_obj_html("", "Group: {}".format(key))
                        report.build_objective()
                        dataframe1 = pd.DataFrame(dataframe)
                        report.set_table_dataframe(dataframe1)
                        report.build_table()

            else:
                dataframe1 = pd.DataFrame({
                    'Wireless Client': self.device_names,
                    'MAC': self.device_mac,
                    'Channel': self.device_channels,
                    'SSID ': self.device_ssid,
                    'Mode': self.device_modes,
                    'Packets Sent': self.packets_sent,
                    'Packets Received': self.packets_received,
                    'Packets Loss': self.packets_dropped,
                })
                if self.expected_passfail_val or self.csv_name:
                    dataframe1[" Percentage of Packet loss %"] = self.percent_pac_loss
                    dataframe1['Expected Packet loss %'] = self.test_input_list
                    dataframe1['Status'] = self.pass_fail_list
                report.set_table_dataframe(dataframe1)
                report.build_table()
            if self.get_live_view:
                self.add_live_view_images_to_report(report=report, report_path=report_path)

        else:
            dataframe1 = pd.DataFrame({
                'Wireless Client': self.device_names,
                'MAC': self.device_mac,
                'Channel': self.device_channels,
                'SSID ': self.device_ssid,
                'Mode': self.device_modes,
                'Packets Sent': self.packets_sent,
                'Packets Received': self.packets_received,
                'Packets Loss': self.packets_dropped,
            })
            report.set_table_dataframe(dataframe1)
            report.build_table()

        # packets latency graph
        report.set_table_title('Ping Latency Graph')
        report.build_table_title()

        # the graph can't plot the text 'NA', so failed-ping devices show as 0 here;
        # the Latency table below (and the Notes table) keep 'NA' for those devices
        plot_min = [0.0 if min_val == 'NA' else min_val for min_val in self.device_min]
        plot_avg = [0.0 if avg_val == 'NA' else avg_val for avg_val in self.device_avg]
        plot_max = [0.0 if max_val == 'NA' else max_val for max_val in self.device_max]

        graph = lf_bar_graph_horizontal(_data_set=[plot_min, plot_avg, plot_max],
                                        _xaxis_name='Time (ms)',
                                        _yaxis_name='Wireless Clients',
                                        _label=[
                                            'Min Latency (ms)', 'Average Latency (ms)', 'Max Latency (ms)'],
                                        _graph_image_name='Ping Latency per client',
                                        _yaxis_label=self.report_names,
                                        _yaxis_categories=self.report_names,
                                        _yaxis_step=1,
                                        _yticks_font=8,
                                        _graph_title='Ping Latency per client',
                                        _title_size=16,
                                        _color=['lightgrey',
                                                'orange', 'steelblue'],
                                        _color_edge='black',
                                        _bar_height=0.15,
                                        _figsize=(x_fig_size, y_fig_size),
                                        _legend_loc="best",
                                        _legend_box=(1.0, 1.0),
                                        _dpi=96,
                                        _show_bar_value=False,
                                        _enable_csv=True,
                                        _color_name=['lightgrey', 'orange', 'steelblue'])

        graph_png = graph.build_bar_graph_horizontal()
        logger.info('graph name {}'.format(graph_png))
        report.set_graph_image(graph_png)
        # need to move the graph image to the results directory
        report.move_graph_image()
        report.set_csv_filename(graph_png)
        report.move_csv_file()
        report.build_graph()

        dataframe2 = pd.DataFrame({
            'Wireless Client': self.device_names,
            'MAC': self.device_mac,
            'Channel': self.device_channels,
            'SSID ': self.device_ssid,
            'Mode': self.device_modes,
            'Min Latency (ms)': self.device_min,
            'Average Latency (ms)': self.device_avg,
            'Max Latency (ms)': self.device_max
        })
        report.set_table_dataframe(dataframe2)
        report.build_table()

        # only show the NA caveat when at least one device actually has NA latency
        if ('NA' in self.device_min):
            report.set_text("Note: Stations which are not reachable to the internet, and the ping failed to receive any packets, "
                            "resulting in 100% packet loss. Hence, the latency is reported as NA.")
            report.build_text_simple()

        # check if there are remarks for any device. If there are remarks, build table else don't
        if (self.remarks != []):
            report.set_table_title('Notes')
            report.build_table_title()
            dataframe3 = pd.DataFrame({
                'Wireless Client': self.device_names_with_errors,
                'Port': self.devices_with_errors,
                'Remarks': self.remarks
            })
            report.set_table_dataframe(dataframe3)
            report.build_table()

        # closing
        report.build_custom()
        report.build_footer()
        report.write_html()
        report.write_pdf()

    def generate_dataframe(self, groupdevlist: List[str], device_names: List[str], device_mac: List[str], device_channels: List[str], device_ssid: List[str], device_modes: List[str],
                           packets_sent: List[int], packets_received: List[int], packets_dropped: List[int], percent_pac_loss: List[float], test_input_list: List[str],
                           pass_fail_list: List[str]) -> Optional[pd.DataFrame]:
        """
        Creates a separate DataFrame for each group of devices.

        Returns:
            DataFrame: A DataFrame for each device group.
            Returns None if neither device in a group is configured.
        """
        dev_names = []
        dev_mac = []
        dev_channels = []
        dev_ssid = []
        dev_modes = []
        pack_sent = []
        pack_received = []
        pac_dropped = []
        pac_loss = []
        input_list = []
        pass_fail = []
        adb_url = '/adb/'
        response = self.json_get(adb_url)
        if not response:
            logger.error("Failed to fetch adb data.\nRequested URL: '{}'\nResponse: {}".format(adb_url, response))
            interop_tab_data = None
        else:
            interop_tab_data = response['devices']
            if isinstance(interop_tab_data, dict):
                interop_tab_data = [{interop_tab_data['name']: interop_tab_data}]
        for i in range(len(device_names)):
            for j in groupdevlist:
                # For a string like "1.360 Lin test3":
                # - device_names[i].split(" ")[0:-1][0] gives 'test3' (device name)
                # - device_names[i].split(" ")[-1] gives 'Lin' (OS type)
                # This condition filters out Android clients and matches device name with j
                if j == device_names[i].split(" ")[0:-1][0] and device_names[i].split(" ")[-1] != 'Android':
                    dev_names.append(device_names[i])
                    dev_mac.append(device_mac[i])
                    dev_channels.append(device_channels[i])
                    dev_ssid.append(device_ssid[i])
                    dev_modes.append(device_modes[i])
                    pack_sent.append(packets_sent[i])
                    pack_received.append(packets_received[i])
                    pac_dropped.append(packets_dropped[i])
                    if self.expected_passfail_val or self.csv_name:
                        pac_loss.append(percent_pac_loss[i])
                        input_list.append(test_input_list[i])
                        pass_fail.append(pass_fail_list[i])
                elif interop_tab_data is not None:
                    for dev in interop_tab_data:
                        for item in dev.values():
                            # For a string like 1.15 android samsungmob:
                            # - device_names[i].split(' ')[0:-1][0] (e.g., 'samsungmob') matches item['user-name']
                            # - The group name (e.g., 'RZCTA09CTXF') matches with item['name'].split('.')[-1]
                            if item['user-name'] == device_names[i].split(' ')[0:-1][0] and j == item['name'].split('.')[-1]:
                                dev_names.append(device_names[i])
                                dev_mac.append(device_mac[i])
                                dev_channels.append(device_channels[i])
                                dev_ssid.append(device_ssid[i])
                                dev_modes.append(device_modes[i])
                                pack_sent.append(packets_sent[i])
                                pack_received.append(packets_received[i])
                                pac_dropped.append(packets_dropped[i])
                                if self.expected_passfail_val or self.csv_name:
                                    pac_loss.append(percent_pac_loss[i])
                                    input_list.append(test_input_list[i])
                                    pass_fail.append(pass_fail_list[i])
        if len(dev_names) != 0:
            dataframe = {
                'Wireless Client': dev_names,
                'MAC': dev_mac,
                'Channel': dev_channels,
                'SSID ': dev_ssid,
                'Mode': dev_modes,
                'Packets Sent': pack_sent,
                'Packets Received': pack_received,
                'Packets Loss': pac_dropped,
            }
            if self.expected_passfail_val or self.csv_name:
                dataframe[' Percentage of Packet loss %'] = pac_loss
                dataframe['Expected Packet loss %'] = input_list
                dataframe['Status '] = pass_fail
            return dataframe
        else:
            return None

    def start_background_ping(self, device_list):
        """Starts a ping on the real clients of the calling test. Returns True when it is running."""
        self.generic_endps_profile.name_prefix = self.BG_NAME_PREFIX

        if not self.target:
            logger.error("No background ping target was given, the background ping is disabled")
            return False

        if not self.check_tab_exists():
            logger.error("The Generic tab is not available, skipping the background ping")
            return False

        try:
            self.change_target_to_ip()
        except RuntimeError as e:
            logger.warning("Could not resolve the background ping target {}, using it as given: {}".format(self.target, e))
        self.generic_endps_profile.dest = self.target
        self.generic_endps_profile.interval = self.interval

        try:
            devices = RealDevice(manager_ip=self.host, selected_bands=[])
            devices.get_devices()
        except Exception as e:
            logger.error("Background ping could not read the devices from the LANforge, skipping it: {}".format(e))
            return False

        pingable = []
        for port in device_list or []:
            hw_version = devices.devices_data.get(port, {}).get('hw version', '')
            if port not in devices.devices_data or (hw_version.startswith('Apple') and 'Apple/x86' not in hw_version):
                logger.warning("Skipping {} for the background ping, it is not a real client that can run a ping".format(port))
            else:
                pingable.append(port)
        if not pingable:
            logger.error("No real devices were found for the background ping, skipping it")
            return False

        self.select_real_devices(real_devices=devices, real_sta_list=pingable, base_interop_obj=devices)
        self.sta_list = list(self.real_sta_list)
        self.bg_packet_results = {device: {} for device in self.real_sta_list}

        try:
            self.cleanup()
            self.create_generic_endp()
            for endpoint in self.generic_endps_profile.created_endp:
                self.generic_endps_profile.set_report_timer(endp_name=endpoint, timer=250)
            self.start_generic()
        except Exception as e:
            logger.error("Background ping could not be started: {}".format(e))
            return False

        self.bg_start_time = time.time()
        self.bg_poll_interval = min(20, max(1, 50 * float(self.interval)))
        if self.bg_poll_stop is None:
            self.bg_poll_stop = threading.Event()
        self.bg_poll_stop.clear()
        self.bg_poll_thread = threading.Thread(target=self.poll_background_ping, daemon=True)
        self.bg_poll_thread.start()
        logger.info("Background ping started on {} towards {}".format(self.real_sta_list, self.target))
        return True

    def poll_background_ping(self):
        """Polls the background ping endpoints until the ping is stopped, warning when one is not running."""
        reported_status = {}
        while not self.bg_poll_stop.wait(self.bg_poll_interval):
            try:
                for entry in self.get_results():
                    for endpoint, data in entry.items():
                        status = str(data.get('status'))
                        if time.time() - self.bg_start_time > 15:
                            if status.lower() != 'run' and reported_status.get(endpoint) != status:
                                logger.warning("Background ping endpoint {} is {} while the test is running".format(endpoint, status))
                            reported_status[endpoint] = status
            except Exception as e:
                logger.warning("Background ping polling failed: {}".format(e))

    def record_packet_results(self, endpoint, last_results):
        """Records the RTT of each ping reply in an endpoint's output, and marks a sequence lost where the running drop count rises across a gap."""
        sequences = self.bg_packet_results.get(endpoint[len(self.BG_NAME_PREFIX) + 1:])
        if sequences is None:
            return
        previous = None
        for line in last_results.split('\n'):
            seq_match = re.search(r'icmp_seq=(\d+)', line)
            rtt_match = re.search(r'time[=<]\s*([0-9.]+)\s*ms', line, re.IGNORECASE)
            drop_match = re.search(r'drop:\s*(\d+)', line)
            if not seq_match or not rtt_match:
                continue
            sequence = int(seq_match.group(1))
            sequences[sequence] = float(rtt_match.group(1))
            if not drop_match:
                continue
            drops = int(drop_match.group(1))
            if previous is not None and sequence > previous[0]:
                new_drops = min(sequence - previous[0] - 1, drops - previous[1])
                for lost in range(previous[0] + 1, previous[0] + 1 + new_drops):
                    sequences.setdefault(lost, 0.0)
            if previous is None or sequence > previous[0]:
                previous = (sequence, drops)

    @staticmethod
    def get_packet_counts(data):
        """Returns (sent, received, dropped) for an endpoint reading."""
        sent = int(data.get('tx pkts') or 0)
        received = int(data.get('rx pkts') or 0)
        dropped = int(data.get('dropped') or 0)
        rx_match = re.search(r'rx:\s*(\d+)', data.get('last results') or '')
        drop_match = re.search(r'drop:\s*(\d+)', data.get('last results') or '')
        if rx_match and drop_match and int(rx_match.group(1)) + int(drop_match.group(1)) > sent:
            received, dropped = int(rx_match.group(1)), int(drop_match.group(1))
            sent = received + dropped
        return sent, received, dropped

    def stop_background_ping(self):
        """Stops the background ping and collects the per-client results. Warns for any client with no ping data."""
        if self.bg_poll_stop is not None:
            self.bg_poll_stop.set()
        if self.bg_poll_thread:
            self.bg_poll_thread.join(timeout=10)
        during_test = dict(self.latest_results)

        before_stop = {}
        for entry in self.get_results():
            for endpoint, data in entry.items():
                before_stop[endpoint] = dict(data)

        self.bg_stop_time = time.time()
        self.stop_generic()
        time.sleep(3)

        after_stop = {}
        for entry in self.get_results():
            for endpoint, data in entry.items():
                after_stop[endpoint] = data

        self.bg_stats = {}
        for device in self.real_sta_list:
            endpoint = self.BG_NAME_PREFIX + '-' + device
            readings = [snapshot[endpoint] for snapshot in (during_test, before_stop, after_stop) if snapshot.get(endpoint)]
            readings.sort(key=lambda reading: self.get_packet_counts(reading)[0], reverse=True)
            data = readings[0] if readings else {}
            sent, received, dropped = self.get_packet_counts(data) if data else (0, 0, 0)
            final_sent = self.get_packet_counts(after_stop[endpoint])[0] if after_stop.get(endpoint) else 0
            if sent > final_sent:
                logger.info("Background ping endpoint for {} showed {} sent after the stop but {} sent earlier, using the fuller earlier reading".format(
                    device, final_sent, sent))
            last_result = data.get('last results') or ''
            rtt_line = next((reading['last results'] for reading in readings if 'min/avg/max' in (reading.get('last results') or '')), '')
            min_rtt, avg_rtt, max_rtt = 'NA', 'NA', 'NA'
            if rtt_line:
                rtt_values = rtt_line.split('min/avg/max:', 1)[1].strip().split()[0].split('/')
                if len(rtt_values) == 3:
                    min_rtt, avg_rtt, max_rtt = self.validate_rtt(rtt_values[0], rtt_values[1], rtt_values[2])

            eid = self.name_to_eid(device)
            response = self.json_get('/port/{}/{}/{}?fields=mac,channel,signal,phantom,down,ip'.format(eid[0], eid[1], eid[2]))
            port_data = response.get('interface', {}) if response else {}
            device_info = self.real_sta_data_dict[device]
            entry = {
                'device': device,
                'name': device_info['user'] or device_info['hostname'],
                'mac': port_data.get('mac', ''),
                'channel': port_data.get('channel', ''),
                'rssi': str(port_data.get('signal', '')).split(' ')[0],
                'os': {'windows': 'Windows', 'macos': 'Mac', 'linux': 'Linux', 'android': 'Android'}[device_info['ostype']],
                'command': data.get('command', ''),
                'sent': sent,
                'recv': received,
                'dropped': dropped,
                'loss_percent': round(((sent - received) / sent) * 100, 2) if sent else 0,
                'min_rtt': min_rtt if min_rtt == 'NA' else float(str(min_rtt).replace(',', '')),
                'avg_rtt': avg_rtt if avg_rtt == 'NA' else float(str(avg_rtt).replace(',', '')),
                'max_rtt': max_rtt if max_rtt == 'NA' else float(str(max_rtt).replace(',', '')),
                'last_result': last_result,
            }
            entry['remarks'] = self.generate_remarks(entry)
            self.bg_stats[device] = entry
            logger.info("Background ping result for {} ({}): sent={}, received={}, dropped={}, min/avg/max={}/{}/{}".format(
                device, entry['name'], sent, received, dropped, min_rtt, avg_rtt, max_rtt))

            if not last_result:
                logger.warning("No background ping data for {} ({}): endpoint {} returned no output. Port manager shows phantom={}, down={}, ip={}. "
                               "Remarks: {}".format(device, entry['name'], self.BG_NAME_PREFIX + '-' + device, port_data.get('phantom', 'unknown'),
                                                    port_data.get('down', 'unknown'), port_data.get('ip', 'unknown'), entry['remarks']))
            elif entry['remarks']:
                logger.warning("Background ping on {} ({}) reported: {}".format(device, entry['name'], entry['remarks']))

        return self.bg_stats

    def add_background_ping_to_report(self, report):
        """Adds the background ping statistics table to the report. Safe to call when the background ping did not run."""
        if not self.bg_stats:
            return False

        rows = list(self.bg_stats.values())
        dataframe = pd.DataFrame({
            'Wireless Client': [row['name'] for row in rows],
            'MAC': [row['mac'] for row in rows],
            'RSSI (dBm)': [row['rssi'] for row in rows],
            'Channel': [row['channel'] for row in rows],
            'Packets Sent': [row['sent'] for row in rows],
            'Packets Received': [row['recv'] for row in rows],
            'Packet Loss %': [row['loss_percent'] for row in rows],
            'AVG RTT (ms)': [row['avg_rtt'] for row in rows],
        })
        test_seconds = self.bg_stop_time - self.bg_start_time
        interval = float(self.interval)
        duration = str(datetime.timedelta(seconds=int(test_seconds)))
        clients, stations, segments = [], [], []
        for device in self.real_sta_list:
            sequences = self.bg_packet_results.get(device)
            if not sequences:
                continue
            spans = []
            covered_until = 0.0
            for sequence in sorted(sequences):
                start, end = (sequence - 1) * interval, min(test_seconds, sequence * interval)
                if end <= start:
                    continue
                if start > covered_until:
                    spans.append(['up', covered_until, start])
                spans.append(['drop' if sequences[sequence] == 0 else 'up', start, end])
                covered_until = max(covered_until, end)
            merged = []
            for span in spans:
                if merged and merged[-1][0] == span[0]:
                    merged[-1][2] = span[2]
                else:
                    merged.append(span)
            segments.extend({'clientIndex': len(clients), 'start': round(start, 3), 'end': round(end, 3), 'status': status} for status, start, end in merged)
            clients.append(self.bg_stats[device]['name'])
            stations.append(device)
        try:
            if segments and hasattr(report, 'build_echarts_chart'):
                report.set_obj_html(
                    _obj_title="Client Connectivity Results Throughout the Test Duration",
                    _obj=("The graph illustrates the connectivity status of all wireless clients during the test based on connectivity monitoring during the test execution. "
                          "Green segments represent successful responses, while red segments indicate packet loss or connectivity drops observed during the test."))
                report.build_objective()
                report.build_echarts_chart(
                    chart_id="background-ping-connectivity-timeline",
                    chart_type="connectivity_timeline",
                    payload={'clients': clients, 'stations': stations, 'segments': segments, 'duration': round(test_seconds, 3),
                             'emptyMessage': "No time-series ping samples were collected"},
                    title="Wireless Client Connectivity Status vs Time")
            report.set_table_title("Ping statistics collected for all wireless clients during the test (target {}, duration {}, {}s interval)".format(
                self.target, duration, self.interval))
            report.build_table_title()
            report.set_table_dataframe(dataframe)
            report.build_table()
        except Exception as e:
            logger.warning("Background ping statistics could not be added to the report: {}".format(e))
            return False

        return True


def validate_args(args):
    # input sanity
    if args.virtual is False and args.real is False:
        logger.error('Atleast one of --real or --virtual is required')
        exit(1)
    if args.virtual is True and args.radio is None:
        logger.error('--radio required')
        exit(1)
    if args.virtual is True and args.ssid is None:
        logger.error('--ssid required for virtual stations')
        exit(1)
    if args.ssid and args.passwd and args.group_name and args.profile_name:
        logger.error('either --ssid,--password,--security or --profile_name,--group_name should be given')
        exit(1)
    if args.use_default_config is False and args.group_name is None and args.file_name is None and args.profile_name is None:
        if args.ssid is None:
            logger.error('--ssid required for Wi-Fi configuration')
            exit(1)

        if args.security.lower() != 'open' and args.passwd == '[BLANK]':
            logger.error('--passwd required for Wi-Fi configuration')
            exit(1)

        if args.server_ip is None:
            logger.error('--server_ip or upstream ip required for Wi-fi configuration')
            exit(1)
        if args.group_name and (args.file_name is None or args.profile_name is None):
            logger.error("Please provide file name and profile name for group configuration")
            exit(1)
        elif args.file_name and (args.group_name is None or args.profile_name is None):
            logger.error("Please provide group name and profile name for file configuration")
            exit(1)
        elif args.profile_name and (args.group_name is None or args.file_name is None):
            logger.error("Please provide group name and file name for profile configuration")
            exit(1)
    # Get group and profile values from arguments and convert comma-separated strings into lists
    if args.group_name:
        selected_groups = args.group_name.split(',')
    else:
        selected_groups = []  # Default to empty list if group name is not provided
    if args.profile_name:
        selected_profiles = args.profile_name.split(',')
    else:
        selected_profiles = []  # Default to empty list if profile name is not provided
    if len(selected_groups) != len(selected_profiles):
        logger.error("Number of groups should match number of profiles")
        exit(1)
    if args.device_csv_name and args.expected_passfail_value:
        logger.error("Enter either --device_csv_name or --expected_passfail_value")
        exit(1)


def main():

    help_summary = '''\
The Candela Tech ping test is to evaluate network connectivity and measure the round-trip time taken for
data packets to travel from the source to the destination and back. It helps assess the reliability and latency of the network,
identifying any packet loss, delays, or variations in response times. The test aims to ensure that devices can communicate
effectively over the network and pinpoint potential issues affecting connectivity.
    '''

    parser = argparse.ArgumentParser(
        prog='interop_ping.py',
        formatter_class=argparse.RawTextHelpFormatter,
        epilog='''
            Allows user to run the ping test on a target IP or port for the given duration and packet interval
            with either selected number of virtual stations or provides the list of available real devices
            and allows the user to select the real devices and run ping test on them.
        ''',
        description='''
        NAME: lf_interop_ping.py

        PURPOSE: lf_interop_ping.py will let the user select real devices, virtual devices or both and then allows them to run
        ping test for user given duration and packet interval on the given target IP or domain name.

        EXAMPLE-1:
        Command Line Interface to run ping test with only virtual clients
        python3 lf_interop_ping.py --mgr 192.168.200.103  --target 192.168.1.3 --virtual --num_sta 1 --radio 1.1.wiphy2 --ssid RDT_wpa2 --security wpa2
        --passwd OpenWifi --ping_interval 1 --ping_duration 1 --server_ip 192.168.1.61 --debug

        EXAMPLE-2:
        Command Line Interface to run ping test with only real clients
        python3 lf_interop_ping.py --mgr 192.168.200.103 --real --target 192.168.1.3 --ping_interval 1 --ping_duration 1 --server_ip 192.168.1.61 --ssid RDT_wpa2
        --security wpa2_personal --passwd OpenWifi

        EXAMPLE-3:
        Command Line Interface to run ping test with both real and virtual clients
        python3 lf_interop_ping.py --mgr 192.168.200.103 --target 192.168.1.3 --real --virtual --num_sta 1 --radio 1.1.wiphy2 --ssid RDT_wpa2 --security wpa2
        --passwd OpenWifi --ping_interval 1 --ping_duration 1 --server_ip 192.168.1.61

        EXAMPLE-4:
        Command Line Interface to run ping test with existing Wi-Fi configuration on the real devices
        python3 lf_interop_ping.py --mgr 192.168.200.63 --real --target 192.168.1.61 --ping_interval 5 --ping_duration 1 --passwd OpenWifi --use_default_config

        EXAMPLE-5:
        Command Line Interface to run ping test by setting device specific Pass/Fail values in the csv file
        python3 lf_interop_ping.py --mgr 192.168.244.97 --real --target 192.168.1.3 --ping_interval 1 --ping_duration 1 --device_csv_name device.csv
        --use_default_config

        EXAMPLE-6:
        Command Line Interface to run ping test by setting the same expected Pass/Fail value for all devices
        python3 lf_interop_ping.py --mgr 192.168.244.97 --real --target 192.168.1.3 --ping_interval 1 --ping_duration 1 --expected_passfail_value 3
        --use_default_config

        EXAMPLE-7:
        Command Line Interface to run ping test by configuring Real Devices with SSID, Password, and Security
        python3 lf_interop_ping.py --mgr 192.168.244.97 --real --target 192.168.1.3 --ping_interval 1 --ping_duration 1  --ssid RDT_wpa2 --security wpa2
        --passwd OpenWifi --server_ip 192.168.244.97 --wait_time 30

        EXAMPLE-8:
        Command Line Interface to run ping test by Configuring Devices in Groups with Specific Profiles
        python3 lf_interop_ping.py --mgr 192.168.244.97 --real --target 192.168.1.3 --ping_interval 1 --ping_duration 1   --group_name grp3 --file_name g219 --profile_name Open5
        --server_ip 192.168.204.60

        EXAMPLE-9:
        Command Line Interface to run ping test by Configuring Devices in Groups with Specific Profiles with expected Pass/Fail values
        python3 lf_interop_ping.py --mgr 192.168.244.97 --real --target 192.168.1.3 --ping_interval 1 --ping_duration 1   --group_name grp3 --file_name g219 --profile_name Open5
        --expected_passfail_value 3 --server_ip 192.168.204.60

        EXAMPLE-10:
        Command Line Interface for Configuring Devices in Groups with Specific Profiles with device_csv_name
        python3 lf_interop_ping.py --mgr 192.168.244.97 --real --target 192.168.1.3 --ping_interval 1 --ping_duration 1   --group_name grp3 --file_name g219 --profile_name Open5
        --device_csv_name device.csv --server_ip 192.168.204.60

        SCRIPT_CLASSIFICATION : Test

        SCRIPT_CATEGORIES: Performance, Functional, Report Generation

        NOTES:
        1.Use './lf_interop_ping.py --help' to see command line usage and options
        2.Please pass ping_duration in minutes
        3.Please pass ping_interval in seconds
        4.After passing the cli, if --real flag is selected, then a list of available real devices will be displayed on the terminal.
        5.Enter the real device resource numbers seperated by commas (,)

        STATUS: BETA RELEASE

        VERIFIED_ON:
        Working date    - 20/09/2023
        Build version   - 5.4.7
        kernel version  - 6.2.16+

        License: Free to distribute and modify. LANforge systems must be licensed.
        Copyright (C) 2020-2026 Candela Technologies Inc.
        '''
    )
    # required = parser.add_argument_group('Required arguments')
    optional = parser.add_argument_group('Optional arguments')

    # optional arguments
    optional.add_argument('--mgr',
                          type=str,
                          help='hostname where LANforge GUI is running',
                          default='localhost')

    optional.add_argument('--target',
                          type=str,
                          help='Target URL or port for ping test',
                          default='1.1.eth1')

    optional.add_argument('--ping_interval',
                          type=str,
                          help='Interval (in seconds) between the echo requests',
                          default='1')

    optional.add_argument('--ping_duration',
                          type=float,
                          help='Duration (in minutes) to run the ping test',
                          default=1)

    optional.add_argument('--ssid',
                          type=str,
                          help='SSID for connecting the stations')

    optional.add_argument('--mgr_port',
                          type=str,
                          default=8080,
                          help='port on which LANforge HTTP service is running'
                          )

    optional.add_argument('--mgr_passwd',
                          type=str,
                          default='lanforge',
                          help='Password to connect to LANforge GUI')

    optional.add_argument('--server_ip',
                          type=str,
                          help='Upstream for configuring the Interop App')

    optional.add_argument('--security',
                          type=str,
                          default='open',
                          help='Security protocol for the specified SSID: <open | wep | wpa | wpa2 | wpa3>')

    optional.add_argument('--passwd',
                          type=str,
                          default='[BLANK]',
                          help='passphrase for the specified SSID')

    optional.add_argument('--virtual',
                          action="store_true",
                          help='specify this flag if the test should run on virtual clients')

    optional.add_argument('--num_sta',
                          type=int,
                          default=1,
                          help='specify the number of virtual stations to be created.')

    optional.add_argument('--radio',
                          type=str,
                          help='specify the radio to create the virtual stations')

    optional.add_argument('--real',
                          action="store_true",
                          help='specify this flag if the test should run on real clients')

    optional.add_argument('--use_default_config',
                          action='store_true',
                          help='specify this flag if wanted to proceed with existing Wi-Fi configuration of the devices')

    optional.add_argument('--debug',
                          action="store_true",
                          help='Enable debugging')

    # local report directory used by lf_report
    parser.add_argument('--local_lf_report_dir',
                        help='--local_lf_report_dir override the report path (lanforge/html-reports), primary used when making another directory lanforge/html-report/<test_rig>',
                        default="")

    # logging configuration:
    parser.add_argument('--log_level', default=None,
                        help='Set logging level: debug | info | warning | error | critical')

    parser.add_argument("--lf_logger_config_json",
                        help="--lf_logger_config_json <json file> , json configuration of logger")
    parser.add_argument('--help_summary', default=None, action="store_true", help='Show summary of what this script does')
    parser.add_argument('--group_name', type=str, help='Specify the groups name that contains a list of devices. Example: group1,group2')
    parser.add_argument('--profile_name', type=str, help='Specify the profile name to apply configurations to the devices.')
    parser.add_argument('--file_name', type=str, help='Specify the file name containing group details. Example:file1')
    parser.add_argument("--eap_method", type=str, default='DEFAULT', help="Specify the EAP method for authentication.")
    parser.add_argument("--eap_identity", type=str, default='', help="Specify the EAP identity for authentication.")
    parser.add_argument("--ieee8021x", action="store_true", help='Enables 802.1X enterprise authentication for test stations.')
    parser.add_argument("--ieee80211u", action="store_true", help='Enables IEEE 802.11u (Hotspot 2.0) support.')
    parser.add_argument("--ieee80211w", type=int, default=1, help='Enables IEEE 802.11w (Management Frame Protection) support.')
    parser.add_argument("--enable_pkc", action="store_true", help='Enables pkc support.')
    parser.add_argument("--bss_transition", action="store_true", help='Enables BSS transition support.')
    parser.add_argument("--power_save", action="store_true", help='Enables power-saving features.')
    parser.add_argument("--disable_ofdma", action="store_true", help='Disables OFDMA support.')
    parser.add_argument("--roam_ft_ds", action="store_true", help='Enables fast BSS transition (FT) support')
    parser.add_argument("--key_management", type=str, default='DEFAULT', help='Specify the key management method (e.g., WPA-PSK, WPA-EAP')
    parser.add_argument("--pairwise", type=str, default='[BLANK]')
    parser.add_argument("--private_key", type=str, default='[BLANK]', help='Specify EAP private key certificate file.')
    parser.add_argument("--ca_cert", type=str, default='[BLANK]', help='Specifiy the CA certificate file name')
    parser.add_argument("--client_cert", type=str, default='[BLANK]', help='Specify the client certificate file name')
    parser.add_argument("--pk_passwd", type=str, default='[BLANK]', help='Specify the password for the private key')
    parser.add_argument("--pac_file", type=str, default='[BLANK]', help='Specify the pac file name')
    parser.add_argument('--expected_passfail_value', help='Enter the expected packet loss', default=None)
    parser.add_argument('--device_csv_name', type=str, help='Enter the csv name to store expected values', default=None)
    parser.add_argument('--wait_time', type=int, help="Enter the maximum wait time for configurations to apply", default=60)

    args = parser.parse_args()

    if args.help_summary:
        print(help_summary)
        exit(0)

    # set the logger level to debug
    logger_config = lf_logger_config.lf_logger_config()

    if args.log_level:
        logger_config.set_level(level=args.log_level)

    if args.lf_logger_config_json:
        # logger_config.lf_logger_config_json = "lf_logger_config.json"
        logger_config.lf_logger_config_json = args.lf_logger_config_json
        logger_config.load_lf_logger_config()
    validate_args(args)

    mgr_ip = args.mgr
    mgr_password = args.mgr_passwd
    mgr_port = args.mgr_port
    server_ip = args.server_ip
    ssid = args.ssid
    security = args.security
    password = args.passwd
    num_sta = args.num_sta
    radio = args.radio
    target = args.target
    interval = args.ping_interval
    duration = args.ping_duration
    configure = not args.use_default_config
    debug = args.debug
    group_name = args.group_name
    file_name = args.file_name
    profile_name = args.profile_name
    eap_method = args.eap_method
    eap_identity = args.eap_identity
    ieee80211 = args.ieee8021x
    ieee80211u = args.ieee80211u
    ieee80211w = args.ieee80211w
    enable_pkc = args.enable_pkc
    bss_transition = args.bss_transition
    power_save = args.power_save
    disable_ofdma = args.disable_ofdma
    roam_ft_ds = args.roam_ft_ds
    key_management = args.key_management
    pairwise = args.pairwise
    private_key = args.private_key
    ca_cert = args.ca_cert
    client_cert = args.client_cert
    pk_passwd = args.pk_passwd
    pac_file = args.pac_file

    if (debug):
        print('''Specified configuration:
              ip:                       {}
              port:                     {}
              ssid:                     {}
              security:                 {}
              password:                 {}
              target:                   {}
              Ping interval:            {}
              Packet Duration (in min): {}
              virtual:                  {}
              num of virtual stations:  {}
              radio:                    {}
              real:                     {}
              debug:                    {}
              '''.format(mgr_ip, mgr_port, ssid, security, password, target, interval, duration, args.virtual, num_sta, radio, args.real, debug))

    # ping object creation
    ping = Ping(host=mgr_ip, port=mgr_port, ssid=ssid, security=security, password=password, radio=radio,
                lanforge_password=mgr_password, target=target, interval=interval, sta_list=[], virtual=args.virtual, real=args.real, duration=duration, debug=debug, csv_name=args.device_csv_name,
                expected_passfail_val=args.expected_passfail_value, wait_time=args.wait_time, group_name=group_name)

    # changing the target from port to IP
    ping.change_target_to_ip()

    # creating virtual stations if --virtual flag is specified
    if (args.virtual):

        logger.info('Proceeding to create {} virtual stations on {}'.format(num_sta, radio))
        station_list = LFUtils.portNameSeries(
            prefix_='sta', start_id_=0, end_id_=num_sta - 1, padding_number_=100000, radio=radio)
        ping.sta_list = station_list
        if (debug):
            logger.info('Virtual Stations: {}'.format(station_list).replace(
                '[', '').replace(']', '').replace('\'', ''))

    # selecting real clients if --real flag is specified
    if (args.real):
        Devices = RealDevice(manager_ip=mgr_ip, selected_bands=[])
        Devices.get_devices()
        ping.Devices = Devices
        # ping.select_real_devices(real_devices=Devices)
        # If config is True, attempt to bring up all devices in the list and perform tests on those that become active
        if (configure):
            config_devices = {}
            obj = DeviceConfig.DeviceConfig(lanforge_ip=mgr_ip, file_name=file_name, wait_time=args.wait_time)
            # Case 1: Group name, file name, and profile name are provided
            if group_name and file_name and profile_name:
                selected_groups = group_name.split(',')
                selected_profiles = profile_name.split(',')
                for i in range(len(selected_groups)):
                    config_devices[selected_groups[i]] = selected_profiles[i]
                obj.initiate_group()
                group_device_map = obj.get_groups_devices(data=selected_groups, groupdevmap=True)
                # Configure devices in the selected group with the selected profile
                eid_list = asyncio.run(obj.connectivity(config=config_devices, upstream=server_ip))
                Devices.get_devices()
                ping.select_real_devices(real_devices=Devices, device_list=eid_list)
            # Case 2: Device list is empty but config flag is True — prompt the user to input device details for configuration
            else:
                all_devices = obj.get_all_devices()
                device_list = []
                config_dict = {
                    'ssid': ssid,
                    'passwd': password,
                    'enc': security,
                    'eap_method': eap_method,
                    'eap_identity': eap_identity,
                    'ieee80211': ieee80211,
                    'ieee80211u': ieee80211u,
                    'ieee80211w': ieee80211w,
                    'enable_pkc': enable_pkc,
                    'bss_transition': bss_transition,
                    'power_save': power_save,
                    'disable_ofdma': disable_ofdma,
                    'roam_ft_ds': roam_ft_ds,
                    'key_management': key_management,
                    'pairwise': pairwise,
                    'private_key': private_key,
                    'ca_cert': ca_cert,
                    'client_cert': client_cert,
                    'pk_passwd': pk_passwd,
                    'pac_file': pac_file,
                    'server_ip': server_ip,
                }
                for device in all_devices:
                    if device["type"] == 'laptop':
                        device_list.append(device["shelf"] + '.' + device["resource"] + " " + device["hostname"])
                    else:
                        device_list.append(device["eid"] + " " + device["serial"])
                logger.info(f"Available devices: {device_list}")
                dev_list = input("Enter the desired resources to run the test:").split(',')
                dev_list = asyncio.run(obj.connectivity(device_list=dev_list, wifi_config=config_dict))
                Devices.get_devices()
                ping.select_real_devices(real_devices=Devices, device_list=dev_list)
        # Case 3: Config is False, no device list is provided, and no group is selected
        # Prompt the user to manually input devices for running the test
        else:
            device_list = ping.Devices.get_devices()
            logger.info(f"Available devices: {device_list}")
            dev_list = input("Enter the desired resources to run the test:").split(',')
            ping.select_real_devices(real_devices=Devices, device_list=dev_list)

    # station precleanup
    ping.cleanup()

    # building station if virtual
    if (args.virtual):
        ping.buildstation()

    # check if generic tab is enabled or not
    if (not ping.check_tab_exists()):
        raise RuntimeError('Generic Tab is not available.\nAborting the test.')

    ping.sta_list += ping.real_sta_list

    # creating generic endpoints
    ping.create_generic_endp()

    # run the test for the given duration
    logger.info('Running the ping test for {} minutes'.format(duration))

    # start generate endpoint
    ping.start_generic()
    start_time = time.time()
    end_time = start_time + (duration * 60)
    while time.time() < end_time:
        success = ping.monitor_endp_availability(ping.generic_endps_profile.created_endp)
        if not success:
            logger.error('All endpoints are not available. So exiting the monitor early.')
            break
        time.sleep(3)
    logger.info('Stopping the test')
    ping.stop_generic()

    result_data = ping.get_results()
    logger.info("Endpoint results after stopping the test: {}".format(result_data))
    if (args.virtual):
        ports_url = '/ports/all/'
        response = ping.json_get(ports_url)
        if not response:
            logger.error("Failed to fetch ports. Received empty response.\nRequested URL: '{}'\nResponse: {}".format(ports_url, response))
            raise RuntimeError('Failed to fetch port data from the LANforge.')
        if 'interfaces' not in response:
            logger.error("'interfaces' key not found in response.\nRequested URL: '{}'\nResponse: {}".format(ports_url, response))
            raise RuntimeError('Failed to fetch port data from the LANforge.')
        ports_data_dict = response['interfaces']
        ports_data = {}
        for ports in ports_data_dict:
            port, port_data = list(ports.keys())[0], list(ports.values())[0]
            ports_data[port] = port_data
        for station in ping.sta_list:
            if (station not in ping.real_sta_list):
                current_device_data = ports_data[station]
                for ping_device in result_data:
                    ping_endp, ping_data = list(ping_device.keys())[
                        0], list(ping_device.values())[0]
                    if (station.split('.')[2] in ping_endp):
                        last_result_lines = []
                        try:
                            last_result_lines = ping_data['last results'].split('\n') if ping_data['last results'] else []
                            if len(last_result_lines) > 1:
                                last_result = last_result_lines[-2]
                            elif last_result_lines:
                                last_result = last_result_lines[-1]
                            else:
                                last_result = ""

                            if 'min/avg/max' in last_result:
                                rtt_values = last_result.split('min/avg/max:', 1)[1].strip().split()[0].split('/')
                                min_rtt, avg_rtt, max_rtt = ping.validate_rtt(rtt_values[0], rtt_values[1], rtt_values[2])
                            else:
                                min_rtt, avg_rtt, max_rtt = 'NA', 'NA', 'NA'

                            ping.result_json[station] = {
                                'command': ping_data['command'],
                                'sent': ping_data['tx pkts'],
                                'recv': ping_data['rx pkts'],
                                'dropped': ping_data['dropped'],
                                'min_rtt': min_rtt,
                                'avg_rtt': avg_rtt,
                                'max_rtt': max_rtt,
                                'mac': current_device_data['mac'],
                                'ssid': current_device_data['ssid'],
                                'channel': current_device_data['channel'],
                                'mode': current_device_data['mode'],
                                'name': station,
                                'os': 'Virtual',
                                'remarks': [],
                                'last_result': last_result
                            }
                            ping.result_json[station]['remarks'] = ping.generate_remarks(ping.result_json[station])
                        except Exception as error:
                            logger.error(
                                "Failed parsing the result for station %s. Error: %s\nLast result lines: %s",
                                station, error, last_result_lines)

    if (args.real):
        for station in ping.real_sta_list:
            current_device_data = Devices.devices_data[station]
            for ping_device in result_data:
                ping_endp, ping_data = list(ping_device.keys())[
                    0], list(ping_device.values())[0]
                if (station in ping_endp):
                    last_result_lines = []
                    try:
                        last_result_lines = ping_data['last results'].split('\n') if ping_data['last results'] else []
                        if len(last_result_lines) > 1:
                            last_result = last_result_lines[-2]
                        elif last_result_lines:
                            last_result = last_result_lines[-1]
                        else:
                            last_result = ""

                        if 'min/avg/max' in last_result:
                            rtt_values = last_result.split('min/avg/max:', 1)[1].strip().split()[0].split('/')
                            min_rtt, avg_rtt, max_rtt = ping.validate_rtt(rtt_values[0], rtt_values[1], rtt_values[2])
                        else:
                            min_rtt, avg_rtt, max_rtt = 'NA', 'NA', 'NA'

                        ping.result_json[station] = {
                            'command': ping_data['command'],
                            'sent': ping_data['tx pkts'],
                            'recv': ping_data['rx pkts'],
                            'dropped': ping_data['dropped'],
                            'min_rtt': min_rtt,
                            'avg_rtt': avg_rtt,
                            'max_rtt': max_rtt,
                            'mac': current_device_data['mac'],
                            'ssid': current_device_data['ssid'],
                            'channel': current_device_data['channel'],
                            'mode': current_device_data['mode'],
                            'name': [current_device_data['user'] if current_device_data['user'] != '' else current_device_data['hostname']][0],
                            'os': ['Windows' if 'Win' in current_device_data['hw version'] else 'Linux' if 'Linux' in current_device_data['hw version'] else 'Mac' if 'Apple' in current_device_data['hw version'] else 'Android'][0],  # noqa E501
                            'remarks': [],
                            'last_result': last_result
                        }
                        ping.result_json[station]['remarks'] = ping.generate_remarks(ping.result_json[station])
                    except Exception as error:
                        logger.error(
                            "Failed parsing the result for station %s. Error: %s\nLast result lines: %s",
                            station, error, last_result_lines)

    logger.info("Final parsed per-station results: {}".format(ping.result_json))

    # station post cleanup
    ping.cleanup()

    if args.local_lf_report_dir == "":
        # Report generation when groups are specified but no custom report path is provided
        if args.group_name:
            ping.generate_report(config_devices=config_devices, group_device_map=group_device_map)
        # Report generation when no group is specified and no custom report path is provided
        else:
            ping.generate_report()
    else:
        # Report generation when groups are specified and a custom report path is provided
        if args.group_name:
            ping.generate_report(config_devices=config_devices, group_device_map=group_device_map, report_path=args.local_lf_report_dir)
        # Report generation when no group is specified but a custom report path is provided
        else:
            ping.generate_report(report_path=args.local_lf_report_dir)


if __name__ == "__main__":
    main()