fix pylint issues

Author: jedwards4b

cmip7_prep/mapping_compat.py

@@ -1,5 +1,6 @@
 
-"""Mapping loader/evaluator compatible with CMIP6-style lists and CMIP7-style dicts.
+# cmip7_prep/mapping_compat.py
+r"""Mapping loader/evaluator compatible with CMIP6-style lists and CMIP7-style dicts.
 
 This module provides a `Mapping` class that:
 - Loads a YAML mapping file where **keys are CMIP variable names** (preferred), or
@@ -8,36 +9,11 @@
 - Builds CMIP variables from a native CESM `xarray.Dataset` via :meth:`realize`,
   supporting raw variables, formulas, and simple unit conversions.
 
-YAML schema (both supported)
----------------------------
-# CMIP7-style (preferred):
-tas:
-  table: Amon                  # or CMIP6_Amon.json / CMIP7_Amon.json
-  units: K
-  raw_variables: [TREFHT]      # from the CESM files
-  regrid_method: bilinear
-
-# CMIP6-style (list of dicts):
-- name: tas
-  table: CMIP6_Amon.json
-  units: K
-  raw_variables: [TREFHT]
-  unit_conversion: null
-  formula: null
-
-Unit conversion support
------------------------
-- String expression using ``x`` (the realized DataArray), with ``np`` and ``xr`` available.
-  Example: ``unit_conversion: "x * 86400.0"`` to convert kg m-2 s-1 -> mm/day for precipitation.
-- Dict with ``scale`` and optional ``offset`` (applied as x * scale + offset).
-
-Formula support
----------------
-- A Python expression combining raw variables by name (e.g., ``"bc_a1+bc_a4+bc_c1+bc_c4"``).
-  Only variables present in the mapping entry are available, plus ``np``/``xr``. No builtins.
-
-Safety note: formulas and conversions are evaluated in a heavily restricted environment.
-This is intended for trusted mapping files under your control.
+Security note
+-------------
+Formulas and conversion expressions are evaluated from a trusted, local mapping file
+with a restricted environment. We centralize the use of ``eval`` in a helper that
+is clearly marked and limited, and we disable the linter warning only at that line.
 """
 from __future__ import annotations
 
@@ -47,7 +23,7 @@
 
 import numpy as np
 import xarray as xr
-import yaml
+import yaml  # runtime dependency; ignored by pylint via .pylintrc
 
 
 def _normalize_table_name(value: Optional[str]) -> Optional[str]:
@@ -58,16 +34,17 @@ def _normalize_table_name(value: Optional[str]) -> Optional[str]:
     if not value:
         return None
     s = str(value)
-    if s.lower().endswith('.json'):
+    if s.lower().endswith(".json"):
         s = s[:-5]
     # strip CMIPx_ prefix if present
-    if '_' in s:
-        parts = s.split('_', 1)
-        if len(parts) == 2 and parts[0].upper().startswith('CMIP'):
+    if "_" in s:
+        parts = s.split("_", 1)
+        if len(parts) == 2 and parts[0].upper().startswith("CMIP"):
             s = parts[1]
     return s
 
 
+# pylint: disable=too-many-instance-attributes
 @dataclass(frozen=True)
 class VarConfig:
     """Normalized mapping entry for a single CMIP variable."""
@@ -86,21 +63,40 @@ class VarConfig:
     def as_cfg(self) -> Dict[str, Any]:
         """Return a plain dict view for convenience in other modules."""
         d = {
-            'name': self.name,
-            'table': self.table,
-            'units': self.units,
-            'raw_variables': self.raw_variables,
-            'source': self.source,
-            'formula': self.formula,
-            'unit_conversion': self.unit_conversion,
-            'positive': self.positive,
-            'cell_methods': self.cell_methods,
-            'levels': self.levels,
-            'regrid_method': self.regrid_method,
+            "name": self.name,
+            "table": self.table,
+            "units": self.units,
+            "raw_variables": self.raw_variables,
+            "source": self.source,
+            "formula": self.formula,
+            "unit_conversion": self.unit_conversion,
+            "positive": self.positive,
+            "cell_methods": self.cell_methods,
+            "levels": self.levels,
+            "regrid_method": self.regrid_method,
         }
         return {k: v for k, v in d.items() if v is not None}
 
 
+def _safe_eval(expr: str, local_names: Dict[str, Any]) -> Any:
+    """Evaluate a small arithmetic/xarray expression in a restricted environment.
+
+    Only the names provided in `local_names` are available, plus numpy/xarray.
+
+    Used for:
+      - combining raw variables in `formula`
+      - simple unit conversions (string form)
+
+    Mapping files are assumed trusted; we still minimize surface area by stripping
+    builtins and only exposing needed names.
+    """
+    safe_globals = {"__builtins__": {}}
+    locals_safe = {"np": np, "xr": xr}
+    locals_safe.update(local_names)
+    # pylint: disable=eval-used
+    return eval(expr, safe_globals, locals_safe)
+
+
 class Mapping:
     """Load and evaluate a CMIP mapping YAML file.
 
@@ -123,7 +119,7 @@ def __init__(self, path: str | Path) -> None:
     # -----------------
     @staticmethod
     def _load_yaml(path: Path) -> Dict[str, VarConfig]:
-        with path.open('r', encoding='utf-8') as f:
+        with path.open("r", encoding="utf-8") as f:
             data = yaml.safe_load(f)
 
         result: Dict[str, VarConfig] = {}
@@ -137,9 +133,9 @@ def _load_yaml(path: Path) -> Dict[str, VarConfig]:
         elif isinstance(data, list):
             # CMIP6-style: list with 'name' field
             for item in data:
