diff --git a/src/cloudai/configurator/env_params.py b/src/cloudai/configurator/env_params.py index a824d6205..07fbf23f1 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,65 @@ 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 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() + + 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 > 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(math.isclose(r, avg_ratio, rel_tol=1e-5) for r in ratios): + return LogEncoding() + + return CategoricalEncoding() + + class EnvParamSpec(BaseModel): """ Annotation marking one cmd_args field as env-sampled. @@ -110,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. """ @@ -121,9 +180,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 +273,16 @@ 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) + 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 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..df92bd5a6 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 @@ -410,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: @@ -454,3 +456,81 @@ 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_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) + + +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) + + +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])) +