[benchmarking] Adds audio curation benchmark to nightly

benchmarking/nightly-benchmark.yaml

@@ -28,7 +28,14 @@ datasets:
     formats:
     - type: "parquet"
       path: "{datasets_path}/commoncrawl/IDs/parquet_data"
-
+  - name: "mscoco"
+    formats:
+    - type: "wds"
+      path: "{datasets_path}/mscoco/wds/truncated_100K_mscoco_benchmarking"
+  - name: "mscoco_model_weights"
+    formats:
+    - type: "files"
+      path: "{datasets_path}/mscoco/model_weights"
 
 default_timeout_s: 7200
 
@@ -171,3 +178,32 @@ entries:
       --duplicate-id-field=_curator_dedup_id
       --blocksize=1.5GiB
     timeout_s: 20000
+
+  - name: image_curation
+    enabled: true
+    script: "{curator_repo_dir}/tutorials/image/getting-started/image_curation_example.py"
+    args: >-
+      --input-wds-dataset-dir {dataset:mscoco,wds}
+      --output-dataset-dir {session_entry_dir}/results_truncated_100K_mscoco
+      --model-dir {dataset:mscoco_model_weights,files}
+      --batch-size 100
+      --embedding-batch-size 100
+      --aesthetic-batch-size 100
+      --nsfw-batch-size 100
+      --tar-files-per-partition 10
+      --aesthetic-threshold 0.9
+      --nsfw-threshold 0.9
+      --skip-download
+      --verbose
+
+  - name: audio_fleurs
+    enabled: true
+    script: audio_fleurs_benchmark.py
+    args: >-
+      --benchmark-results-path={session_entry_dir}
+      --scratch-output-path={session_entry_dir}/scratch
+      --model-name=nvidia/stt_hy_fastconformer_hybrid_large_pc
+      --lang=hy_am
+      --split=dev
+      --wer-threshold=5.5
+      --gpus=1

benchmarking/runner/entry.py

@@ -25,13 +25,16 @@
     from runner.datasets import DatasetResolver
     from runner.path_resolver import PathResolver
 
+_curator_repo_path = Path(__file__).parent.parent.parent
+_entry_script_base_path = _curator_repo_path / "benchmarking/scripts"
+
 
 @dataclass
 class Entry:
     name: str
     script: str | None = None
     args: str | None = None
-    script_base_dir: Path = Path(__file__).parent.parent / "scripts"
+    script_base_path: Path = _entry_script_base_path
     timeout_s: int | None = None
     sink_data: list[dict[str, Any]] | dict[str, Any] = field(default_factory=dict)
     requirements: list[dict[str, Any]] | dict[str, Any] = field(default_factory=dict)
@@ -84,9 +87,7 @@ def __post_init__(self) -> None:  # noqa: C901
         else:
             msg = f"Invalid requirements type: {type(self.requirements)}"
             raise TypeError(msg)
-        # For each requirement dict in requirements, check that if both min_value and max_value are present,
-        # then max_value >= min_value. Raise ValueError if not.
-        # Raise TypeError if req is not a dict.
+        # For each requirement dict in requirements, ensure that max_value >= min_value if both are present.
         for metric_name, req in requirements.items():
             if not isinstance(req, dict):
                 msg = f"Requirement for metric '{metric_name}' is not a dict: {type(req)}"
@@ -108,13 +109,19 @@ def get_command_to_run(
         dataset_resolver: DatasetResolver,
     ) -> str:
         if self.script:
-            script_path = self.script_base_dir / self.script
+            script = self.script
+            script = self.substitute_reserved_placeholders(script, session_entry_path, dataset_resolver)
+            script = self.substitute_container_or_host_paths(script, path_resolver, allow_other_placeholders=False)
+
+            # Using the Path "/" operator means that if script is an abs path here then
+            # self.script_base_path will be ignored automatically.
+            script_path = self.script_base_path / script
             cmd = f"python {script_path} {self.args or ''}"
 
-            cmd = self.substitute_paths_in_cmd(cmd, path_resolver, dataset_resolver)
-            cmd = self.substitute_template_placeholders(cmd, session_entry_path)
+            cmd = self.substitute_reserved_placeholders(cmd, session_entry_path, dataset_resolver)
+            cmd = self.substitute_container_or_host_paths(cmd, path_resolver, allow_other_placeholders=False)
         else:
