import copy from typing import Any, Dict, List, Optional, Tuple import json5 from jinja2 import Template from config.configuration_manager import ConfigurationManager from framework.resources.resource_finder import get_stx_resource_path from keywords.base_keyword import BaseKeyword from keywords.ptp.setup.object.ptp_setup import PTPSetup class PTPSetupKeywords(BaseKeyword): """ This class is responsible for reading ptp_setup files and matching them with the ptp config. """ def generate_ptp_setup_from_template(self, template_file_location: str) -> PTPSetup: """ This function will read the template_file specified and will replace values based on the ptp_config. Args: template_file_location (str): Path in the repo to the Template JSON5 file. e.g. 'resources/ptp/setup/ptp_setup_template.json5' Returns: PTPSetup: An object representing the setup. """ # Load the Template JSON5 file. with open(template_file_location, "r") as template_file: json5_template = template_file.read() ptp_default_status_values_file_path = get_stx_resource_path("resources/ptp/ptp_default_status_values.json5") with open(ptp_default_status_values_file_path, "r") as ptp_default_status_values_template_file: ptp_default_status_values_template = json5.load(ptp_default_status_values_template_file) # Build a replacement dictionary from the PTP Config ptp_config = ConfigurationManager.get_ptp_config() replacement_dictionary = ptp_config.get_all_hosts_dictionary() # Update lab_topology dict with ptp_default_status_values replacement_dictionary.update(ptp_default_status_values_template) # Render the JSON5 file by replacing the tokens. template = Template(json5_template) rendered_json_string = template.render(replacement_dictionary) json_data = json5.loads(rendered_json_string) # Turn the JSON Data into a ptp_setup object. ptp_setup = PTPSetup(json_data) return ptp_setup def filter_and_render_ptp_config(self, template_file_location: str, selected_instances: List[Tuple[str, str, List[str]]], custom_expected_dict_template: Optional[str] = None, expected_dict_overrides: Optional[Dict[str, Any]] = None) -> PTPSetup: """ Filters and renders a PTP configuration from a JSON5 template based on selected instances. This function is useful for generating a partial configuration from a complete PTP setup, based on specific instance names, hostnames, and interfaces. It also allows: - Overriding expected_dict via a custom Jinja2 template string - Applying deep overrides to specific values (e.g., changing `clock_class`) Args: template_file_location (str): Path to the JSON5 template file. selected_instances (List[Tuple[str, str, List[str]]]): List of tuples, where each tuple contains: - ptp_instance_name (str) - hostname (str) - list of associated interface names (List[str]) custom_expected_dict_template (Optional[str]): Jinja2-formatted string representing an expected_dict override. If provided, it replaces the auto-filtered expected_dict. expected_dict_overrides (Optional[Dict[str, Any]]): A dictionary of specific overrides to apply on the generated or provided expected_dict. Supports nested structure (e.g. overriding grandmaster_settings -> clock_class). Returns: PTPSetup: A PTPSetup object containing the filtered configuration. Example: filter_and_render_ptp_config( template_file_location="resources/ptp/setup/ptp_setup_template.json5", selected_instances=[("ptp4", "controller-1", ["ptp4if1"])], expected_dict_overrides={ "ptp4l": [ { "name": "ptp4", "controller-1": { "grandmaster_settings": { "clock_class": 165 } } } ] } ) """ # Load and render the JSON5 template with open(template_file_location, "r") as template_file: json5_template = template_file.read() ptp_defaults_path = get_stx_resource_path("resources/ptp/ptp_default_status_values.json5") with open(ptp_defaults_path, "r") as defaults_file: ptp_defaults = json5.load(defaults_file) ptp_config = ConfigurationManager.get_ptp_config() render_context = ptp_config.get_all_hosts_dictionary() render_context.update(ptp_defaults) # Render main config template rendered_config = Template(json5_template).render(render_context) ptp_config_dict = json5.loads(rendered_config) # Optionally render custom expected_dict custom_expected_dict = None if custom_expected_dict_template: rendered_custom_expected = Template(custom_expected_dict_template).render(render_context) custom_expected_dict = json5.loads(rendered_custom_expected) filtered_json = {"ptp_instances": {"ptp4l": []}, "ptp_host_ifs": [], "expected_dict": {"ptp4l": []}} ptp_selection = {} all_required_ifaces = set() for ptp_name, hostname, iface_list in selected_instances: if ptp_name not in ptp_selection: ptp_selection[ptp_name] = {} ptp_selection[ptp_name][hostname] = iface_list all_required_ifaces.update(iface_list) # Filter ptp_instances.ptp4l for instance in ptp_config_dict.get("ptp_instances", {}).get("ptp4l", []): name = instance.get("name") if name in ptp_selection: hosts = ptp_selection[name] selected_ifaces = [iface for iface_list in hosts.values() for iface in iface_list] filtered_json["ptp_instances"]["ptp4l"].append( { "name": name, "instance_hostnames": list(hosts.keys()), "instance_parameters": instance.get("instance_parameters", ""), "ptp_interface_names": selected_ifaces, } ) # Filter ptp_host_ifs for iface in ptp_config_dict.get("ptp_host_ifs", []): if iface.get("name") in all_required_ifaces: filtered_json["ptp_host_ifs"].append(iface) # Use custom expected_dict if provided if custom_expected_dict: filtered_json["expected_dict"]["ptp4l"] = custom_expected_dict.get("ptp4l", []) return PTPSetup(filtered_json) # Auto-generate expected_dict by filtering for expected_instance in ptp_config_dict.get("expected_dict", {}).get("ptp4l", []): name = expected_instance.get("name") if name not in ptp_selection: continue filtered_instance = {"name": name} for hostname, _ in ptp_selection[name].items(): instance_data = expected_instance.get(hostname) if not instance_data: continue filtered_instance[hostname] = {key: instance_data.get(key) for key in ["parent_data_set", "time_properties_data_set", "grandmaster_settings", "port_data_set"]} filtered_json["expected_dict"]["ptp4l"].append(filtered_instance) # Apply single-value overrides (like clock_class) if expected_dict_overrides: for override in expected_dict_overrides.get("ptp4l", []): override_name = override.get("name") for inst in filtered_json["expected_dict"]["ptp4l"]: if inst.get("name") == override_name: for hostname, host_data in override.items(): if hostname == "name": continue inst.setdefault(hostname, {}) inst[hostname] = self.deep_merge(inst[hostname], host_data) return PTPSetup(filtered_json) def deep_merge(self, dest: Dict[str, Any], src: Dict[str, Any]) -> Dict[str, Any]: """ Recursively merges the contents of `src` into `dest`. If both `dest` and `src` contain a value for the same key and both values are dictionaries, they will be merged recursively. Otherwise, the value from `src` overrides the one in `dest`. This is useful for applying nested configuration overrides without losing existing structure. Args: dest (Dict[str, Any]): The original dictionary to merge into. src (Dict[str, Any]): The dictionary containing overriding or additional values. Returns: Dict[str, Any]: A new dictionary representing the merged result. Example: deep_merge({"ptp4l": [{"name": "ptp4", "controller-1": {"grandmaster_settings": {"clock_class": 165}}}]}} """ result = copy.deepcopy(dest) for key, value in src.items(): if isinstance(value, dict) and isinstance(result.get(key), dict): result[key] = self.deep_merge(result[key], value) else: result[key] = value return result