diff --git a/keywords/linux/ip/ip_keywords.py b/keywords/linux/ip/ip_keywords.py index f0f1c393..c1df4910 100644 --- a/keywords/linux/ip/ip_keywords.py +++ b/keywords/linux/ip/ip_keywords.py @@ -116,3 +116,16 @@ class IPKeywords(BaseKeyword): ip_br_addr_output = IPBrAddrOutput(output) return ip_br_addr_output + + def set_ip_port_state(self, port: str, state: str = 'up'): + """ + sets a ip specific port state UP or DOWN as specified via ip link cmd + + Args: + port : port to set + state: state to set port (up or down) + + Returns: + """ + self.ssh_connection.send_as_sudo(f"ip link set dev {port} {state}") + self.validate_success_return_code(self.ssh_connection) diff --git a/keywords/ptp/phc_ctl_keywords.py b/keywords/ptp/phc_ctl_keywords.py new file mode 100644 index 00000000..7c0e4d98 --- /dev/null +++ b/keywords/ptp/phc_ctl_keywords.py @@ -0,0 +1,106 @@ +from keywords.base_keyword import BaseKeyword + + +class PhcCtlKeywords(BaseKeyword): + """ + Directly control PHC device clock using given SSH connection. + + Attributes: + ssh_connection: An instance of an SSH connection. + """ + + def __init__(self, ssh_connection): + """ + Initializes the PhcCtlKeywords with an SSH connection. + + Args: + ssh_connection: An instance of an SSH connection. + """ + self.ssh_connection = ssh_connection + + def phc_ctl_get(self, device: str) -> str: + """ + Get the current time of the PHC clock device + + Args: + device : may be either CLOCK_REALTIME, any /dev/ptpX device, or any ethernet device which supports ethtool's get_ts_info ioctl. + + Example: + phc_ctl[643764.828]: clock time is 1739856255.215802036 or Tue Feb 18 05:24:15 2025 + + Returns: + """ + output = self.ssh_connection.send_as_sudo(f"phc_ctl {device} get") + self.validate_success_return_code(self.ssh_connection) + output_str = ''.join(output).replace('\n', '') + if output_str and len(output_str.split()) > 4: + return output_str.split()[4] + else: + raise "output_str.split() is expected to be a List with four elements." + + def phc_ctl_cmp(self, device: str) -> str: + """ + Compare the PHC clock device to CLOCK_REALTIME + + Args: + device : may be either CLOCK_REALTIME, any /dev/ptpX device, or any ethernet device which supports ethtool's get_ts_info ioctl. + + Example: + phc_ctl[645639.878]: offset from CLOCK_REALTIME is -37000000008ns + + Returns: + """ + output = self.ssh_connection.send_as_sudo(f"phc_ctl {device} cmp") + self.validate_success_return_code(self.ssh_connection) + output_str = ''.join(output) + if output_str and len(output_str.split()) > 5: + return output_str.split()[5] + else: + raise "output_str.split() is expected to be a List with five elements." + + def phc_ctl_adj(self, device: str, seconds: str) -> str: + """ + Adjust the PHC clock by an amount of seconds provided + + Args: + device : may be either CLOCK_REALTIME, any /dev/ptpX device, or any ethernet device which supports ethtool's get_ts_info ioctl. + seconds : + + Example: + phc_ctl[646368.470]: adjusted clock by 0.000001 seconds + + Returns: + """ + output = self.ssh_connection.send_as_sudo(f"phc_ctl {device} adj {seconds}") + self.validate_success_return_code(self.ssh_connection) + output_str = ''.join(output).replace('\n', '') + if output_str and len(output_str.split()) > 5: + return output_str.split()[4] + else: + raise "output_str.split() is expected to be a List with five elements." + + def phc_ctl_set(self, device: str, seconds: str = None) -> str: + """ + Set the PHC clock time to the value specified in seconds. Defaults to reading CLOCK_REALTIME if no value is provided. + + Args: + device : may be either CLOCK_REALTIME, any /dev/ptpX device, or any ethernet device which supports ethtool's get_ts_info ioctl. + seconds : + + Example : + phc_ctl[647759.416]: set clock time to 1739860212.789318498 or Tue Feb 18 06:30:12 2025 + + Returns: + """ + if seconds: + cmd = f"phc_ctl {device} set {seconds}" + else: + cmd = f"phc_ctl {device} set" + + output = self.ssh_connection.send_as_sudo(cmd) + self.validate_success_return_code(self.ssh_connection) + output_str = ''.join(output).replace('\n', '') + if output_str and len(output_str.split()) > 5: + return output_str.split()[5] + else: + raise "output_str.split() is expected to be a List with five elements." diff --git a/keywords/ptp/pmc/objects/pmc_get_current_data_set_object.py b/keywords/ptp/pmc/objects/pmc_get_current_data_set_object.py new file mode 100644 index 00000000..690ca391 --- /dev/null +++ b/keywords/ptp/pmc/objects/pmc_get_current_data_set_object.py @@ -0,0 +1,66 @@ +class PMCGetCurrentDataSetObject: + """ + Object to hold the values of Current data set + """ + + def __init__(self): + self.steps_removed: int = -1 + self.offset_from_master: float = 0.0 + self.mean_path_delay: float = 0.0 + + def get_steps_removed(self) -> int: + """ + Getter for steps_removed + Returns: the steps_removed value + + """ + return self.steps_removed + + def set_steps_removed(self, steps_removed: int): + """ + Setter for steps_removed + Args: + steps_removed (): the steps_removed value + + Returns: + + """ + self.steps_removed = steps_removed + + def get_offset_from_master(self) -> float: + """ + Getter for offset_from_master + Returns: offset_from_master + + """ + return self.offset_from_master + + def set_offset_from_master(self, offset_from_master: float): + """ + Setter for offset_from_master + Args: + offset_from_master (): the offset_from_master value + + Returns: + + """ + self.offset_from_master = offset_from_master + + def get_mean_path_delay(self) -> float: + """ + Getter for mean_path_delay + Returns: mean_path_delay value + + """ + return self.mean_path_delay + + def set_mean_path_delay(self, mean_path_delay: float): + """ + Setter for mean_path_delay + Args: + mean_path_delay (): mean_path_delay value + + Returns: + + """ + self.mean_path_delay = mean_path_delay diff --git a/keywords/ptp/pmc/objects/pmc_get_current_data_set_output.py b/keywords/ptp/pmc/objects/pmc_get_current_data_set_output.py new file mode 100644 index 00000000..5c7549ea --- /dev/null +++ b/keywords/ptp/pmc/objects/pmc_get_current_data_set_output.py @@ -0,0 +1,46 @@ +from keywords.ptp.pmc.objects.pmc_get_current_data_set_object import PMCGetCurrentDataSetObject +from keywords.ptp.pmc.pmc_table_parser import PMCTableParser + + +class PMCGetCurrentDataSetOutput: + """ + This class parses the output of commands such as 'pmc CURRENT_DATA_SET' + + Example: + sending: GET CURRENT_DATA_SET + 507c6f.fffe.0b5a4d-0 seq 0 RESPONSE MANAGEMENT CURRENT_DATA_SET + stepsRemoved 0 + offsetFromMaster 0.0 + meanPathDelay 0.0 + """ + + def __init__(self, pmc_output: [str]): + """ + Constructor. + Create an internal PMCGetCurrentDataSet from the passed parameter. + Args: + pmc_output (list[str]): a list of strings representing the output of the pmc command + + """ + pmc_table_parser = PMCTableParser(pmc_output) + output_values = pmc_table_parser.get_output_values_dict() + self.pmc_get_current_data_set_object = PMCGetCurrentDataSetObject() + + if 'stepsRemoved' in output_values: + self.pmc_get_current_data_set_object.set_steps_removed(int(output_values['stepsRemoved'])) + + if 'offsetFromMaster' in output_values: + self.pmc_get_current_data_set_object.set_offset_from_master(float(output_values['offsetFromMaster'])) + + if 'meanPathDelay' in output_values: + self.pmc_get_current_data_set_object.set_mean_path_delay(float(output_values['meanPathDelay'])) + + def get_pmc_get_current_data_set_object(self) -> PMCGetCurrentDataSetObject: + """ + Getter for pmc_get_current_data_set_object object. + + Returns: + A PMCGetCurrentDataSetObject + + """ + return self.pmc_get_current_data_set_object diff --git a/keywords/ptp/pmc/objects/pmc_get_domain_object.py b/keywords/ptp/pmc/objects/pmc_get_domain_object.py new file mode 100644 index 00000000..7d9bce9c --- /dev/null +++ b/keywords/ptp/pmc/objects/pmc_get_domain_object.py @@ -0,0 +1,29 @@ +class PMCGetDomainObject: + """ + Object to hold the values of DOMAIN + """ + + def __init__(self): + self.domain_number: int = -1 + + def get_domain_number(self) -> int: + """ + Getter for domain_number + Returns: the domain_number value + + """ + return self.domain_number + + def set_domain_number(self, domain_number: int): + """ + Setter for domain_number + Args: + domain_number : the domain_number value + + Returns: + + """ + self.domain_number = domain_number + + + diff --git a/keywords/ptp/pmc/objects/pmc_get_domain_output.py b/keywords/ptp/pmc/objects/pmc_get_domain_output.py new file mode 100644 index 00000000..975b52f2 --- /dev/null +++ b/keywords/ptp/pmc/objects/pmc_get_domain_output.py @@ -0,0 +1,39 @@ +from keywords.ptp.pmc.objects.pmc_get_domain_object import PMCGetDomainObject +from keywords.ptp.pmc.pmc_table_parser import PMCTableParser + + +class PMCGetDomainOutput: + """ + This class parses the output of commands such as 'pmc GET DOMAIN' + + Example: + sending: GET DOMAIN + 507c6f.fffe.0b5a4d-0 seq 0 RESPONSE MANAGEMENT DOMAIN + domainNumber 24 + """ + + def __init__(self, pmc_output: [str]): + """ + Constructor. + Create an internal DOMAIN from the passed parameter. + Args: + pmc_output (list[str]): a list of strings representing the output of the pmc command + + """ + pmc_table_parser = PMCTableParser(pmc_output) + output_values = pmc_table_parser.get_output_values_dict() + self.pmc_get_domain_object = PMCGetDomainObject() + + if 'domainNumber' in output_values: + self.pmc_get_domain_object.set_domain_number(int(output_values['domainNumber'])) + + + def get_pmc_get_domain_object(self) -> PMCGetDomainObject: + """ + Getter for pmc_get_domain_object object. + + Returns: + A PMCGetDomainObject + + """ + return self.pmc_get_domain_object diff --git a/keywords/ptp/pmc/objects/pmc_get_grandmaster_settings_np_object.py b/keywords/ptp/pmc/objects/pmc_get_grandmaster_settings_np_object.py new file mode 100644 index 00000000..9e984c1c --- /dev/null +++ b/keywords/ptp/pmc/objects/pmc_get_grandmaster_settings_np_object.py @@ -0,0 +1,229 @@ +class PMCGetGrandmasterSettingsNpObject: + """ + Object to hold the values of GRANDMASTER_SETTINGS_NP + """ + + def __init__(self): + self.clock_class: int = -1 + self.clock_accuracy: str = '' + self.offset_scaled_log_variance: str = '' + self.current_utc_offset: int = -1 + self.leap61: int = -1 + self.leap59: int = -1 + self.current_utc_off_set_valid: int = -1 + self.ptp_time_scale: int = -1 + self.time_traceable: int = -1 + self.frequency_traceable: int = -1 + self.time_source: str = '' + + def get_clock_class(self) -> int: + """ + Getter for clock_class + Returns: the clock_class value + + """ + return self.clock_class + + def set_clock_class(self, clock_class: int): + """ + Setter for clock_class + Args: + clock_class : the clock_class value + + Returns: + + """ + self.clock_class = clock_class + + def get_clock_accuracy(self) -> str: + """ + Getter for clock_accuracy + Returns: the clock_accuracy value + + """ + return self.clock_accuracy + + def set_clock_accuracy(self, clock_accuracy: str): + """ + Setter for clock_accuracy + Args: + clock_accuracy : the clock_accuracy value + + Returns: + + """ + self.clock_accuracy = clock_accuracy + + def get_offset_scaled_log_variance(self) -> str: + """ + Getter for offset_scaled_log_variance + Returns: the offset_scaled_log_variance value + + """ + return self.offset_scaled_log_variance + + def set_offset_scaled_log_variance(self, offset_scaled_log_variance: str): + """ + Setter for offset_scaled_log_variance + Args: + offset_scaled_log_variance : the offset_scaled_log_variance value + + Returns: + + """ + self.offset_scaled_log_variance = offset_scaled_log_variance + + def get_current_utc_offset(self) -> int: + """ + Getter for current_utc_offset + Returns: the current_utc_offset value + + """ + return self.current_utc_offset + + def set_current_utc_offset(self, current_utc_offset: int): + """ + Setter for current_utc_offset + Args: + current_utc_offset : the current_utc_offset value + + Returns: + + """ + self.current_utc_offset = current_utc_offset + + def get_leap61(self) -> int: + """ + Getter for leap61 + Returns: the leap61 value + + """ + return self.leap61 + + def set_leap61(self, leap61: int): + """ + Setter for leap61 + Args: + leap61 : the leap61 value + + Returns: + + """ + self.leap61 = leap61 + + def get_leap59(self) -> int: + """ + Getter for leap59 + Returns: the leap59 value + + """ + return self.leap59 + + def set_leap59(self, leap59: int): + """ + Setter for leap59 + Args: + leap59 : the leap59 value + + Returns: + + """ + self.leap59 = leap59 + + def get_current_utc_off_set_valid(self) -> int: + """ + Getter for current_utc_off_set_valid + Returns: the current_utc_off_set_valid value + + """ + return self.current_utc_off_set_valid + + def set_current_utc_off_set_valid(self, current_utc_off_set_valid: int): + """ + Setter for current_utc_off_set_valid + Args: + current_utc_off_set_valid : the current_utc_off_set_valid value + + Returns: + + """ + self.current_utc_off_set_valid = current_utc_off_set_valid + + def get_ptp_time_scale(self) -> int: + """ + Getter for ptp_time_scale + Returns: the ptp_time_scale value + + """ + return self.ptp_time_scale + + def set_ptp_time_scale(self, ptp_time_scale: int): + """ + Setter for ptp_time_scale + Args: + ptp_time_scale : the ptp_time_scale value + + Returns: + + """ + self.ptp_time_scale = ptp_time_scale + + def get_time_traceable(self) -> int: + """ + Getter for time_traceable + Returns: the time_traceable value + + """ + return self.time_traceable + + def set_time_traceable(self, time_traceable: int): + """ + Setter for time_traceable + Args: + time_traceable : the time_traceable value + + Returns: + + """ + self.time_traceable = time_traceable + + def get_frequency_traceable(self) -> int: + """ + Getter for frequency_traceable + Returns: the frequency_traceable value + + """ + return self.frequency_traceable + + def set_frequency_traceable(self, frequency_traceable: int): + """ + Setter for frequency_traceable + Args: + frequency_traceable : the frequency_traceable value + + Returns: + + """ + self.frequency_traceable = frequency_traceable + + def get_time_source(self) -> str: + """ + Getter for time_source + Returns: the time_source value + + """ + return self.time_source + + def set_time_source(self, time_source: str): + """ + Setter for time_source + Args: + time_source : the time_source value + + Returns: + + """ + self.time_source = time_source + + + diff --git a/keywords/ptp/pmc/objects/pmc_get_grandmaster_settings_np_output.py b/keywords/ptp/pmc/objects/pmc_get_grandmaster_settings_np_output.py new file mode 100644 index 00000000..0512adde --- /dev/null +++ b/keywords/ptp/pmc/objects/pmc_get_grandmaster_settings_np_output.py @@ -0,0 +1,79 @@ +from keywords.ptp.pmc.objects.pmc_get_grandmaster_settings_np_object import PMCGetGrandmasterSettingsNpObject +from keywords.ptp.pmc.pmc_table_parser import PMCTableParser + + +class PMCGetGrandmasterSettingsNpOutput: + """ + This class parses the output of commands such as 'pmc GRANDMASTER_SETTINGS_NP' + + Example: + sending: GET GRANDMASTER_SETTINGS_NP + 507c6f.fffe.0b5a4d-0 seq 0 RESPONSE MANAGEMENT GRANDMASTER_SETTINGS_NP + clockClass 6 + clockAccuracy 0x20 + offsetScaledLogVariance 0x4e5d + currentUtcOffset 37 + leap61 0 + leap59 0 + currentUtcOffsetValid 0 + ptpTimescale 1 + timeTraceable 1 + frequencyTraceable 1 + timeSource 0x20 + """ + + def __init__(self, pmc_output: [str]): + """ + Constructor. + Create an internal GRANDMASTER_SETTINGS_NP from the passed parameter. + Args: + pmc_output (list[str]): a list of strings representing the output of the pmc command + + """ + pmc_table_parser = PMCTableParser(pmc_output) + output_values = pmc_table_parser.get_output_values_dict() + self.pmc_get_grandmaster_settings_np_object = PMCGetGrandmasterSettingsNpObject() + + if 'clockClass' in output_values: + self.pmc_get_grandmaster_settings_np_object.set_clock_class(int(output_values['clockClass'])) + + if 'clockAccuracy' in output_values: + self.pmc_get_grandmaster_settings_np_object.set_clock_accuracy(output_values['clockAccuracy']) + + if 'offsetScaledLogVariance' in output_values: + self.pmc_get_grandmaster_settings_np_object.set_offset_scaled_log_variance(output_values['offsetScaledLogVariance']) + + if 'currentUtcOffset' in output_values: + self.pmc_get_grandmaster_settings_np_object.set_current_utc_offset(int(output_values['currentUtcOffset'])) + + if 'leap61' in output_values: + self.pmc_get_grandmaster_settings_np_object.set_leap61(int(output_values['leap61'])) + + if 'leap59' in output_values: + self.pmc_get_grandmaster_settings_np_object.set_leap59(int(output_values['leap59'])) + + if 'currentUtcOffsetValid' in output_values: + self.pmc_get_grandmaster_settings_np_object.set_current_utc_off_set_valid(int(output_values['currentUtcOffsetValid'])) + + if 'ptpTimescale' in output_values: + self.pmc_get_grandmaster_settings_np_object.set_ptp_time_scale(int(output_values['ptpTimescale'])) + + if 'timeTraceable' in output_values: + self.pmc_get_grandmaster_settings_np_object.set_time_traceable(int(output_values['timeTraceable'])) + + if 'frequencyTraceable' in output_values: + self.pmc_get_grandmaster_settings_np_object.set_frequency_traceable(int(output_values['frequencyTraceable'])) + + if 'timeSource' in output_values: + self.pmc_get_grandmaster_settings_np_object.set_time_source(output_values['timeSource']) + + + def get_pmc_get_grandmaster_settings_np_object(self) -> PMCGetGrandmasterSettingsNpObject: + """ + Getter for pmc_get_grandmaster_settings_np_object object. + + Returns: + A PMCGetGrandmasterSettingsNpObject + + """ + return self.pmc_get_grandmaster_settings_np_object diff --git a/keywords/ptp/pmc/objects/pmc_get_time_properties_data_set_object.py b/keywords/ptp/pmc/objects/pmc_get_time_properties_data_set_object.py new file mode 100644 index 00000000..fa6e7443 --- /dev/null +++ b/keywords/ptp/pmc/objects/pmc_get_time_properties_data_set_object.py @@ -0,0 +1,169 @@ +class PMCGetTimePropertiesDataSetObject: + """ + Object to hold the values of TIME_PROPERTIES_DATA_SET + """ + + def __init__(self): + self.current_utc_offset: int = -1 + self.leap61: int = -1 + self.leap59: int = -1 + self.current_utc_off_set_valid: int = -1 + self.ptp_time_scale: int = -1 + self.time_traceable: int = -1 + self.frequency_traceable: int = -1 + self.time_source: str = '' + + def get_current_utc_offset(self) -> int: + """ + Getter for current_utc_offset + Returns: the current_utc_offset value + + """ + return self.current_utc_offset + + def set_current_utc_offset(self, current_utc_offset: int): + """ + Setter for current_utc_offset + Args: + current_utc_offset : the current_utc_offset value + + Returns: + + """ + self.current_utc_offset = current_utc_offset + + def get_leap61(self) -> int: + """ + Getter for leap61 + Returns: the leap61 value + + """ + return self.leap61 + + def set_leap61(self, leap61: int): + """ + Setter for leap61 + Args: + leap61 : the leap61 value + + Returns: + + """ + self.leap61 = leap61 + + def get_leap59(self) -> int: + """ + Getter for leap59 + Returns: the leap59 value + + """ + return self.leap59 + + def set_leap59(self, leap59: int): + """ + Setter for leap59 + Args: + leap59 : the leap59 value + + Returns: + + """ + self.leap59 = leap59 + + def get_current_utc_off_set_valid(self) -> int: + """ + Getter for current_utc_off_set_valid + Returns: the current_utc_off_set_valid value + + """ + return self.current_utc_off_set_valid + + def set_current_utc_off_set_valid(self, current_utc_off_set_valid: int): + """ + Setter for current_utc_off_set_valid + Args: + current_utc_off_set_valid : the current_utc_off_set_valid value + + Returns: + + """ + self.current_utc_off_set_valid = current_utc_off_set_valid + + def get_ptp_time_scale(self) -> int: + """ + Getter for ptp_time_scale + Returns: the ptp_time_scale value + + """ + return self.ptp_time_scale + + def set_ptp_time_scale(self, ptp_time_scale: int): + """ + Setter for ptp_time_scale + Args: + ptp_time_scale : the ptp_time_scale value + + Returns: + + """ + self.ptp_time_scale = ptp_time_scale + + def get_time_traceable(self) -> int: + """ + Getter for time_traceable + Returns: the time_traceable value + + """ + return self.time_traceable + + def set_time_traceable(self, time_traceable: int): + """ + Setter for time_traceable + Args: + time_traceable : the time_traceable value + + Returns: + + """ + self.time_traceable = time_traceable + + def get_frequency_traceable(self) -> int: + """ + Getter for frequency_traceable + Returns: the frequency_traceable value + + """ + return self.frequency_traceable + + def set_frequency_traceable(self, frequency_traceable: int): + """ + Setter for frequency_traceable + Args: + frequency_traceable : the frequency_traceable value + + Returns: + + """ + self.frequency_traceable = frequency_traceable + + def get_time_source(self) -> str: + """ + Getter for time_source + Returns: the time_source value + + """ + return self.time_source + + def set_time_source(self, time_source: str): + """ + Setter for time_source + Args: + time_source : the time_source value + + Returns: + + """ + self.time_source = time_source + + + diff --git a/keywords/ptp/pmc/objects/pmc_get_time_properties_data_set_output.py b/keywords/ptp/pmc/objects/pmc_get_time_properties_data_set_output.py new file mode 100644 index 00000000..96b26ac3 --- /dev/null +++ b/keywords/ptp/pmc/objects/pmc_get_time_properties_data_set_output.py @@ -0,0 +1,67 @@ +from keywords.ptp.pmc.objects.pmc_get_time_properties_data_set_object import PMCGetTimePropertiesDataSetObject +from keywords.ptp.pmc.pmc_table_parser import PMCTableParser + + +class PMCGetTimePropertiesDataSetOutput: + """ + This class parses the output of commands such as 'pmc TIME_PROPERTIES_DATA_SET' + + Example: + sending: GET TIME_PROPERTIES_DATA_SET + 507c6f.fffe.0b5a4d-0 seq 0 RESPONSE MANAGEMENT TIME_PROPERTIES_DATA_SET + currentUtcOffset 37 + leap61 0 + leap59 0 + currentUtcOffsetValid 0 + ptpTimescale 1 + timeTraceable 1 + frequencyTraceable 0 + timeSource 0x20 + """ + + def __init__(self, pmc_output: [str]): + """ + Constructor. + Create an internal TIME_PROPERTIES_DATA_SET from the passed parameter. + Args: + pmc_output (list[str]): a list of strings representing the output of the pmc command + + """ + pmc_table_parser = PMCTableParser(pmc_output) + output_values = pmc_table_parser.get_output_values_dict() + self.pmc_get_time_properties_data_set_object = PMCGetTimePropertiesDataSetObject() + + if 'currentUtcOffset' in output_values: + self.pmc_get_time_properties_data_set_object.set_current_utc_offset(int(output_values['currentUtcOffset'])) + + if 'leap61' in output_values: + self.pmc_get_time_properties_data_set_object.set_leap61(int(output_values['leap61'])) + + if 'leap59' in output_values: + self.pmc_get_time_properties_data_set_object.set_leap59(int(output_values['leap59'])) + + if 'currentUtcOffsetValid' in output_values: + self.pmc_get_time_properties_data_set_object.set_current_utc_off_set_valid(int(output_values['currentUtcOffsetValid'])) + + if 'ptpTimescale' in output_values: + self.pmc_get_time_properties_data_set_object.set_ptp_time_scale(int(output_values['ptpTimescale'])) + + if 'timeTraceable' in output_values: + self.pmc_get_time_properties_data_set_object.set_time_traceable(int(output_values['timeTraceable'])) + + if 'frequencyTraceable' in output_values: + self.pmc_get_time_properties_data_set_object.set_frequency_traceable(int(output_values['frequencyTraceable'])) + + if 'timeSource' in output_values: + self.pmc_get_time_properties_data_set_object.set_time_source(output_values['timeSource']) + + + def get_pmc_get_time_properties_data_set_object(self) -> PMCGetTimePropertiesDataSetObject: + """ + Getter for pmc_get_time_properties_data_set_object object. + + Returns: + A PMCGetTimePropertiesDataSetObject + + """ + return self.pmc_get_time_properties_data_set_object diff --git a/keywords/ptp/pmc/objects/pmc_get_time_status_np_object.py b/keywords/ptp/pmc/objects/pmc_get_time_status_np_object.py new file mode 100644 index 00000000..b3070c28 --- /dev/null +++ b/keywords/ptp/pmc/objects/pmc_get_time_status_np_object.py @@ -0,0 +1,134 @@ +class PMCGetTimeStatusNpObject: + """ + Object to hold the values of TIME_STATUS_NP + """ + + def __init__(self): + self.master_offset = -1 + self.ingress_time = -1 + self.cumulative_scaled_rate_offset = '' + self.scaled_last_gm_phase_change = -1 + self.gm_time_base_indicator = -1 + self.last_gm_phase_change = '' + self.gm_present = '' + self.gm_identity = '' + + def get_master_offset(self) -> int: + """Getter for master offset + + Returns: the master offset + """ + return self.master_offset + + def set_master_offset(self, master_offset : int): + """Setter for master offset + + Args: + master_offset : the master offset value + """ + self.master_offset = master_offset + + def get_ingress_time(self) -> int: + """Getter for ingress time + + Returns: the ingress time + """ + return self.ingress_time + + def set_ingress_time(self, ingress_time : int): + """Setter for ingress time + + Args: + ingress_time : the ingress time value + """ + self.ingress_time = ingress_time + + def get_cumulative_scaled_rate_offset(self) -> str: + """Getter for cumulative scaled rate offset + + Returns: the cumulative scaled rate offset + """ + return self.cumulative_scaled_rate_offset + + def set_cumulative_scaled_rate_offset(self, cumulative_scaled_rate_offset : str): + """Setter for cumulative scaled rate offset + + Args: + cumulative_scaled_rate_offset : the cumulative scaled rate offset value + """ + self.cumulative_scaled_rate_offset = cumulative_scaled_rate_offset + + def get_scaled_last_gm_phase_change(self) -> int: + """Getter for scaled last GM phase change + + Returns: the scaled last GM phase change + """ + return self.scaled_last_gm_phase_change + + def set_scaled_last_gm_phase_change(self, scaled_last_gm_phase_change : int): + """Setter for scaled last GM phase change + + Args: + scaled_last_gm_phase_change : the scaled last GM phase change value + """ + self.scaled_last_gm_phase_change = scaled_last_gm_phase_change + + def get_gm_time_base_indicator(self) -> int: + """Getter for GM time base indicator + + Returns: the GM time base indicator value + """ + return self.gm_time_base_indicator + + def set_gm_time_base_indicator(self, gm_time_base_indicator : int): + """Setter for GM time base indicator + + Args: + gm_time_base_indicator : the GM time base indicator value + """ + self.gm_time_base_indicator = gm_time_base_indicator + + def get_last_gm_phase_change(self) -> str: + """Getter for last GM phase change + + Returns: the last GM phase change value + """ + return self.last_gm_phase_change + + def set_last_gm_phase_change(self, last_gm_phase_change : str): + """Setter for last GM phase change + + Args: + last_gm_phase_change : the last GM phase change value + """ + self.last_gm_phase_change = last_gm_phase_change + + def get_gm_present(self) -> str: + """Getter for GM presence + + Returns: the GM present + """ + return self.gm_present + + def set_gm_present(self, gm_present : str): + """Setter for GM presence. + + Args: + gm_present: the GM present value + """ + self.gm_present = gm_present + + def get_gm_identity(self) -> str: + """Getter for GM identity + + Returns: the GM identity + """ + return self.gm_identity + + def set_gm_identity(self, gm_identity : str): + """Setter for GM identity. + + Args: + gm_identity: the GM identity value + """ + self.gm_identity = gm_identity \ No newline at end of file diff --git a/keywords/ptp/pmc/objects/pmc_get_time_status_np_output.py b/keywords/ptp/pmc/objects/pmc_get_time_status_np_output.py new file mode 100644 index 00000000..c2d88c06 --- /dev/null +++ b/keywords/ptp/pmc/objects/pmc_get_time_status_np_output.py @@ -0,0 +1,66 @@ +from keywords.ptp.pmc.objects.pmc_get_time_status_np_object import PMCGetTimeStatusNpObject +from keywords.ptp.pmc.pmc_table_parser import PMCTableParser + +class PMCGetTimeStatusNpOutput: + """ + This class parses the output of commands such as 'pmc TIME_STATUS_NP' + + Example: + sending: GET TIME_STATUS_NP + 507c6f.fffe.0b5a4d-0 seq 0 RESPONSE MANAGEMENT TIME_STATUS_NP + master_offset -62 + ingress_time 0 + cumulativeScaledRateOffset +0.000000000 + scaledLastGmPhaseChange 0 + gmTimeBaseIndicator 0 + lastGmPhaseChange 0x0000'0000000000000000.0000 + gmPresent false + gmIdentity 507c6f.fffe.0b5a4d + """ + + def __init__(self, pmc_output: [str]): + """ + Constructor. + Create an internal TIME_STATUS_NP from the passed parameter. + Args: + pmc_output (list[str]): a list of strings representing the output of the pmc command + + """ + pmc_table_parser = PMCTableParser(pmc_output) + output_values = pmc_table_parser.get_output_values_dict() + self.pmc_get_time_status_np_object = PMCGetTimeStatusNpObject() + + if 'master_offset' in output_values: + self.pmc_get_time_status_np_object.set_master_offset(int(output_values['master_offset'])) + + if 'ingress_time' in output_values: + self.pmc_get_time_status_np_object.set_ingress_time(int(output_values['ingress_time'])) + + if 'cumulativeScaledRateOffset' in output_values: + self.pmc_get_time_status_np_object.set_cumulative_scaled_rate_offset(output_values['cumulativeScaledRateOffset']) + + if 'scaledLastGmPhaseChange' in output_values: + self.pmc_get_time_status_np_object.set_scaled_last_gm_phase_change(int(output_values['scaledLastGmPhaseChange'])) + + if 'gmTimeBaseIndicator' in output_values: + self.pmc_get_time_status_np_object.set_gm_time_base_indicator(int(output_values['gmTimeBaseIndicator'])) + + if 'lastGmPhaseChange' in output_values: + self.pmc_get_time_status_np_object.set_last_gm_phase_change(output_values['lastGmPhaseChange']) + + if 'gmPresent' in output_values: + self.pmc_get_time_status_np_object.set_gm_present(output_values['gmPresent']) + + if 'gmIdentity' in output_values: + self.pmc_get_time_status_np_object.set_gm_identity(output_values['gmIdentity']) + + + def get_pmc_get_time_status_np_object(self) -> PMCGetTimeStatusNpObject: + """ + Getter for pmc_get_time_status_np_object object. + + Returns: + A PMCGetTimeStatusNpObject + + """ + return self.pmc_get_time_status_np_object diff --git a/keywords/ptp/pmc/pmc_keywords.py b/keywords/ptp/pmc/pmc_keywords.py index 7328ac0f..4811f383 100644 --- a/keywords/ptp/pmc/pmc_keywords.py +++ b/keywords/ptp/pmc/pmc_keywords.py @@ -1,7 +1,12 @@ from framework.ssh.ssh_connection import SSHConnection from keywords.base_keyword import BaseKeyword +from keywords.ptp.pmc.objects.pmc_get_current_data_set_output import PMCGetCurrentDataSetOutput from keywords.ptp.pmc.objects.pmc_get_default_data_set_output import PMCGetDefaultDataSetOutput +from keywords.ptp.pmc.objects.pmc_get_domain_output import PMCGetDomainOutput +from keywords.ptp.pmc.objects.pmc_get_grandmaster_settings_np_output import PMCGetGrandmasterSettingsNpOutput from keywords.ptp.pmc.objects.pmc_get_parent_data_set_output import PMCGetParentDataSetOutput +from keywords.ptp.pmc.objects.pmc_get_time_properties_data_set_output import PMCGetTimePropertiesDataSetOutput +from keywords.ptp.pmc.objects.pmc_get_time_status_np_output import PMCGetTimeStatusNpOutput class PMCKeywords(BaseKeyword): @@ -12,6 +17,99 @@ class PMCKeywords(BaseKeyword): def __init__(self, ssh_connection: SSHConnection): self.ssh_connection = ssh_connection + def pmc_get_time_status_np(self, config_file: str, socket_file: str, unicast: bool = True, boundry_clock: int = 0) -> PMCGetTimeStatusNpOutput: + """ + Gets the TIME_STATUS_NP + Args: + config_file (): the config file + socket_file (): the socket file + unicast (): true to use unicast + boundry_clock (): the boundry clock + + Example: PMCKeywords(ssh_connection).pmc_get_time_status_np('/etc/linuxptp/ptpinstance/ptp4l-ptp5.conf', ' /var/run/ptp4l-ptp5') + + """ + cmd = f"pmc {'-u' if unicast else ''} -b {boundry_clock} -f {config_file} -s {socket_file} 'GET TIME_STATUS_NP'" + + output = self.ssh_connection.send_as_sudo(cmd) + pmc_get_time_status_np_output = PMCGetTimeStatusNpOutput(output) + return pmc_get_time_status_np_output + + def pmc_get_current_data_set(self, config_file: str, socket_file: str, unicast: bool = True, boundry_clock: int = 0) -> PMCGetCurrentDataSetOutput: + """ + Gets the CURRENT_DATA_SET + Args: + config_file (): the config file + socket_file (): the socket file + unicast (): true to use unicast + boundry_clock (): the boundry clock + + Example: PMCKeywords(ssh_connection).pmc_get_current_data_set('/etc/linuxptp/ptpinstance/ptp4l-ptp5.conf', ' /var/run/ptp4l-ptp5') + + """ + cmd = f"pmc {'-u' if unicast else ''} -b {boundry_clock} -f {config_file} -s {socket_file} 'GET CURRENT_DATA_SET'" + + output = self.ssh_connection.send_as_sudo(cmd) + pmc_get_current_data_set_output = PMCGetCurrentDataSetOutput(output) + return pmc_get_current_data_set_output + + def pmc_set_grandmaster_settings_np(self, config_file: str, socket_file: str, clock_class: int, time_traceable: int) -> PMCGetGrandmasterSettingsNpOutput: + """ + SET GRANDMASTER_SETTINGS_NP + + Args: + config_file : the config file + socket_file : the socket file + clock_class : clockClass of ptp + time_traceable : timeTraceable of ptp + + Example: PMCKeywords(ssh_connection).pmc_set_grandmaster_settings_np('/etc/linuxptp/ptpinstance/ptp4l-ptp5.conf', '/var/run/ptp4l-ptp5', 7, 1) + + """ + cmd = f"pmc -u -b 0 -f {config_file} -s {socket_file} 'SET GRANDMASTER_SETTINGS_NP clockClass {clock_class} clockAccuracy 0xfe offsetScaledLogVariance 0xffff currentUtcOffset 37 leap61 0 leap59 0 currentUtcOffsetValid 0 ptpTimescale 1 timeTraceable {time_traceable} frequencyTraceable 0 timeSource 0xa0'" + + output = self.ssh_connection.send_as_sudo(cmd) + pmc_set_grandmaster_settings_np_output = PMCGetGrandmasterSettingsNpOutput(output) + return pmc_set_grandmaster_settings_np_output + + def pmc_get_grandmaster_settings_np(self, config_file: str, socket_file: str, unicast: bool = True, boundry_clock: int = 0) -> PMCGetGrandmasterSettingsNpOutput: + """ + Gets the grandmaster_settings_np + + Args: + config_file : the config file + socket_file : the socket file + unicast : true to use unicast + boundry_clock : the boundry clock + + Example: PMCKeywords(ssh_connection).pmc_get_grandmaster_settings_np('/etc/linuxptp/ptpinstance/ptp4l-ptp5.conf', ' /var/run/ptp4l-ptp5') + + """ + cmd = f"pmc {'-u' if unicast else ''} -b {boundry_clock} -f {config_file} -s {socket_file} 'GET GRANDMASTER_SETTINGS_NP'" + + output = self.ssh_connection.send_as_sudo(cmd) + pmc_get_grandmaster_settings_np_output = PMCGetGrandmasterSettingsNpOutput(output) + return pmc_get_grandmaster_settings_np_output + + def pmc_get_time_properties_data_set(self, config_file: str, socket_file: str, unicast: bool = True, boundry_clock: int = 0) -> PMCGetTimePropertiesDataSetOutput: + """ + Gets the time_properties_data_set_object + + Args: + config_file : the config file + socket_file : the socket file + unicast : true to use unicast + boundry_clock : the boundry clock + + Example: PMCKeywords(ssh_connection).pmc_get_time_properties_data_set('/etc/linuxptp/ptpinstance/ptp4l-ptp5.conf', ' /var/run/ptp4l-ptp5') + + """ + cmd = f"pmc {'-u' if unicast else ''} -b {boundry_clock} -f {config_file} -s {socket_file} 'GET TIME_PROPERTIES_DATA_SET'" + + output = self.ssh_connection.send_as_sudo(cmd) + pmc_get_time_properties_data_set_output = PMCGetTimePropertiesDataSetOutput(output) + return pmc_get_time_properties_data_set_output + def pmc_get_default_data_set(self, config_file: str, socket_file: str, unicast: bool = True, boundry_clock: int = 0) -> PMCGetDefaultDataSetOutput: """ Gets the default data et @@ -30,6 +128,24 @@ class PMCKeywords(BaseKeyword): pmc_get_default_data_set_output = PMCGetDefaultDataSetOutput(output) return pmc_get_default_data_set_output + def pmc_get_domain(self, config_file: str, socket_file: str, unicast: bool = True, boundry_clock: int = 0) -> PMCGetDomainOutput: + """ + Gets the domain + Args: + config_file (): the config file + socket_file (): the socket file + unicast (): true to use unicast + boundry_clock (): the boundry clock + + Example: PMCKeywords(ssh_connection).pmc_get_domain('/etc/linuxptp/ptpinstance/ptp4l-ptp5.conf', ' /var/run/ptp4l-ptp5') + + """ + cmd = f"pmc {'-u' if unicast else ''} -b {boundry_clock} -f {config_file} -s {socket_file} 'GET DOMAIN'" + + output = self.ssh_connection.send_as_sudo(cmd) + pmc_get_domain_output = PMCGetDomainOutput(output) + return pmc_get_domain_output + def pmc_get_parent_data_set(self, config_file: str, socket_file: str, unicast: bool = True, boundry_clock: int = 0) -> PMCGetParentDataSetOutput: """ Gets the parent data set