-            msg = f"Entry {self.name} must specify either cmd or script"
+            msg = f"Entry {self.name} must specify a script to run"
             raise ValueError(msg)
 
         return cmd
@@ -123,37 +130,58 @@ def get_sink_data(self, sink_name: str) -> dict[str, Any]:
         return self.sink_data.get(sink_name, {})
 
     @staticmethod
-    def substitute_paths_in_cmd(cmd: str, path_resolver: PathResolver, dataset_resolver: DatasetResolver) -> str:
-        dataset_pattern = re.compile(r"\{dataset:([^,}]+),([^}]+)\}")
-
-        def _replace_dataset(match: re.Match[str]) -> str:
-            dataset_name = match.group(1).strip()
-            dataset_format = match.group(2).strip()
-            return str(dataset_resolver.resolve(dataset_name, dataset_format))
+    def substitute_container_or_host_paths(
+        cmd: str, path_resolver: PathResolver, allow_other_placeholders: bool = True
+    ) -> str:
+        """
+        Substitute paths in the command string that are intended to be resolved by PathResolver.
 
+        This replaces placeholders in the form {path_name} with their corresponding host or container path.
+        If the placeholder does not correspond to a known path (as defined in PathResolver), it is left unchanged.
+        If allow_other_placeholders is False, then ValueError is raised if the placeholder does not correspond to a recognized path type.
+        """
         path_pattern = re.compile(r"\{([a-zA-Z_][a-zA-Z0-9_]*)\}")
 
         def _replace_path(match: re.Match[str]) -> str:
             path_name = match.group(1).strip()
             # PathResolver.resolve() only matches specific paths intended to be mapped between host and container.
-            # ValueError is raised if this is not one of those paths, meaning either the path is meant for template
-            # substitution instead or possibly should be used as-is, in which case simply return the original string.
+            # ValueError is raised if the placeholder (word inside {}) is not one of those paths.
             try:
                 return str(path_resolver.resolve(path_name))
-            except ValueError:
-                return match.group(0)
+            except ValueError as e:
+                if allow_other_placeholders:
+                    return match.group(0)
+                else:
+                    msg = f"Unknown path placeholder: {path_name}"
+                    raise ValueError(msg) from e
 
-        return path_pattern.sub(_replace_path, dataset_pattern.sub(_replace_dataset, cmd))
+        return path_pattern.sub(_replace_path, cmd)
 
     @staticmethod
