python-math #13: refactor(math): move impl-flag resolver to polismath/utils/env_flags.py

delphi/polismath/pca_kmeans_rep/pca.py

@@ -6,11 +6,11 @@
 """
 
 import logging
-import os
 import numpy as np
 import pandas as pd
 from typing import Dict, List, Optional, Sequence, Tuple, Union, Any
 
+from polismath.utils.env_flags import resolve_impl_flag
 from polismath.utils.general import AGREE
 
 logger = logging.getLogger(__name__)
@@ -20,42 +20,16 @@
 # Implementation switch: legacy/Clojure-parity vs improved
 # =============================================================================
 #
-# Pattern for legacy-vs-improved switches (reuse this idiom for future ones,
-# e.g. a k-means solver switch): a module-level env var name + default +
-# allowed values, resolved by `_resolve_impl_flag` AT CALL TIME (never at
-# import time), so tests and operators can flip the env var without
-# re-importing. Unknown values fall back to the default with a warning
-# (defensive: a typo in a deployment env must not crash the math worker).
+# The switch idiom (env var + default + allowed values, resolved AT CALL TIME
+# by the shared `resolve_impl_flag`) is documented in
+# polismath/utils/env_flags.py, where the resolver lives.
 
 PCA_IMPL_ENV_VAR = 'POLISMATH_PCA_IMPL'
 PCA_IMPL_POWERIT = 'powerit'   # legacy/Clojure-parity solver (default)
 PCA_IMPL_SKLEARN = 'sklearn'   # improved solver (exact SVD)
 PCA_IMPL_DEFAULT = PCA_IMPL_POWERIT
 PCA_IMPL_CHOICES = (PCA_IMPL_POWERIT, PCA_IMPL_SKLEARN)
 
-
-def _resolve_impl_flag(env_var: str, default: str, choices: Sequence[str]) -> str:
-    """
-    Resolve a legacy-vs-improved implementation switch from the environment.
-
-    Args:
-        env_var: Environment variable name to read (at call time).
-        default: Value to use when the variable is unset or invalid.
-        choices: Allowed values (lowercase).
-
-    Returns:
-        One of `choices`.
-    """
-    raw = os.environ.get(env_var)
-    if raw is None:
-        return default
-    value = raw.strip().lower()
-    if value not in choices:
-        logger.warning("%s=%r is not one of %s; falling back to %r",
-                       env_var, raw, tuple(choices), default)
-        return default
-    return value
-
 # =============================================================================
 # Clojure-parity power-iteration PCA
 # =============================================================================
@@ -338,12 +312,12 @@ def pca_project_dataframe(df: pd.DataFrame,
     # "Determinism verification" entry (2026-07-04/05) in
     # docs/CLJ-PARITY-FIXES-JOURNAL.md.
 
-    # Solver switch (read at call time — see _resolve_impl_flag):
+    # Solver switch (read at call time — see utils.env_flags.resolve_impl_flag):
     #   POLISMATH_PCA_IMPL=powerit  (default) legacy/Clojure-parity power iteration
     #   POLISMATH_PCA_IMPL=sklearn  improved exact-SVD path
     # The imputation above and sparsity scaling below are IDENTICAL for both;
     # only the eigen-solver differs.
-    impl = _resolve_impl_flag(PCA_IMPL_ENV_VAR, PCA_IMPL_DEFAULT, PCA_IMPL_CHOICES)
+    impl = resolve_impl_flag(PCA_IMPL_ENV_VAR, PCA_IMPL_DEFAULT, PCA_IMPL_CHOICES)
 
     # Warm-start parity (PR-B): power iteration is the ONLY solver that can be
     # seeded with the previous tick's components (Clojure :start-vectors,

delphi/polismath/utils/engine_mode.py

@@ -16,17 +16,16 @@
   - 'improved'       (default): today's cold-recompute behavior, byte-for-byte.
   - 'clojure-legacy'         : threads the warm-start state described above.
 
-The flag is resolved AT CALL TIME (never cached at import), reusing the exact
-idiom of `pca._resolve_impl_flag` (pca.py:37-57): unknown values fall back to
-the default with a warning so a typo in a deployment env cannot crash the math
-worker. This lives in a shared spot (polismath.utils) because the mode
-cross-cuts both PCA (conversation._compute_pca) and clustering
-(conversation._compute_clusters).
+The flag is resolved AT CALL TIME (never cached at import) by the shared
+`utils.env_flags.resolve_impl_flag`: unknown values fall back to the default
+with a warning so a typo in a deployment env cannot crash the math worker.
+This lives in a shared spot (polismath.utils) because the mode cross-cuts both
+PCA (conversation._compute_pca) and clustering (conversation._compute_clusters).
 """
 
 from typing import Sequence
 
