Consolidate duplicated input-coercion helpers into _helpers

src/nns/_helpers.py

@@ -24,6 +24,73 @@ def _warn_unsupported(**was_set: bool) -> None:
         )
 
 
+def _as_matrix(x: NDArray[np.float64], name: str) -> NDArray[np.float64]:
+    """Return x as a non-empty, all-finite 2D float64 matrix."""
+    values = np.asarray(x, dtype=np.float64)
+    if values.ndim != 2:
+        raise ValueError(f"{name} must be 2D.")
+    if values.shape[0] == 0 or values.shape[1] == 0:
+        raise ValueError(f"{name} must be non-empty.")
+    if not np.all(np.isfinite(values)):
+        raise ValueError(f"{name} must contain only finite values.")
+    return values
+
+
+def _as_vector_or_matrix(x: NDArray[np.float64], name: str) -> NDArray[np.float64]:
+    """Return x as a non-empty, all-finite 2D float64 matrix, promoting 1D to a column."""
+    values = np.asarray(x, dtype=np.float64)
+    if values.ndim == 1:
+        values = values.reshape(-1, 1)
+    if values.ndim != 2 or values.shape[0] == 0 or values.shape[1] == 0:
+        raise ValueError(f"{name} must be a non-empty numeric vector or matrix.")
+    if not np.all(np.isfinite(values)):
+        raise ValueError(f"{name} must contain only finite values.")
+    return values
+
+
+def _as_point_matrix(x: NDArray[np.float64], n_cols: int) -> NDArray[np.float64]:
+    """Return evaluation points as an all-finite matrix with the training column count."""
+    values = np.asarray(x, dtype=np.float64)
+    if values.ndim == 1:
+        if n_cols == 1:
+            values = values.reshape(-1, 1)
+        else:
+            values = values.reshape(1, -1)
+    if values.ndim != 2 or values.shape[1] != n_cols:
+        raise ValueError("ivs_test must have the same column count as ivs_train.")
+    if not np.all(np.isfinite(values)):
+        raise ValueError("ivs_test must contain only finite values.")
+    return values
+
+
+def _as_flat_vector(x: NDArray[np.float64], name: str) -> NDArray[np.float64]:
+    """Return x flattened to a non-empty, all-finite 1D float64 vector."""
+    values = np.asarray(x, dtype=np.float64).reshape(-1)
+    if values.size == 0:
+        raise ValueError(f"{name} must be non-empty.")
+    if not np.all(np.isfinite(values)):
+        raise ValueError(f"{name} must contain only finite values.")
+    return values
+
+
+def _as_pair(
+    x: NDArray[np.float64],
+    y: NDArray[np.float64],
+) -> tuple[NDArray[np.float64], NDArray[np.float64]]:
+    """Return x and y as equal-length, non-empty, all-finite 1D float64 vectors."""
+    x_values = np.asarray(x, dtype=np.float64)
+    y_values = np.asarray(y, dtype=np.float64)
+    if x_values.ndim != 1 or y_values.ndim != 1:
+        raise ValueError("x and y must be 1D.")
+    if x_values.size == 0:
+        raise ValueError("x and y must be non-empty.")
+    if x_values.size != y_values.size:
+        raise ValueError("x and y must have the same length.")
+    if not np.all(np.isfinite(x_values)) or not np.all(np.isfinite(y_values)):
+        raise ValueError("x and y must contain only finite values.")
+    return x_values, y_values
+
+
 def _fast_lm(x: NDArray[np.float64], y: NDArray[np.float64]) -> tuple[float, float]:
     """Return intercept and slope matching R's fast_lm helper."""
     x_values = np.asarray(x, dtype=np.float64)

src/nns/boost.py

@@ -9,6 +9,7 @@
 import numpy as np
 from numpy.typing import NDArray
 
