rocm-systems/projects/rocprofiler-compute/tools/config_management/apply_config_deltas.py

#!/usr/bin/env python3
##############################################################################
# MIT License
#
# Copyright (c) 2025 Advanced Micro Devices, Inc. All Rights Reserved.
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.  IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
# THE SOFTWARE.

##############################################################################

"""
Apply delta YAML to base architecture to produce target architecture.
Usage: python apply_config_deltas.py <base_arch_dir> <delta_yaml> <output_dir>
"""

from __future__ import annotations

import shutil
import sys
from pathlib import Path
from typing import Any, Optional, Union

PROJECT_ROOT = Path(__file__).resolve().parents[1]
if str(PROJECT_ROOT) not in sys.path:
    sys.path.insert(0, str(PROJECT_ROOT))

from config_management import utils_ruamel as cm_utils  # noqa: E402


def find_table(config: dict, table_id: Any) -> Optional[dict]:
    """Find and return the table with given id, or None."""
    for item in config.get("Panel Config", {}).get("data source", []):
        table = item.get("metric_table")
        if isinstance(table, dict) and table.get("id") == table_id:
            return table
    return None


def add_table(config: dict, metric_table: dict) -> None:
    """Add entire new table to config."""
    config.setdefault("Panel Config", {}).setdefault("data source", []).append({
        "metric_table": metric_table
    })
    print(f"Added table: {metric_table.get('id')} - {metric_table.get('title')}")


def add_metrics(config: dict, table_id: Any, metrics: list[dict]) -> None:
    """Add metrics to existing table."""
    table = find_table(config, table_id)
    if table is None:
        print(f"WARNING: Table {table_id} not found for metric addition")
        return

    table.setdefault("metric", {})
    for metric_dict in metrics:
        for metric_name, metric_data in metric_dict.items():
            table["metric"][metric_name] = metric_data
            print(f"Added metric: {metric_name} to table {table_id}")


def delete_table(config: dict, table_id: Any) -> None:
    """Remove entire table from config."""
    data_source = config.get("Panel Config", {}).get("data source", [])
    for idx, item in enumerate(list(data_source)):
        table = item.get("metric_table")
        if isinstance(table, dict) and table.get("id") == table_id:
            del data_source[idx]
            print(f"Deleted table: {table_id}")
            return
    print(f"WARNING: Table {table_id} not found for deletion")


def delete_metrics(config: dict, table_id: Any, metrics: list[dict]) -> None:
    """Remove specific metrics from table."""
    table = find_table(config, table_id)
    if table is None or "metric" not in table:
        print(f"WARNING: Table {table_id} not found or has no metrics")
        return

    for metric_dict in metrics:
        for metric_name in metric_dict.keys():
            if metric_name in table["metric"]:
                del table["metric"][metric_name]
                print(f"Deleted metric: {metric_name} from table {table_id}")


def modify_metrics(config: dict, table_id: Any, metrics: list[dict]) -> None:
    """Modify specific fields in existing metrics."""
    table = find_table(config, table_id)
    if table is None or "metric" not in table:
        print(f"WARNING: Table {table_id} not found or has no metrics")
        return

    for metric_dict in metrics:
        for metric_name, new_fields in metric_dict.items():
            if metric_name not in table["metric"]:
                print(f"WARNING: Metric '{metric_name}' not found in table {table_id}")
                continue
            for field_name, field_value in new_fields.items():
                table["metric"][metric_name][field_name] = field_value
                print(f"Modified {metric_name}.{field_name} in table {table_id}")


def add_descriptions(config: dict, descriptions: dict) -> None:
    """Add metric descriptions to config."""
    md = config["Panel Config"].setdefault("metrics_description", {})

    for metric_name, desc_data in descriptions.items():
        md[metric_name] = dict(desc_data) if isinstance(desc_data, dict) else desc_data

        print(f"Added description: {metric_name}")