-from polismath.pca_kmeans_rep.pca import _resolve_impl_flag
+from polismath.utils.env_flags import resolve_impl_flag
 
 ENGINE_MODE_ENV_VAR = 'POLISMATH_ENGINE_MODE'
 ENGINE_MODE_LEGACY = 'clojure-legacy'   # warm-start parity with Clojure
@@ -39,12 +38,12 @@ def resolve_engine_mode() -> str:
     """
     Resolve `POLISMATH_ENGINE_MODE` from the environment, at call time.
 
-    Reuses `pca._resolve_impl_flag` (pca.py:37-57) so the resolution rules
+    Reuses `utils.env_flags.resolve_impl_flag` so the resolution rules
     (strip + lowercase, unknown -> default with a warning) are identical to
     the PCA-solver switch.
 
     Returns:
         Either 'improved' (default) or 'clojure-legacy'.
     """
-    return _resolve_impl_flag(
+    return resolve_impl_flag(
         ENGINE_MODE_ENV_VAR, ENGINE_MODE_DEFAULT, ENGINE_MODE_CHOICES)

delphi/polismath/utils/env_flags.py

@@ -0,0 +1,45 @@
+"""
+Shared resolver for legacy-vs-improved implementation switches.
+
+Pattern for env-var implementation switches (POLISMATH_PCA_IMPL,
+POLISMATH_ENGINE_MODE, and future ones like a k-means solver switch): a
+module-level env var name + default + allowed values, resolved by
+`resolve_impl_flag` AT CALL TIME (never at import time), so tests and
+operators can flip the env var without re-importing. Unknown values fall back
+to the default with a warning (defensive: a typo in a deployment env must not
+crash the math worker).
+
+This lives in polismath.utils (not pca.py, where it originated) so that
+lightweight consumers — e.g. `utils.engine_mode`, read on every conv-update
+tick — do not drag in the numpy/pandas pca import chain, and resolution
+warnings are logged under this module's logger rather than pca's.
+"""
+
+import logging
+import os
+from typing import Sequence
+
+logger = logging.getLogger(__name__)
+
+
+def resolve_impl_flag(env_var: str, default: str, choices: Sequence[str]) -> str:
+    """
+    Resolve a legacy-vs-improved implementation switch from the environment.
+
+    Args:
+        env_var: Environment variable name to read (at call time).
+        default: Value to use when the variable is unset or invalid.
+        choices: Allowed values (lowercase).
+
+    Returns:
+        One of `choices`.
+    """
+    raw = os.environ.get(env_var)
+    if raw is None:
+        return default
+    value = raw.strip().lower()
+    if value not in choices:
+        logger.warning("%s=%r is not one of %s; falling back to %r",
+                       env_var, raw, tuple(choices), default)
+        return default
+    return value

delphi/tests/test_env_flags.py

@@ -0,0 +1,85 @@
+"""
+Tests for polismath.utils.env_flags — the shared legacy-vs-improved
+implementation-switch resolver.
+
+The resolver started life as `pca._resolve_impl_flag` (pca.py) and was imported
+from there by `utils.engine_mode`, which dragged the whole numpy/pandas pca
+import chain into anything that only wanted to read POLISMATH_ENGINE_MODE, and
+emitted resolution warnings under the pca logger. These tests pin the move to
+`polismath.utils.env_flags`: identical resolution rules, warnings under the
+env_flags logger, and a light `utils.engine_mode` import.
+"""
+
+import logging
+import subprocess
+import sys
+
+import pytest
+
+from polismath.utils.env_flags import resolve_impl_flag
+
+
+class TestResolveImplFlag:
+    """Resolution rules (identical to the original pca._resolve_impl_flag)."""
+
+    ENV = 'POLISMATH_TEST_FLAG'
+    CHOICES = ('legacy', 'improved')
+
+    def test_unset_returns_default(self, monkeypatch):
+        monkeypatch.delenv(self.ENV, raising=False)
+        assert resolve_impl_flag(self.ENV, 'legacy', self.CHOICES) == 'legacy'
+
+    def test_valid_value_returned(self, monkeypatch):
+        monkeypatch.setenv(self.ENV, 'improved')
+        assert resolve_impl_flag(self.ENV, 'legacy', self.CHOICES) == 'improved'
+
+    def test_value_stripped_and_lowercased(self, monkeypatch):
+        monkeypatch.setenv(self.ENV, '  IMPROVED ')
+        assert resolve_impl_flag(self.ENV, 'legacy', self.CHOICES) == 'improved'
+
+    def test_invalid_value_falls_back_with_warning(self, monkeypatch, caplog):
+        monkeypatch.setenv(self.ENV, 'bogus')
+        with caplog.at_level(logging.WARNING, logger='polismath.utils.env_flags'):
+            assert resolve_impl_flag(self.ENV, 'legacy', self.CHOICES) == 'legacy'
+        records = [r for r in caplog.records
+                   if r.name == 'polismath.utils.env_flags']
+        assert len(records) == 1
+        assert 'POLISMATH_TEST_FLAG' in records[0].getMessage()
+
+    def test_read_at_call_time(self, monkeypatch):
+        monkeypatch.setenv(self.ENV, 'improved')
+        assert resolve_impl_flag(self.ENV, 'legacy', self.CHOICES) == 'improved'
+        monkeypatch.setenv(self.ENV, 'legacy')
+        assert resolve_impl_flag(self.ENV, 'legacy', self.CHOICES) == 'legacy'
+
+
+class TestSharedResolver:
+    """Both switch modules resolve through the ONE shared function."""
+
+    def test_pca_uses_shared_resolver(self):
+        from polismath.pca_kmeans_rep import pca
+        from polismath.utils import env_flags
+        assert pca.resolve_impl_flag is env_flags.resolve_impl_flag
+
+    def test_engine_mode_uses_shared_resolver(self):
+        from polismath.utils import engine_mode, env_flags
+        assert engine_mode.resolve_impl_flag is env_flags.resolve_impl_flag
+
+    def test_engine_mode_import_does_not_load_pca(self):
+        # The point of the move: reading POLISMATH_ENGINE_MODE must not drag
+        # the numpy/pandas pca import chain. Fresh interpreter so this
+        # process's already-imported modules can't mask a regression.
+        code = (
+            "import sys; import polismath.utils.engine_mode; "
+            "sys.exit(1 if 'polismath.pca_kmeans_rep.pca' in sys.modules else 0)"
+        )
+        proc = subprocess.run([sys.executable, '-c', code],
+                              capture_output=True, text=True)
+        assert proc.returncode == 0, (
+            "importing polismath.utils.engine_mode pulled in "
+            "polismath.pca_kmeans_rep.pca:\n" + proc.stderr
+        )
+
+
+if __name__ == '__main__':
+    pytest.main([__file__, '-v'])