+from nns._helpers import _as_flat_vector, _as_point_matrix, _as_vector_or_matrix
 from nns.categorical import _balance_class_training, _dense_factor_codes, encode_factor_codes
 from nns.dependence import _gravity
 from nns.regression import (
@@ -80,7 +81,7 @@ def nns_boost(
     ):
         raise ValueError("string/object predictor values require explicit factor_levels.")
 
-    x_train = _as_matrix(x_input, "ivs_train")
+    x_train = _as_vector_or_matrix(x_input, "ivs_train")
     x_test = (
         x_train.copy() if x_test_input is None else _as_point_matrix(x_test_input, x_train.shape[1])
     )
@@ -97,7 +98,7 @@ def nns_boost(
         )
         class_codes = np.unique(y_train[np.isfinite(y_train)])
     else:
-        y_train = _as_vector(dv_train, "dv_train")
+        y_train = _as_flat_vector(dv_train, "dv_train")
         class_codes = np.empty(0, dtype=np.float64)
     if x_train.shape[0] != y_train.size:
         raise ValueError("ivs_train and dv_train must have the same row count.")
@@ -663,37 +664,3 @@ def _sse(predicted: NDArray[np.float64], actual: NDArray[np.float64]) -> float:
 
 def _accuracy(predicted: NDArray[np.float64], actual: NDArray[np.float64]) -> float:
     return float(np.mean(predicted == actual))
-
-
-def _as_matrix(x: NDArray[np.float64], name: str) -> NDArray[np.float64]:
-    values = np.asarray(x, dtype=np.float64)
-    if values.ndim == 1:
-        values = values.reshape(-1, 1)
-    if values.ndim != 2 or values.shape[0] == 0 or values.shape[1] == 0:
-        raise ValueError(f"{name} must be a non-empty numeric vector or matrix.")
-    if not np.all(np.isfinite(values)):
-        raise ValueError(f"{name} must contain only finite values.")
-    return values
-
-
-def _as_point_matrix(x: NDArray[np.float64], n_cols: int) -> NDArray[np.float64]:
-    values = np.asarray(x, dtype=np.float64)
-    if values.ndim == 1:
-        if n_cols == 1:
-            values = values.reshape(-1, 1)
-        else:
-            values = values.reshape(1, -1)
-    if values.ndim != 2 or values.shape[1] != n_cols:
-        raise ValueError("ivs_test must have the same column count as ivs_train.")
-    if not np.all(np.isfinite(values)):
-        raise ValueError("ivs_test must contain only finite values.")
-    return values
-
-
-def _as_vector(x: NDArray[np.float64], name: str) -> NDArray[np.float64]:
-    values = np.asarray(x, dtype=np.float64).reshape(-1)
-    if values.size == 0:
-        raise ValueError(f"{name} must be non-empty.")
-    if not np.all(np.isfinite(values)):
-        raise ValueError(f"{name} must contain only finite values.")
-    return values

src/nns/causation.py

@@ -5,9 +5,9 @@
 import numpy as np
 from numpy.typing import NDArray
 
+from nns._helpers import _as_matrix, _as_pair
 from nns.core import lpm_ratio, upm_ratio
 from nns.dependence import (
-    _as_pair,
     _copula_degree0_unsigned,
     _copula_signed,
     _directional_dep,
@@ -68,7 +68,7 @@ def causal_matrix(
     tau: int | str = 0,
 ) -> NDArray[np.float64]:
     """Return R's NNS.caus.matrix antisymmetric net-causation matrix."""
-    values = _as_matrix(x)
+    values = _as_matrix(x, "x")
     if tau != "ts":
         tau = _tau_value(tau)
     n_variables = values.shape[1]
@@ -175,14 +175,3 @@ def _cap_inf100(value: float, cap: float = 100.0) -> float:
     if abs(value) > cap:
         return math.copysign(cap, value)
     return value
-
-
-def _as_matrix(x: NDArray[np.float64]) -> NDArray[np.float64]:
-    values = np.asarray(x, dtype=np.float64)
-    if values.ndim != 2:
-        raise ValueError("x must be 2D.")
-    if values.shape[0] == 0 or values.shape[1] == 0:
-        raise ValueError("x must be non-empty.")
-    if not np.all(np.isfinite(values)):
-        raise ValueError("x must contain only finite values.")
-    return values

src/nns/dependence.py

@@ -6,6 +6,7 @@
 import numpy as np
 from numpy.typing import NDArray
 
+from nns._helpers import _as_pair
 from nns._native import native_fn
 from nns.co_moments import co_lpm, co_upm, d_lpm, d_upm
 
@@ -409,20 +410,3 @@ def _finite_or_zero(value: float) -> float:
 
 def _is_constant(values: NDArray[np.float64]) -> bool:
     return bool(np.all(values == values[0]))
-
-
-def _as_pair(
-    x: NDArray[np.float64],
-    y: NDArray[np.float64],
-) -> tuple[NDArray[np.float64], NDArray[np.float64]]:
-    x_values = np.asarray(x, dtype=np.float64)
-    y_values = np.asarray(y, dtype=np.float64)
-    if x_values.ndim != 1 or y_values.ndim != 1:
-        raise ValueError("x and y must be 1D.")
-    if x_values.size == 0:
-        raise ValueError("x and y must be non-empty.")
-    if x_values.size != y_values.size:
-        raise ValueError("x and y must have the same length.")
-    if not np.all(np.isfinite(x_values)) or not np.all(np.isfinite(y_values)):
-        raise ValueError("x and y must contain only finite values.")
-    return x_values, y_values

src/nns/distance.py

@@ -6,6 +6,8 @@
 import numpy as np
 from numpy.typing import NDArray
 
+from nns._helpers import _as_matrix
+
 KValue = int | Literal["all"]
 
 
@@ -231,14 +233,3 @@ def _as_vector(x: NDArray[np.float64]) -> NDArray[np.float64]:
     if not np.all(np.isfinite(values)):
         raise ValueError("dist_estimate must contain only finite values.")
     return values
-
-
-def _as_matrix(x: NDArray[np.float64], name: str) -> NDArray[np.float64]:
-    values = np.asarray(x, dtype=np.float64)
-    if values.ndim != 2:
-        raise ValueError(f"{name} must be 2D.")
-    if values.shape[0] == 0 or values.shape[1] == 0:
-        raise ValueError(f"{name} must be non-empty.")
-    if not np.all(np.isfinite(values)):
-        raise ValueError(f"{name} must contain only finite values.")
-    return values

src/nns/norm.py

@@ -6,6 +6,7 @@
 import numpy as np
 from numpy.typing import NDArray
 
+from nns._helpers import _as_matrix
 from nns.dependence import nns_dep
 
 
@@ -39,14 +40,14 @@ def nns_norm(
       vector.
     """
     if isinstance(x, np.ndarray):
-        values = _as_matrix(x)
+        values = _as_matrix(x, "x")
     else:
         series = [_as_vector(item, index) for index, item in enumerate(x)]
         if not series:
             raise ValueError("x must be non-empty.")
         if len({item.size for item in series}) > 1:
             return _norm_unequal_series(series)
-        values = _as_matrix(np.column_stack(series))
+        values = _as_matrix(np.column_stack(series), "x")
     means = np.mean(values, axis=0)
     means = means.copy()
     means[means == 0.0] = 1e-10
@@ -97,14 +98,3 @@ def _as_vector(x: NDArray[np.float64], index: int) -> NDArray[np.float64]:
     if not np.all(np.isfinite(values)):
         raise ValueError(f"x[{index}] must contain only finite values.")
     return values
-
-
-def _as_matrix(x: NDArray[np.float64]) -> NDArray[np.float64]:
-    values = np.asarray(x, dtype=np.float64)
-    if values.ndim != 2:
-        raise ValueError("x must be 2D.")
-    if values.shape[0] == 0 or values.shape[1] == 0:
-        raise ValueError("x must be non-empty.")
-    if not np.all(np.isfinite(values)):
-        raise ValueError("x must contain only finite values.")
-    return values

src/nns/part.py

@@ -6,6 +6,7 @@
 import numpy as np
 from numpy.typing import NDArray
 
+from nns._helpers import _as_pair
 from nns._native import native_fn
 from nns.central_tendencies import _nearest_int_half_up_array, nns_mode
 from nns.dependence import _gravity
@@ -259,20 +260,3 @@ def _validate_noise_reduction(value: str) -> NoiseReduction:
             "noise_reduction must be one of 'mean', 'median', 'mode', 'mode_class', 'off'."
         )
     return cast(NoiseReduction, noise)
-
-
-def _as_pair(
-    x: NDArray[np.float64],
-    y: NDArray[np.float64],
-) -> tuple[NDArray[np.float64], NDArray[np.float64]]:
-    x_values = np.asarray(x, dtype=np.float64)
-    y_values = np.asarray(y, dtype=np.float64)
-    if x_values.ndim != 1 or y_values.ndim != 1:
-        raise ValueError("x and y must be 1D.")
-    if x_values.size == 0:
-        raise ValueError("x and y must be non-empty.")
-    if x_values.size != y_values.size:
-        raise ValueError("x and y must have the same length.")
-    if not np.all(np.isfinite(x_values)) or not np.all(np.isfinite(y_values)):
-        raise ValueError("x and y must contain only finite values.")
-    return x_values, y_values