def delete_descriptions(config: dict, descriptions: dict) -> None:
    """Remove metric descriptions from config."""
    md = config["Panel Config"].setdefault("metrics_description", {})
    for metric_name in descriptions.keys():
        if metric_name in md:
            del md[metric_name]
            print(f"Deleted description: {metric_name}")


def modify_descriptions(config: dict, descriptions: dict) -> None:
    """Modify metric descriptions in config."""
    md = config["Panel Config"].setdefault("metrics_description", {})

    for metric_name, desc_data in descriptions.items():
        if isinstance(desc_data, dict):
            new_dict = {}
            for k, v in desc_data.items():
                new_dict[k] = v
            md[metric_name] = new_dict
        else:
            md[metric_name] = desc_data

        print(f"Added description: {metric_name}")


def apply_changes(config: dict, changes: list[dict], category: str) -> None:
    """Apply delta changes to configuration."""
    for change in changes:
        mt = change.get("metric_table")
        if mt:
            table_id = mt.get("id")

            if category == "Addition":
                if "metrics" in mt:
                    add_metrics(config, table_id, mt["metrics"])
                elif "metric" in mt:
                    add_table(config, mt)

            elif category == "Deletion":
                if "metrics" in mt:
                    delete_metrics(config, table_id, mt["metrics"])
                else:
                    delete_table(config, table_id)

            elif category == "Modification":
                if "metrics" in mt:
                    modify_metrics(config, table_id, mt["metrics"])

        descriptions = change.get("metric_descriptions", {})
        if descriptions:
            if category == "Addition":
                add_descriptions(config, descriptions)
            elif category == "Deletion":
                delete_descriptions(config, descriptions)
            elif category == "Modification":
                modify_descriptions(config, descriptions)


def apply_delta(
    base_dir: Union[str, Path],
    delta_file: Union[str, Path],
    output_dir: Union[str, Path],
) -> None:
    """Apply delta YAML to all files in base directory."""
    delta = cm_utils.load_yaml(delta_file, round_trip=True)
    output_path = Path(output_dir)
    output_path.mkdir(parents=True, exist_ok=True)

    changes_by_panel: dict[Any, dict[str, list[dict]]] = {}
    for category in ("Addition", "Deletion", "Modification"):
        for change in delta.get(category, []):
            panel_id = change.get("Panel Config", {}).get("id")
            panel_bucket = changes_by_panel.setdefault(
                panel_id, {"Addition": [], "Deletion": [], "Modification": []}
            )
            panel_bucket[category].append(change)

    base_path = Path(base_dir)
    for yaml_file in base_path.glob("*.yaml"):
        config = cm_utils.load_yaml(yaml_file, round_trip=True)
        panel_id = config.get("Panel Config", {}).get("id")

        if panel_id in changes_by_panel:
            print(f"\nApplying deltas to {yaml_file.name} (Panel ID: {panel_id})")
            # Process in safe order: deletions -> modifications -> additions
            for category in ("Deletion", "Modification", "Addition"):
                if changes_by_panel[panel_id][category]:
                    apply_changes(
                        config, changes_by_panel[panel_id][category], category
                    )

            cm_utils.strip_existing_header(config)
            cm_utils.save_yaml(config, output_path / yaml_file.name)
            print(f"Saved: {yaml_file.name}")
        else:
            shutil.copy(yaml_file, output_path / yaml_file.name)


def main() -> None:
    if len(sys.argv) != 4:
        print(
            "Usage: python apply_config_deltas.py "
            "<base_arch_dir> <delta_yaml> <output_dir>"
        )
        sys.exit(1)

    base_dir, delta_file, output_dir = sys.argv[1:4]

    if not Path(base_dir).is_dir():
        print(f"Error: {base_dir} is not a directory")
        sys.exit(1)

    if not Path(delta_file).is_file():
        print(f"Error: {delta_file} is not a file")
        sys.exit(1)

    apply_delta(base_dir, delta_file, output_dir)
    print("\nDelta application complete!")


if __name__ == "__main__":
    main()