-    def substitute_template_placeholders(cmd: str, session_entry_path: Path) -> str:
-        """Substitute template placeholders in command.
+    def substitute_reserved_placeholders(cmd: str, session_entry_path: Path, dataset_resolver: DatasetResolver) -> str:
+        """Substitute reserved placeholders in command.
         Example:
         - {session_entry_dir}/results.json -> /path/to/session/entry/results.json
         """
-        session_entry_pattern = re.compile(r"\{session_entry_dir\}")
+        session_entry_dir_pattern = re.compile(r"\{session_entry_dir\}")
 
-        def replace_session_entry_path(match: re.Match[str]) -> str:  # noqa: ARG001
+        def _replace_session_entry_dir(match: re.Match[str]) -> str:  # noqa: ARG001
             return str(session_entry_path)
 
-        return session_entry_pattern.sub(replace_session_entry_path, cmd)
+        curator_repo_dir_pattern = re.compile(r"\{curator_repo_dir\}")
+
+        def _replace_curator_repo_dir(match: re.Match[str]) -> str:  # noqa: ARG001
+            return str(_curator_repo_path)
+
+        dataset_pattern = re.compile(r"\{dataset:([^,}]+),([^}]+)\}")
+
+        def _replace_dataset(match: re.Match[str]) -> str:
+            dataset_name = match.group(1).strip()
+            dataset_format = match.group(2).strip()
+            return str(dataset_resolver.resolve(dataset_name, dataset_format))
+
+        new_cmd = cmd
+        new_cmd = session_entry_dir_pattern.sub(_replace_session_entry_dir, new_cmd)
+        new_cmd = curator_repo_dir_pattern.sub(_replace_curator_repo_dir, new_cmd)
+        new_cmd = dataset_pattern.sub(_replace_dataset, new_cmd)
+        return new_cmd  # noqa: RET504

benchmarking/runner/session.py

@@ -95,7 +95,7 @@ def create_from_dict(cls, data: dict) -> Session:
         mc_data = {k: v for k, v in data.items() if k in mc_field_names}
         sinks = cls.create_sinks_from_dict(mc_data.get("sinks", []))
         # Load entries only if enabled (enabled by default)
-        # TODO: should entries be created unconditionally and have an "enabled" field instead?
+        # TODO: should entries be created unconditionally and use their "enabled" field instead?
         entries = [Entry(**e) for e in mc_data["entries"] if e.get("enabled", True)]
 
         mc_data["results_path"] = path_resolver.resolve("results_path")

benchmarking/runner/utils.py

@@ -12,7 +12,9 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import json
 import os
+import pickle
 import re
 import subprocess
 from pathlib import Path
@@ -27,35 +29,42 @@
     import logging as logger
 
 
-def get_obj_for_json(obj: object) -> str | int | float | bool | list | dict:
+def write_benchmark_results(results: dict, output_path: Path) -> None:
+    """Write results to the standard files expected by the benchmark framework.
+
+    This utility is typically used by developer-written benchmark scripts to write results
+    to the standard files expected by the benchmark framework.
+    """
+    output_path.mkdir(parents=True, exist_ok=True)
+    if "params" in results:
+        (output_path / "params.json").write_text(json.dumps(results["params"], indent=2))
+    if "metrics" in results:
+        (output_path / "metrics.json").write_text(json.dumps(results["metrics"], indent=2))
+    if "tasks" in results:
+        (output_path / "tasks.pkl").write_bytes(pickle.dumps(results["tasks"]))
+
+
+# TODO: This utility contains some special cases for Slack JSON messages used in the Slack sink.
+# Consider moving these special cases to the Slack sink itself.
+def get_obj_for_json(obj: object) -> object:
     """
-    Recursively convert objects to Python primitives for JSON serialization.
-    Useful for objects like Path, sets, bytes, etc.
+    Convert common objects used in the benchmark framework to JSON-friendly primitives.
     """
     if isinstance(obj, dict):
         retval = {get_obj_for_json(k): get_obj_for_json(v) for k, v in obj.items()}
     elif isinstance(obj, (list, tuple, set)):
         retval = [get_obj_for_json(item) for item in obj]
-    elif hasattr(obj, "as_posix"):  # Path objects
-        retval = obj.as_posix()
-    elif isinstance(obj, bytes):
-        retval = obj.decode("utf-8", errors="replace")
-    elif hasattr(obj, "to_json") and callable(obj.to_json):
-        retval = obj.to_json()
-    elif hasattr(obj, "__dict__"):
-        retval = get_obj_for_json(vars(obj))
-    elif obj is None:
+    elif isinstance(obj, Path):
+        retval = str(obj)
+    elif obj is None:  # special case for Slack: JSON null not allowed, convert to string
         retval = "null"
-    elif isinstance(obj, str) and len(obj) == 0:  # special case for Slack, empty strings not allowed
+    elif isinstance(obj, str) and len(obj) == 0:  # special case for Slack: empty strings not allowed
         retval = " "
     else:
         retval = obj
     return retval
 
 
-_env_var_pattern = re.compile(r"\$\{([^}]+)\}")  # Pattern to match ${VAR_NAME}
-
-
 def _replace_env_var(match: re.Match[str]) -> str:
     env_var_name = match.group(1)
     env_value = os.getenv(env_var_name)
@@ -66,6 +75,9 @@ def _replace_env_var(match: re.Match[str]) -> str:
         raise ValueError(msg)
 
 
+_env_var_pattern = re.compile(r"\$\{([^}]+)\}")  # Pattern to match ${VAR_NAME}
+
+
 def resolve_env_vars(data: dict | list | str | object) -> dict | list | str | object:
     """Recursively resolve environment variables in strings in/from various objects.