Add --eval-filter

src/con_duct/suite/ls.py

@@ -3,6 +3,7 @@
 import glob
 import json
 import logging
+import re
 from typing import Any, Dict, List, Optional
 from packaging.version import Version
 
@@ -54,22 +55,30 @@
 MINIMUM_SCHEMA_VERSION: str = "0.2.0"
 
 
-def load_duct_runs(info_files: List[str]) -> List[Dict[str, Any]]:
+def load_duct_runs(
+    info_files: List[str], eval_filter: Optional[str] = None
+) -> List[Dict[str, Any]]:
     loaded: List[Dict[str, Any]] = []
     for info_file in info_files:
         with open(info_file) as file:
             try:
                 this: Dict[str, Any] = json.load(file)
                 # this["prefix"] is the path at execution time, could have moved
                 this["prefix"] = info_file.split("info.json")[0]
-                if Version(this["schema_version"]) >= Version(MINIMUM_SCHEMA_VERSION):
-                    loaded.append(this)
-                else:
+                if Version(this["schema_version"]) < Version(MINIMUM_SCHEMA_VERSION):
                     # TODO lower log level once --log-level is respected
                     lgr.warning(
                         f"Skipping {this['prefix']}, schema version {this['schema_version']} "
                         f"is below minimum schema version {MINIMUM_SCHEMA_VERSION}."
                     )
+                    continue
+                if eval_filter is not None and not (eval_results := eval(
+                    eval_filter, _flatten_dict(this), dict(re=re)
+                )):
+                    lgr.debug("Filtering out %s due to filter results matching: %s", this, eval_results)
+                    continue
+
+                loaded.append(this)
             except Exception as exc:
                 lgr.warning("Failed to load file %s: %s", file, exc)
     return loaded
@@ -148,7 +157,7 @@ def ls(args: argparse.Namespace) -> int:
         args.paths = [p for p in glob.glob(pattern)]
 
     info_files = [path for path in args.paths if path.endswith("info.json")]
-    run_data_raw = load_duct_runs(info_files)
+    run_data_raw = load_duct_runs(info_files, args.eval_filter)
     formatter = SummaryFormatter(enable_colors=args.colors)
     output_rows = process_run_data(run_data_raw, args.fields, formatter)
 

src/con_duct/suite/main.py

@@ -66,7 +66,7 @@ def main(argv: Optional[List[str]] = None) -> None:
         nargs="+",
         metavar="FIELD",
         help=f"List of fields to show. Prefix is always included implicitly as the first field. "
-        f"Available choices: {', '.join(LS_FIELD_CHOICES)}.",
+        f"Available choices: {', '.join(sorted(LS_FIELD_CHOICES))}.",
         choices=LS_FIELD_CHOICES,
         default=[
             "command",
@@ -87,6 +87,15 @@ def main(argv: Optional[List[str]] = None) -> None:
         help="Path to duct report files, only `info.json` would be considered. "
         "If not provided, the program will glob for files that match DUCT_OUTPUT_PREFIX.",
     )
+    parser_ls.add_argument(
+        "-e",
+        "--eval-filter",
+        help=f"Python expression to filter results based on available fields. "
+        f"The expression is evaluated for each entry, and only those that return True are included. "
+        f"Available fields: {', '.join(sorted(LS_FIELD_CHOICES))}. "
+        f"Example: --eval-filter \"filter_this=='yes'\" filters entries where 'filter_this' is 'yes'. "
+        f"You can use 're' for regex operations (e.g., --eval-filter \"re.search('2025.02.09.*', prefix)\").",
+    )
     parser_ls.set_defaults(func=ls)
 
     args = parser.parse_args(argv)

test/test_suite.py

@@ -4,7 +4,7 @@
 import json
 import os
 import tempfile
-from typing import Any
+from typing import Any, Optional
 import unittest
 from unittest.mock import MagicMock, mock_open, patch
 import pytest
@@ -155,10 +155,12 @@ def setUp(self) -> None:
             "file1_info.json": {
                 "schema_version": MINIMUM_SCHEMA_VERSION,
                 "prefix": "test1",
+                "filter_this": "yes",
             },
             "file2_info.json": {
                 "schema_version": MINIMUM_SCHEMA_VERSION,
                 "prefix": "test2",
+                "filter_this": "no",
             },
             "file3_info.json": {"schema_version": "0.1.0", "prefix": "old_version"},
             "not_matching.json": {
@@ -181,15 +183,19 @@ def tearDown(self) -> None:
         os.chdir(self.old_cwd)
         self.temp_dir.cleanup()
 
-    def _run_ls(self, paths: list[str], fmt: str) -> str:
+    def _run_ls(
+        self, paths: list[str], fmt: str, args: Optional[argparse.Namespace] = None
+    ) -> str:
         """Helper function to run ls() and capture stdout."""
-        args = argparse.Namespace(
-            paths=[os.path.join(self.temp_dir.name, path) for path in paths],
-            colors=False,
-            fields=["prefix", "schema_version"],
-            format=fmt,
-            func=ls,
-        )
+        if args is None:
+            args = argparse.Namespace(
+                paths=[os.path.join(self.temp_dir.name, path) for path in paths],
+                colors=False,
+                fields=["prefix", "schema_version"],
+                eval_filter=None,
+                format=fmt,
+                func=ls,
+            )
         buf = StringIO()
         with contextlib.redirect_stdout(buf):
             exit_code = ls(args)
@@ -210,6 +216,30 @@ def test_ls_sanity(self) -> None:
         assert len(prefixes) == 1
         assert any("file1" in p for p in prefixes)
 
+    def test_ls_with_filter(self) -> None:
+        """Basic sanity test to ensure ls() runs without crashing."""
+        paths = ["file1_info.json", "file2_info.json"]
+        args = argparse.Namespace(
+            paths=[os.path.join(self.temp_dir.name, path) for path in paths],
+            colors=False,
+            fields=["prefix", "schema_version"],
+            eval_filter="filter_this=='yes'",
+            format="summaries",
+            func=ls,
+        )
+        result = self._run_ls(paths, "summaries", args)
+
+        assert "Prefix:" in result
+        prefixes = [
+            line.split(":", 1)[1].strip()
+            for line in result.splitlines()
+            if line.startswith("Prefix:")
+        ]
+        assert len(prefixes) == 1
+        assert any("file1" in p for p in prefixes)
+        # filter_this == 'no'
+        assert "file2" not in result
+
     def test_ls_no_pos_args(self) -> None:
         result = self._run_ls([], "summaries")