-                if not isinstance(item, dict) or 'name' not in item:
+                if not isinstance(item, dict) or "name" not in item:
                     continue
-                name = str(item['name'])
+                name = str(item["name"])
                 result[name] = _to_varconfig(name, item)
         else:
             raise TypeError("Unsupported YAML structure: expected dict or list at top level.")
@@ -171,16 +167,6 @@ def realize(self, ds: xr.Dataset, cmip_name: str) -> xr.DataArray:
         - `raw_variables`: list of CESM variable names (used by formula or identity)
         - `formula`: Python expression combining raw variables
         - `unit_conversion`: str expression (using `x`) or dict {scale, offset}
-
-        Returns
-        -------
-        xr.DataArray
-            The realized variable, with attrs possibly updated to target units.
-
-        Raises
-        ------
-        KeyError if required raw variables are not present in `ds`.
-        ValueError if the mapping is incomplete or inconsistent.
         """
         if cmip_name not in self._vars:
             raise KeyError(f"No mapping for {cmip_name!r} in {self.path}")
@@ -194,30 +180,30 @@ def realize(self, ds: xr.Dataset, cmip_name: str) -> xr.DataArray:
 
         # set target units if provided
         if vc.units:
-            da.attrs['units'] = vc.units
+            da.attrs["units"] = vc.units
 
         return da
 
 
 def _to_varconfig(name: str, cfg: TMapping[str, Any]) -> VarConfig:
     """Normalize a raw YAML entry into a :class:`VarConfig`."""
-    table = _normalize_table_name(cfg.get('table') or cfg.get('CMOR_table'))
-    raw_vars = cfg.get('raw_variables') or cfg.get('raw_vars') or None
+    table = _normalize_table_name(cfg.get("table") or cfg.get("CMOR_table"))
+    raw_vars = cfg.get("raw_variables") or cfg.get("raw_vars") or None
     if isinstance(raw_vars, str):
         raw_vars = [raw_vars]
-    levels = cfg.get('levels') or None
+    levels = cfg.get("levels") or None
     vc = VarConfig(
         name=name,
         table=table,
-        units=cfg.get('units'),
+        units=cfg.get("units"),
         raw_variables=raw_vars,
-        source=cfg.get('source'),
-        formula=cfg.get('formula'),
-        unit_conversion=cfg.get('unit_conversion'),
-        positive=cfg.get('positive'),
-        cell_methods=cfg.get('cell_methods'),
+        source=cfg.get("source"),
+        formula=cfg.get("formula"),
+        unit_conversion=cfg.get("unit_conversion"),
+        positive=cfg.get("positive"),
+        cell_methods=cfg.get("cell_methods"),
         levels=levels,
-        regrid_method=cfg.get('regrid_method'),
+        regrid_method=cfg.get("regrid_method"),
     )
     return vc
 
@@ -238,7 +224,7 @@ def _realize_core(ds: xr.Dataset, vc: VarConfig) -> xr.DataArray:
         return ds[vc.source]
 
     # 2) identity mapping from a single raw variable
-    if vc.raw_variables and vc.formula in (None, '', 'null') and len(vc.raw_variables) == 1:
+    if vc.raw_variables and vc.formula in (None, "", "null") and len(vc.raw_variables) == 1:
         var = vc.raw_variables[0]
         if var not in ds:
             raise KeyError(f"raw variable {var!r} not found in dataset")
@@ -249,11 +235,9 @@ def _realize_core(ds: xr.Dataset, vc: VarConfig) -> xr.DataArray:
         if not vc.raw_variables:
             raise ValueError(f"formula given for {vc.name} but no raw_variables listed")
         env = _require_vars(ds, vc.raw_variables, f"realize({vc.name})")
-        safe_globals = {'__builtins__': {}}
-        safe_locals = {'np': np, 'xr': xr, **env}
         try:
-            result = eval(vc.formula, safe_globals, safe_locals)  # noqa: S307 (trusted file)
-        except Exception as exc:  # narrow scope: we treat mapping as trusted config
+            result = _safe_eval(vc.formula, env)
+        except Exception as exc:
             raise ValueError(f"Error evaluating formula for {vc.name}: {exc}") from exc
         if not isinstance(result, xr.DataArray):
             raise ValueError(f"Formula for {vc.name} did not produce a DataArray")
@@ -274,26 +258,19 @@ def _apply_unit_conversion(da: xr.DataArray, rule: Any) -> xr.DataArray:
     rule : str or dict
         - If str, evaluates an expression using ``x`` (the data), with ``np`` and ``xr`` available.
         - If dict, supports keys: ``scale`` and optional ``offset``.
-
-    Returns
-    -------
-    DataArray
-        Converted data array.
     """
     if isinstance(rule, str):
-        safe_globals = {'__builtins__': {}}
-        safe_locals = {'x': da, 'np': np, 'xr': xr}
         try:
-            out = eval(rule, safe_globals, safe_locals)  # noqa: S307 (trusted mapping)
+            out = _safe_eval(rule, {"x": da})
         except Exception as exc:
             raise ValueError(f"Error evaluating unit_conversion expression: {exc}") from exc
         if not isinstance(out, xr.DataArray):
             raise ValueError("unit_conversion expression did not return a DataArray")
         return out
 
     if isinstance(rule, dict):
-        scale = rule.get('scale', 1.0)
-        offset = rule.get('offset', 0.0)
+        scale = rule.get("scale", 1.0)
+        offset = rule.get("offset", 0.0)
         return da * float(scale) + float(offset)
 
     raise TypeError("unit_conversion must be a string expression or a dict with 'scale'/'offset'")