Handle reading scenario from file in factory method

Author: sjmonson

src/guidellm/__main__.py

@@ -284,18 +284,7 @@ def benchmark(
 ):
     click_ctx = click.get_current_context()
 
-    # If a scenario file was specified read from it
-    # TODO: This should probably be a factory method
-    if scenario is None:
-        _scenario = {}
-    else:
-        # TODO: Support pre-defined scenarios
-        # TODO: Support other formats
-        with Path(scenario).open() as f:
-            _scenario = json.load(f)
-
-    # If any command line arguments are specified, override the scenario
-    _scenario.update(set_if_not_default(
+    overrides = set_if_not_default(
         click_ctx,
         target=target,
         backend_type=backend_type,
@@ -314,11 +303,18 @@ def benchmark(
         cooldown_percent=cooldown_percent,
         output_sampling=output_sampling,
         random_seed=random_seed,
-    ))
+    )
+
+    # If a scenario file was specified read from it
+    if scenario is None:
+        _scenario = GenerativeTextScenario.model_validate(overrides)
+    else:
+        # TODO: Support pre-defined scenarios
+        _scenario = GenerativeTextScenario.from_file(scenario, overrides)
 
     asyncio.run(
         benchmark_with_scenario(
-            scenario=GenerativeTextScenario(**_scenario),
+            scenario=_scenario,
             show_progress=not disable_progress,
             show_progress_scheduler_stats=display_scheduler_stats,
             output_console=not disable_console_outputs,

src/guidellm/benchmark/scenario.py

@@ -1,8 +1,11 @@
+import json
 from collections.abc import Iterable
 from pathlib import Path
-from typing import Any, Literal, Optional, Self, Union
+from typing import Any, Literal, Optional, TypeVar, Union
 
+import yaml
 from datasets import Dataset, DatasetDict, IterableDataset, IterableDatasetDict
+from loguru import logger
 from transformers.tokenization_utils_base import (  # type: ignore[import]
     PreTrainedTokenizerBase,
 )
@@ -14,20 +17,28 @@
 
 __ALL__ = ["Scenario", "GenerativeTextScenario"]
 
+T = TypeVar("T", bound="Scenario")
+
 
 class Scenario(StandardBaseModel):
     target: str
 
-    def _update(self, **fields: Any) -> Self:
-        for k, v in fields.items():
-            if not hasattr(self, k):
-                raise ValueError(f"Invalid field {k}")
-            setattr(self, k, v)
-
-        return self
+    @classmethod
+    def from_file(
+        cls: type[T], filename: Union[str, Path], overrides: Optional[dict] = None
+    ) -> T:
+        try:
+            with open(filename) as f:
+                if str(filename).endswith(".yaml") or str(filename).endswith(".yml"):
+                    data = yaml.safe_load(f)
+                else:  # Assume everything else is json
+                    data = json.load(f)
+        except (json.JSONDecodeError, yaml.YAMLError) as e:
+            logger.error("Failed to parse scenario")
+            raise e
 
-    def update(self, **fields: Any) -> Self:
-        return self._update(**{k: v for k, v in fields.items() if v is not None})
+        data.update(overrides)
+        return cls.model_validate(data)
 
 
 class GenerativeTextScenario(Scenario):