From a43764e8176c095e8092202f14cbb6dc80fbb132 Mon Sep 17 00:00:00 2001 From: Atharva Sawant Date: Mon, 6 Jul 2026 13:25:28 +0530 Subject: [PATCH] feat: add compare_ppl_configs, get_ppl_history, and plf compare/log CLI Add management utilities to diff pipeline configs and inspect run history, with full pytest coverage for the new API and CLI commands. Co-authored-by: Cursor --- README.md | 4 + src/plf/cli.py | 45 ++++++++++++ src/plf/experiment.py | 153 ++++++++++++++++++++++++++++++++++++++- tests/test_cli.py | 62 ++++++++++++++++ tests/test_experiment.py | 114 +++++++++++++++++++++++++++++ tests/test_utils.py | 33 +++++++++ 6 files changed, 408 insertions(+), 3 deletions(-) create mode 100644 tests/test_experiment.py create mode 100644 tests/test_utils.py diff --git a/README.md b/README.md index 76a37e8..497c525 100644 --- a/README.md +++ b/README.md @@ -199,6 +199,8 @@ The `plf.experiment` module provides powerful tools for managing your PPL databa * **`get_ppls()`**: List all active pipeline IDs in the current Lab. * **`get_ppl_status()`**: Returns a DataFrame summarizing the status, last run, and key metrics for all PPLs. +* **`compare_ppl_configs(pplid_a, pplid_b)`**: Compare two pipeline configs and return a structured diff of their `workflow` and `args` sections. +* **`get_ppl_history(pplid)`**: Return a DataFrame of run timestamps and session origins for a single pipeline. * **`filter_ppls(query)`**: Filters PPLs based on configuration arguments (e.g., `filter_ppls("data_source=my_workflows.MyComponent")`). * **`archive_ppl(ppls)`**: Archives a pipeline, moving its configurations and artifacts to an archived folder for safe storage. * **`archive_ppl(ppls, reverse=True)`**: Unarchives a pipeline and returns it to the active environment. @@ -219,6 +221,8 @@ plf run # run a pipeline plf archive # archive a pipeline plf delete # delete from archive plf stop # gracefully stop running pipeline(s) +plf compare # diff two pipeline configs +plf log # show run history for a pipeline plf export --format yaml # export config (yaml or json) plf init --config settings.json # initialize lab from config file ``` diff --git a/src/plf/cli.py b/src/plf/cli.py index 25df574..e1b2967 100644 --- a/src/plf/cli.py +++ b/src/plf/cli.py @@ -15,7 +15,9 @@ from plf.experiment import ( PipeLine, archive_ppl, + compare_ppl_configs, delete_ppl, + get_ppl_history, get_ppl_status, get_ppls, get_runnings, @@ -83,6 +85,49 @@ def list_ppls(ctx): click.echo(ppl) +@main.command() +@click.argument("pplid_a") +@click.argument("pplid_b") +@click.pass_context +def compare(ctx, pplid_a, pplid_b): + """Compare configs of two pipelines and show differences.""" + _load_lab(ctx.obj["settings"]) + try: + result = compare_ppl_configs(pplid_a, pplid_b) + except ValueError as exc: + click.echo(f"Error: {exc}", err=True) + sys.exit(1) + + if result["identical"]: + click.echo(f"Configs for '{pplid_a}' and '{pplid_b}' are identical.") + return + + click.echo(f"Differences between '{pplid_a}' and '{pplid_b}':") + for path, values in sorted(result["differences"].items()): + click.echo(f" {path}:") + click.echo(f" {pplid_a}: {values['left']!r}") + click.echo(f" {pplid_b}: {values['right']!r}") + + +@main.command(name="log") +@click.argument("pplid") +@click.pass_context +def show_log(ctx, pplid): + """Show run history for a pipeline.""" + _load_lab(ctx.obj["settings"]) + try: + df = get_ppl_history(pplid) + except ValueError as exc: + click.echo(f"Error: {exc}", err=True) + sys.exit(1) + + if df.empty: + click.echo(f"No run history found for '{pplid}'.") + return + + click.echo(df.to_string(index=False)) + + @main.command() @click.argument("pplid") @click.pass_context diff --git a/src/plf/experiment.py b/src/plf/experiment.py index 7ed97bb..5f3811a 100644 --- a/src/plf/experiment.py +++ b/src/plf/experiment.py @@ -2,7 +2,7 @@ This module have all function for initiating pipeline and training """ -from typing import Optional, Dict, List +from typing import Any, Optional, Dict, List, Union import json import os import shutil @@ -15,15 +15,17 @@ __all__ = [ - "PipeLine","TransferContext", + "PipeLine", "TransferContext", "get_ppls", "get_ppl_details", "get_ppl_status", + "get_ppl_history", + "compare_ppl_configs", "archive_ppl", "delete_ppl", "transfer_ppl", "group_by_common_columns", - 'filter_ppls' + "filter_ppls", ] @@ -83,6 +85,151 @@ def get_ppl_status(ppls: Optional[list] = None) -> pd.DataFrame: df = pd.DataFrame.from_dict(data, orient='index') return df + +def _extract_comparable_config(cnfg: Dict[str, Any]) -> Dict[str, Any]: + """Return the parts of a config that define reproducibility (used for hashing). + + Args: + cnfg: Full pipeline configuration dictionary loaded from disk. + + Returns: + A dictionary containing only ``workflow`` and ``args`` keys. + """ + return { + "workflow": cnfg.get("workflow"), + "args": cnfg.get("args"), + } + + +def _config_diff( + left: Dict[str, Any], + right: Dict[str, Any], + path: str = "", +) -> Dict[str, Dict[str, Any]]: + """Recursively compare two config dictionaries and collect differences. + + Args: + left: Configuration from the first pipeline. + right: Configuration from the second pipeline. + path: Dot-separated key path used during recursion (internal). + + Returns: + Mapping of dotted key paths to ``{"left": ..., "right": ...}`` entries. + """ + differences: Dict[str, Dict[str, Any]] = {} + all_keys = set(left.keys()) | set(right.keys()) + + for key in sorted(all_keys): + current_path = f"{path}.{key}" if path else key + + if key not in left: + differences[current_path] = {"left": None, "right": right[key]} + continue + if key not in right: + differences[current_path] = {"left": left[key], "right": None} + continue + + left_val = left[key] + right_val = right[key] + + # Recurse into nested dicts (e.g. workflow.args, component configs). + if isinstance(left_val, dict) and isinstance(right_val, dict): + differences.update(_config_diff(left_val, right_val, current_path)) + elif left_val != right_val: + differences[current_path] = {"left": left_val, "right": right_val} + + return differences + + +def compare_ppl_configs(pplid_a: str, pplid_b: str) -> Dict[str, Any]: + """Compare reproducibility-relevant settings of two pipelines. + + Loads both pipeline configs and diffs their ``workflow`` and ``args`` + sections — the same fields used when computing the config hash. + + Args: + pplid_a: ID of the first pipeline. + pplid_b: ID of the second pipeline. + + Returns: + A result dictionary with keys: + + - ``identical`` (bool): True when no differences were found. + - ``pplid_a`` / ``pplid_b`` (str): The compared pipeline IDs. + - ``differences`` (dict): Dotted-path diffs, each with ``left`` and + ``right`` values (``None`` when a key exists in only one config). + + Raises: + ValueError: If either pipeline ID is not found in the active lab. + """ + pipeline_a = PipeLine() + if not pipeline_a.verify(pplid=pplid_a): + raise ValueError(f"Pipeline '{pplid_a}' not found") + pipeline_a.load(pplid=pplid_a) + + pipeline_b = PipeLine() + if not pipeline_b.verify(pplid=pplid_b): + raise ValueError(f"Pipeline '{pplid_b}' not found") + pipeline_b.load(pplid=pplid_b) + + comparable_a = _extract_comparable_config(pipeline_a.cnfg) + comparable_b = _extract_comparable_config(pipeline_b.cnfg) + differences = _config_diff(comparable_a, comparable_b) + + return { + "identical": len(differences) == 0, + "pplid_a": pplid_a, + "pplid_b": pplid_b, + "differences": differences, + } + + +def get_ppl_history(pplid: str) -> pd.DataFrame: + """Return the run history for a single pipeline. + + Queries the ``runnings`` table in ``ppls.db`` and enriches each row with + session origin information from ``logs.db`` (via ``logid``). + + Args: + pplid: Pipeline ID to look up. + + Returns: + A DataFrame with columns ``runid``, ``pplid``, ``logid``, ``parity``, + ``started_time``, and ``called_at``. Empty when the pipeline has never + been executed. + + Raises: + ValueError: If the pipeline ID is not found in the active lab. + """ + data_path = os.path.abspath(get_shared_data()["data_path"]) + + pipeline = PipeLine() + if not pipeline.verify(pplid=pplid): + raise ValueError(f"Pipeline '{pplid}' not found") + + ppls_db = Db(db_path=os.path.join(data_path, "ppls.db")) + rows = ppls_db.query( + "SELECT runid, pplid, logid, parity, started_time " + "FROM runnings WHERE pplid = ? ORDER BY started_time", + (pplid,), + ) + ppls_db.close() + + columns = ["runid", "pplid", "logid", "parity", "started_time"] + if not rows: + return pd.DataFrame(columns=columns + ["called_at"]) + + history = pd.DataFrame(rows, columns=columns) + + # Join session metadata from logs.db (separate SQLite file). + logs_db = Db(db_path=os.path.join(data_path, "logs.db")) + log_rows = logs_db.query("SELECT logid, called_at FROM logs") + logs_db.close() + log_map = {logid: called_at for logid, called_at in log_rows} + history["called_at"] = history["logid"].map(log_map) + + return history + def multi_run(ppls: Dict[str, int], last_epoch: int = 10, patience: int = 5) -> None: """ Train multiple pipelines up to a maximum number of epochs with optional patience. diff --git a/tests/test_cli.py b/tests/test_cli.py index 8fe8b43..dc2ef47 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -50,6 +50,68 @@ def test_status_empty(mock_status, mock_setup): assert "No pipelines found." in result.output +@patch("plf.cli.lab_setup") +@patch("plf.cli.compare_ppl_configs") +def test_compare_identical(mock_compare, mock_setup): + mock_compare.return_value = { + "identical": True, + "pplid_a": "ppl_a", + "pplid_b": "ppl_b", + "differences": {}, + } + result = invoke(["compare", "ppl_a", "ppl_b"]) + assert result.exit_code == 0 + assert "identical" in result.output + + +@patch("plf.cli.lab_setup") +@patch("plf.cli.compare_ppl_configs") +def test_compare_differences(mock_compare, mock_setup): + mock_compare.return_value = { + "identical": False, + "pplid_a": "ppl_a", + "pplid_b": "ppl_b", + "differences": { + "args.lr": {"left": 0.01, "right": 0.001}, + }, + } + result = invoke(["compare", "ppl_a", "ppl_b"]) + assert result.exit_code == 0 + assert "args.lr" in result.output + + +@patch("plf.cli.lab_setup") +@patch("plf.cli.compare_ppl_configs", side_effect=ValueError("Pipeline 'x' not found")) +def test_compare_not_found(mock_compare, mock_setup): + result = invoke(["compare", "ppl_a", "x"]) + assert result.exit_code == 1 + assert "not found" in result.output + + +@patch("plf.cli.lab_setup") +@patch("plf.cli.get_ppl_history") +def test_log_with_history(mock_history, mock_setup): + mock_history.return_value = pd.DataFrame({ + "runid": [1], + "pplid": ["ppl_001"], + "logid": ["log0"], + "parity": [None], + "started_time": ["2025-01-01 00:00:00"], + "called_at": ["script:test.py"], + }) + result = invoke(["log", "ppl_001"]) + assert result.exit_code == 0 + assert "ppl_001" in result.output + + +@patch("plf.cli.lab_setup") +@patch("plf.cli.get_ppl_history", return_value=pd.DataFrame()) +def test_log_empty(mock_history, mock_setup): + result = invoke(["log", "ppl_001"]) + assert result.exit_code == 0 + assert "No run history found" in result.output + + @patch("plf.cli.lab_setup") @patch("plf.cli.PipeLine") def test_run(mock_pipeline_cls, mock_setup): diff --git a/tests/test_experiment.py b/tests/test_experiment.py new file mode 100644 index 0000000..d3d2828 --- /dev/null +++ b/tests/test_experiment.py @@ -0,0 +1,114 @@ +"""Tests for plf.experiment management utilities.""" + +import json +import os + +import pytest + +from plf.experiment import compare_ppl_configs, get_ppl_history +from plf.utils import Db, hash_args + + +def _register_ppl(data_path: str, pplid: str, cnfg: dict) -> None: + """Insert a pipeline record and write its config file (no workflow needed).""" + configs_dir = os.path.join(data_path, "Configs") + os.makedirs(configs_dir, exist_ok=True) + + config_path = os.path.join(configs_dir, f"{pplid}.json") + with open(config_path, "w", encoding="utf-8") as handle: + json.dump(cnfg, handle) + + args_for_hash = {"workflow": cnfg["workflow"], "args": cnfg["args"]} + db = Db(db_path=os.path.join(data_path, "ppls.db")) + db.execute( + "INSERT INTO ppls (pplid, args_hash) VALUES (?, ?)", + (pplid, hash_args(args_for_hash)), + ) + db.close() + + +def _sample_config(pplid: str, learning_rate: float) -> dict: + """Build a minimal pipeline config for testing.""" + return { + "pplid": pplid, + "workflow": {"loc": "examples.DemoWorkflow", "args": {}}, + "args": { + "data_source": { + "loc": "examples.DataComponent", + "args": {"learning_rate": learning_rate}, + } + }, + } + + +def test_compare_ppl_configs_identical(setup_lab_env): + """Two pipelines with the same workflow/args should compare as identical.""" + data_path = setup_lab_env["settings"]["data_path"] + cnfg_a = _sample_config("ppl_alpha", learning_rate=0.01) + cnfg_b = _sample_config("ppl_beta", learning_rate=0.01) + + _register_ppl(data_path, "ppl_alpha", cnfg_a) + _register_ppl(data_path, "ppl_beta", cnfg_b) + + result = compare_ppl_configs("ppl_alpha", "ppl_beta") + assert result["identical"] is True + assert result["differences"] == {} + + +def test_compare_ppl_configs_detects_differences(setup_lab_env): + """Changed args should appear in the differences dict.""" + data_path = setup_lab_env["settings"]["data_path"] + _register_ppl(data_path, "ppl_low", _sample_config("ppl_low", 0.001)) + _register_ppl(data_path, "ppl_high", _sample_config("ppl_high", 0.1)) + + result = compare_ppl_configs("ppl_low", "ppl_high") + assert result["identical"] is False + assert "args.data_source.args.learning_rate" in result["differences"] + diff = result["differences"]["args.data_source.args.learning_rate"] + assert diff["left"] == 0.001 + assert diff["right"] == 0.1 + + +def test_compare_ppl_configs_missing_pipeline(setup_lab_env): + """Requesting an unknown pipeline ID should raise ValueError.""" + data_path = setup_lab_env["settings"]["data_path"] + _register_ppl(data_path, "ppl_exists", _sample_config("ppl_exists", 0.01)) + + with pytest.raises(ValueError, match="not found"): + compare_ppl_configs("ppl_exists", "ppl_missing") + + +def test_get_ppl_history_empty(setup_lab_env): + """A pipeline that has never run should return an empty DataFrame.""" + data_path = setup_lab_env["settings"]["data_path"] + _register_ppl(data_path, "ppl_idle", _sample_config("ppl_idle", 0.01)) + + history = get_ppl_history("ppl_idle") + assert history.empty + assert "called_at" in history.columns + + +def test_get_ppl_history_with_runs(setup_lab_env): + """Run records should be returned with session info from logs.db.""" + data_path = setup_lab_env["settings"]["data_path"] + _register_ppl(data_path, "ppl_runner", _sample_config("ppl_runner", 0.01)) + + ppls_db = Db(db_path=os.path.join(data_path, "ppls.db")) + ppls_db.execute( + "INSERT INTO runnings (pplid, logid, parity) VALUES (?, ?, ?)", + ("ppl_runner", "log0", None), + ) + ppls_db.close() + + history = get_ppl_history("ppl_runner") + assert len(history) == 1 + assert history.iloc[0]["pplid"] == "ppl_runner" + assert history.iloc[0]["logid"] == "log0" + # log0 is created during lab_setup in the fixture. + assert history.iloc[0]["called_at"] is not None + + +def test_get_ppl_history_missing_pipeline(setup_lab_env): + """Unknown pipeline IDs should raise ValueError.""" + with pytest.raises(ValueError, match="not found"): + get_ppl_history("does_not_exist") diff --git a/tests/test_utils.py b/tests/test_utils.py new file mode 100644 index 0000000..6be8278 --- /dev/null +++ b/tests/test_utils.py @@ -0,0 +1,33 @@ +"""Tests for plf.utils helper functions.""" + +from plf.utils import hash_args + + +def test_hash_args_is_deterministic(): + """Same input dict must always produce the same SHA-256 hash.""" + config = {"workflow": {"loc": "pkg.WF"}, "args": {"lr": 0.01}} + first = hash_args(config) + second = hash_args(config) + assert first == second + + +def test_hash_args_key_order_independent(): + """JSON sorting ensures key order does not change the hash.""" + config_a = {"b": 2, "a": 1} + config_b = {"a": 1, "b": 2} + assert hash_args(config_a) == hash_args(config_b) + + +def test_hash_args_detects_value_changes(): + """Different values must produce different hashes.""" + base = {"args": {"epochs": 10}} + changed = {"args": {"epochs": 20}} + assert hash_args(base) != hash_args(changed) + + +def test_hash_args_returns_hex_string(): + """Hash output should be a 64-character lowercase hex string.""" + result = hash_args({"key": "value"}) + assert isinstance(result, str) + assert len(result) == 64 + assert all(ch in "0123456789abcdef" for ch in result)