feat: Infer DSE parameter encoding from candidate values at parse time

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.<name>`` 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))

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]))
+