From af61b590304be931d3f1a10ea89fbed1d3f24f77 Mon Sep 17 00:00:00 2001 From: KhushviB Date: Mon, 13 Jul 2026 19:41:50 +0530 Subject: [PATCH 1/2] feat: Infer DSE parameter encoding from candidate values at parse time --- src/cloudai/configurator/env_params.py | 69 ++++++++++++++++++++++++-- tests/test_env_params.py | 55 ++++++++++++++++++++ 2 files changed, 119 insertions(+), 5 deletions(-) diff --git a/src/cloudai/configurator/env_params.py b/src/cloudai/configurator/env_params.py index a824d6205..660aa042e 100644 --- a/src/cloudai/configurator/env_params.py +++ b/src/cloudai/configurator/env_params.py @@ -32,7 +32,7 @@ import math import random from pathlib import Path -from typing import TYPE_CHECKING, Any, Dict, List, Literal, Optional, Protocol, runtime_checkable +from typing import TYPE_CHECKING, Annotated, Any, Dict, List, Literal, Optional, Protocol, Union, runtime_checkable from pydantic import BaseModel, ConfigDict, Field, model_validator from typing_extensions import Self @@ -102,6 +102,62 @@ def encode(self, value: Any, candidates: List[Any]) -> int: return candidates.index(value) +class LogEncoding(BaseModel): + """ + Log-scale encoding: observe the drawn value as its log-scaled float. + """ + + model_config = ConfigDict(extra="forbid") + + type: Literal["log"] = "log" + + def observation_descriptor(self, candidates: List[Any]) -> ObsLeafDescriptor: + return ObsLeafDescriptor(kind="box", dim=1) + + def encode(self, value: Any, candidates: List[Any]) -> float: + return float(math.log(float(value))) + + +AnyEncoding = Annotated[ + Union[CategoricalEncoding, LogEncoding], + Field(discriminator="type") +] + + +def _infer_encoding(candidates: List[Any]) -> AnyEncoding: + """Infer the appropriate encoding strategy for a list of candidate values.""" + if not candidates: + return CategoricalEncoding() + + if all(isinstance(c, str) for c in candidates): + return CategoricalEncoding() + + if len(candidates) < 3: + return CategoricalEncoding() + + if not all(isinstance(c, (int, float)) and c > 0 for c in candidates): + return CategoricalEncoding() + + try: + sorted_c = sorted(float(c) for c in candidates) + except (ValueError, TypeError): + return CategoricalEncoding() + + # Check perfectly uniform diffs (arithmetic) -> not log + diffs = [sorted_c[i] - sorted_c[i-1] for i in range(1, len(sorted_c))] + avg_diff = sum(diffs) / len(diffs) + if avg_diff > 1e-9 and all(abs(d - avg_diff) < 1e-5 for d in diffs): + return CategoricalEncoding() + + # Check constant ratio within tolerance (geometric series) + ratios = [sorted_c[i] / sorted_c[i-1] for i in range(1, len(sorted_c))] + avg_ratio = sum(ratios) / len(ratios) + if avg_ratio > 1.0 + 1e-9 and all(abs(r - avg_ratio) < 1e-5 for r in ratios): + return LogEncoding() + + return CategoricalEncoding() + + class EnvParamSpec(BaseModel): """ Annotation marking one cmd_args field as env-sampled. @@ -121,9 +177,9 @@ class EnvParamSpec(BaseModel): default=None, description="Optional probability weights aligned with the cmd_args candidate list; uniform if omitted.", ) - encoding: CategoricalEncoding = Field( - default_factory=CategoricalEncoding, - description="How the drawn value is encoded as an observation leaf (categorical index into the candidates).", + encoding: Optional[AnyEncoding] = Field( + default=None, + description="How the drawn value is encoded as an observation leaf. If omitted, inferred from candidates.", ) @model_validator(mode="after") @@ -214,7 +270,10 @@ def from_test(cls, test: "TestDefinition") -> Optional["EnvParams"]: value = getattr(test.cmd_args, name, None) if not isinstance(value, list): continue - params[name] = EnvParam(candidates=value, weights=spec.weights, encoding=spec.encoding) + encoding = spec.encoding + if encoding is None: + encoding = _infer_encoding(value) + params[name] = EnvParam(candidates=value, weights=spec.weights, encoding=encoding) if not params: return None seed = int((test.agent_config or {}).get("random_seed", 0)) diff --git a/tests/test_env_params.py b/tests/test_env_params.py index 9eb11d2ac..1a14f3446 100644 --- a/tests/test_env_params.py +++ b/tests/test_env_params.py @@ -44,7 +44,9 @@ EnvParam, EnvParams, EnvParamSpec, + LogEncoding, ObsLeafDescriptor, + _infer_encoding, write_env_params, ) from cloudai.core import TestRun @@ -454,3 +456,56 @@ def encode(self, value: Any, candidates: List[Any]) -> list: assert knob.observation_descriptor() == ObsLeafDescriptor(kind="box", dim=1) assert knob.encode(20) == [20.0] + + +# --- Encoding Inference: _infer_encoding --- + + +def test_infer_encoding_strings_categorical() -> None: + assert isinstance(_infer_encoding(["a", "b", "c"]), CategoricalEncoding) + + +def test_infer_encoding_geometric_series_log() -> None: + assert isinstance(_infer_encoding([1, 10, 100]), LogEncoding) + assert isinstance(_infer_encoding([0.01, 0.1, 1.0]), LogEncoding) + assert isinstance(_infer_encoding([2, 4, 8, 16]), LogEncoding) + + +def test_infer_encoding_arithmetic_series_categorical() -> None: + assert isinstance(_infer_encoding([1, 2, 3]), CategoricalEncoding) + assert isinstance(_infer_encoding([10, 20, 30, 40]), CategoricalEncoding) + + +def test_infer_encoding_ambiguous_two_point_categorical() -> None: + """Length < 3 cannot confidently be inferred as log.""" + assert isinstance(_infer_encoding([0.0, 0.001]), CategoricalEncoding) + assert isinstance(_infer_encoding([1, 10]), CategoricalEncoding) + + +def test_infer_encoding_zero_or_negative_categorical() -> None: + """Log scale requires strictly positive candidate values.""" + assert isinstance(_infer_encoding([0, 1, 10]), CategoricalEncoding) + assert isinstance(_infer_encoding([-1, 1, 10]), CategoricalEncoding) + + +def test_infer_encoding_unordered_candidates() -> None: + """The heuristic should work even if the candidates are not pre-sorted.""" + assert isinstance(_infer_encoding([100, 1, 10]), LogEncoding) + + +def test_env_params_from_test_uses_inferred_encoding() -> None: + """If encoding is not explicitly provided, it infers from candidates.""" + tdef = _tdef({"ball_speed": EnvParamSpec()}, ball_speed=[1, 10, 100]) + env_params = EnvParams.from_test(tdef) + assert env_params is not None + assert isinstance(env_params.params["ball_speed"].encoding, LogEncoding) + + +def test_env_params_from_test_explicit_override_wins() -> None: + """An explicit encoding in TOML overrides inference (even for 2-point lists or strings).""" + spec = EnvParamSpec.model_validate({"encoding": {"type": "log"}}) + tdef = _tdef({"ball_speed": spec}, ball_speed=[0.1, 0.5]) + env_params = EnvParams.from_test(tdef) + assert env_params is not None + assert isinstance(env_params.params["ball_speed"].encoding, LogEncoding) + From 6d4e2588b9cfe2ca6051a588613277c8d0492709 Mon Sep 17 00:00:00 2001 From: KhushviB Date: Mon, 13 Jul 2026 21:15:43 +0530 Subject: [PATCH 2/2] fix: validate explicit log encoding candidates in EnvParams --- src/cloudai/configurator/env_params.py | 15 ++++++++++--- tests/test_env_params.py | 31 +++++++++++++++++++++++--- 2 files changed, 40 insertions(+), 6 deletions(-) diff --git a/src/cloudai/configurator/env_params.py b/src/cloudai/configurator/env_params.py index 660aa042e..07fbf23f1 100644 --- a/src/cloudai/configurator/env_params.py +++ b/src/cloudai/configurator/env_params.py @@ -135,6 +135,9 @@ def _infer_encoding(candidates: List[Any]) -> AnyEncoding: if len(candidates) < 3: return CategoricalEncoding() + if any(isinstance(c, bool) for c in candidates): + return CategoricalEncoding() + if not all(isinstance(c, (int, float)) and c > 0 for c in candidates): return CategoricalEncoding() @@ -146,13 +149,13 @@ def _infer_encoding(candidates: List[Any]) -> AnyEncoding: # Check perfectly uniform diffs (arithmetic) -> not log diffs = [sorted_c[i] - sorted_c[i-1] for i in range(1, len(sorted_c))] avg_diff = sum(diffs) / len(diffs) - if avg_diff > 1e-9 and all(abs(d - avg_diff) < 1e-5 for d in diffs): + if avg_diff > 0 and all(math.isclose(d, avg_diff, rel_tol=1e-5) for d in diffs): return CategoricalEncoding() # Check constant ratio within tolerance (geometric series) ratios = [sorted_c[i] / sorted_c[i-1] for i in range(1, len(sorted_c))] avg_ratio = sum(ratios) / len(ratios) - if avg_ratio > 1.0 + 1e-9 and all(abs(r - avg_ratio) < 1e-5 for r in ratios): + if avg_ratio > 1.0 + 1e-9 and all(math.isclose(r, avg_ratio, rel_tol=1e-5) for r in ratios): return LogEncoding() return CategoricalEncoding() @@ -166,7 +169,7 @@ class EnvParamSpec(BaseModel): ``cmd_args.`` as a plain list. ``weights`` (optional) are positional, aligned 1:1 with that candidate list; omit for uniform sampling. ``encoding`` (optional) selects how the drawn value is exposed to the policy as an - observation leaf, defaulting to a categorical index. The length match against + observation leaf, defaulting to None and is inferred from candidates. The length match against the candidate list is a cross-field check enforced by ``TestDefinition`` (which can see ``cmd_args``); here we validate only the weights' intrinsic shape. """ @@ -273,6 +276,12 @@ def from_test(cls, test: "TestDefinition") -> Optional["EnvParams"]: encoding = spec.encoding if encoding is None: encoding = _infer_encoding(value) + elif isinstance(encoding, LogEncoding): + for c in value: + if not isinstance(c, (int, float)) or isinstance(c, bool): + raise TypeError(f"LogEncoding for '{name}' requires numeric candidates, got {type(c).__name__}") + if not math.isfinite(c) or c <= 0: + raise ValueError(f"LogEncoding for '{name}' requires strictly positive, finite candidates, got {c}") params[name] = EnvParam(candidates=value, weights=spec.weights, encoding=encoding) if not params: return None diff --git a/tests/test_env_params.py b/tests/test_env_params.py index 1a14f3446..df92bd5a6 100644 --- a/tests/test_env_params.py +++ b/tests/test_env_params.py @@ -412,9 +412,9 @@ def test_obs_leaf_descriptor_rejects_bad_dim_and_extra_fields() -> None: # --- Encoding: pluggable strategy mapping a drawn value to an observation leaf --- -def test_env_param_spec_defaults_to_categorical_encoding() -> None: - """An unspecified encoding defaults to categorical (back-compat with bare ``EnvParamSpec()``).""" - assert EnvParamSpec().encoding == CategoricalEncoding() +def test_env_param_spec_defaults_to_none_encoding() -> None: + """An unspecified encoding defaults to None so that it can be inferred later.""" + assert EnvParamSpec().encoding is None def test_env_param_spec_parses_encoding_from_config() -> None: @@ -488,6 +488,11 @@ def test_infer_encoding_zero_or_negative_categorical() -> None: assert isinstance(_infer_encoding([-1, 1, 10]), CategoricalEncoding) +def test_infer_encoding_boolean_containing_list_categorical() -> None: + """A candidate list containing booleans should fallback to categorical encoding.""" + assert isinstance(_infer_encoding([True, 2, 4]), CategoricalEncoding) + + def test_infer_encoding_unordered_candidates() -> None: """The heuristic should work even if the candidates are not pre-sorted.""" assert isinstance(_infer_encoding([100, 1, 10]), LogEncoding) @@ -509,3 +514,23 @@ def test_env_params_from_test_explicit_override_wins() -> None: assert env_params is not None assert isinstance(env_params.params["ball_speed"].encoding, LogEncoding) + +def test_env_params_from_test_explicit_log_override_validation() -> None: + """An explicit log encoding rejects zero, negative, non-numeric, or non-finite candidates.""" + spec = EnvParamSpec.model_validate({"encoding": {"type": "log"}}) + + with pytest.raises(ValueError, match="strictly positive"): + EnvParams.from_test(_tdef({"ball_speed": spec}, ball_speed=[0, 1])) + + with pytest.raises(ValueError, match="strictly positive"): + EnvParams.from_test(_tdef({"ball_speed": spec}, ball_speed=[-1, 1])) + + with pytest.raises(ValueError, match="strictly positive"): + EnvParams.from_test(_tdef({"ball_speed": spec}, ball_speed=[1.0, float("inf")])) + + with pytest.raises(TypeError, match="numeric"): + EnvParams.from_test(_tdef({"ball_speed": spec}, ball_speed=["a", "b"])) + + with pytest.raises(TypeError, match="numeric"): + EnvParams.from_test(_tdef({"ball_speed": spec}, ball_speed=[True, False])) +