modelling

py3dinterpolations.modelling

Modelling pipeline for 3D interpolation.

Estimator(griddata, params, scoring='neg_mean_absolute_error', verbose=3)

Parameter estimation via sklearn GridSearchCV.

Currently supports pykrige's Krige wrapper for cross-validation.

Parameters:

Name	Type	Description	Default
griddata	GridData	Training data.	required
params	dict[str, list[object]]	Parameter grid for GridSearchCV.	required
scoring	str	Scoring method. See sklearn docs.	'neg_mean_absolute_error'
verbose	int	Verbosity level (0-3).	3

Source code in py3dinterpolations/modelling/estimator.py

def __init__(
    self,
    griddata: GridData,
    params: dict[str, list[object]],
    scoring: str = "neg_mean_absolute_error",
    verbose: int = 3,
):
    self.estimator = GridSearchCV(
        Krige(),
        params,
        scoring=scoring,
        verbose=verbose,
    )
    self.estimator.fit(
        y=griddata.numpy_data[:, 3],
        X=griddata.numpy_data[:, 0:3],
    )

Modeler(griddata, grid, model)

Orchestrates fitting a model and predicting on a 3D grid.

Handles normalization-aware grid selection and standardization reversal.

Parameters:

Name	Type	Description	Default
griddata	GridData	Training data.	required
grid	Grid3D	3D grid for predictions.	required
model	BaseModel	Fitted or unfitted BaseModel instance. Will be fit on construction.	required

Source code in py3dinterpolations/modelling/modeler.py

def __init__(
    self,
    griddata: GridData,
    grid: Grid3D,
    model: BaseModel,
):
    self._griddata = griddata
    self._grid = grid
    self._model = model
    self._result: InterpolationResult | None = None

    # Fit the model on training data
    data = griddata.numpy_data
    self._model.fit(data[:, 0], data[:, 1], data[:, 2], data[:, 3])
    logger.info("Model %s fitted on %d points", model.name, len(data))

predict(**kwargs)

Make predictions, handling normalization and standardization reversal.

Returns:

Type	Description
ndarray	Interpolated numpy array.

Source code in py3dinterpolations/modelling/modeler.py

def predict(self, **kwargs: object) -> np.ndarray:
    """Make predictions, handling normalization and standardization reversal.

    Returns:
        Interpolated numpy array.
    """
    logger.info("Starting prediction on grid %s", self._grid)

    # Use normalized grid if normalization was applied
    params = self._griddata.preprocessing_params
    if params is not None and params.normalization is not None:
        grid_arrays = self._grid.normalized_grid
    else:
        grid_arrays = self._grid.grid

    # Predict
    result = self._model.predict(
        grid_arrays["X"],
        grid_arrays["Y"],
        grid_arrays["Z"],
        **kwargs,
    )

    interpolated = result.interpolated
    variance = result.variance

    # Reverse standardization if it was applied
    if params is not None and params.standardization is not None:
        std_params = params.standardization
        interpolated = interpolated * std_params.std + std_params.mean
        if variance is not None:
            # Variance scales with the square of the standard deviation
            # and is not shifted by the mean
            variance = variance * (std_params.std**2)

    self._result = InterpolationResult(
        interpolated=interpolated,
        variance=variance,
        probability=result.probability,
    )

    # Also attach to grid
    self._grid.result = self._result

    logger.info("Prediction complete")
    return interpolated

BaseModel

Bases: ABC

Interface for interpolation models.

All models must implement fit() and predict() with consistent signatures.

name abstractmethod property

Human-readable model name.

fit(x, y, z, v) abstractmethod

Fit the model to training data.

Parameters:

Name	Type	Description	Default
x	ndarray	X coordinates of training points.	required
y	ndarray	Y coordinates of training points.	required
z	ndarray	Z coordinates of training points.	required
v	ndarray	Values at training points.	required

Source code in py3dinterpolations/modelling/models/base.py

@abstractmethod
def fit(self, x: np.ndarray, y: np.ndarray, z: np.ndarray, v: np.ndarray) -> None:
    """Fit the model to training data.

    Args:
        x: X coordinates of training points.
        y: Y coordinates of training points.
        z: Z coordinates of training points.
        v: Values at training points.
    """
    ...

predict(grid_x, grid_y, grid_z, **kwargs) abstractmethod

Predict on 1D grid arrays.

Parameters:

Name	Type	Description	Default
grid_x	ndarray	1D array of X grid coordinates.	required
grid_y	ndarray	1D array of Y grid coordinates.	required
grid_z	ndarray	1D array of Z grid coordinates.	required

Returns:

Type	Description
InterpolationResult	Interpolation result with at least the interpolated field.

Source code in py3dinterpolations/modelling/models/base.py

@abstractmethod
def predict(
    self,
    grid_x: np.ndarray,
    grid_y: np.ndarray,
    grid_z: np.ndarray,
    **kwargs: object,
) -> InterpolationResult:
    """Predict on 1D grid arrays.

    Args:
        grid_x: 1D array of X grid coordinates.
        grid_y: 1D array of Y grid coordinates.
        grid_z: 1D array of Z grid coordinates.

    Returns:
        Interpolation result with at least the interpolated field.
    """
    ...

IDWModel(power=1.0, threshold=1e-10)

Bases: BaseModel

Vectorized IDW interpolation.

Uses numpy broadcasting instead of Python loops for ~1000x speedup on typical workloads. Batches computation for memory safety.

Parameters:

Name	Type	Description	Default
power	float	Power parameter controlling distance decay. Higher values give more weight to nearby points.	1.0
threshold	float	Distance below which a point is treated as coincident with a training point (exact interpolation).	1e-10

Source code in py3dinterpolations/modelling/models/idw.py

def __init__(self, power: float = 1.0, threshold: float = 1e-10):
    self._power = power
    self._threshold = threshold
    self._points: np.ndarray | None = None
    self._values: np.ndarray | None = None

fit(x, y, z, v)

Store training data.

Source code in py3dinterpolations/modelling/models/idw.py

def fit(self, x: np.ndarray, y: np.ndarray, z: np.ndarray, v: np.ndarray) -> None:
    """Store training data."""
    self._points = np.column_stack([x, y, z])
    self._values = v

predict(grid_x, grid_y, grid_z, **kwargs)

Predict on a regular grid defined by 1D arrays.

Returns:

Type	Description
InterpolationResult	InterpolationResult with shape (len(grid_z), len(grid_y), len(grid_x))
InterpolationResult	to match pykrige's output convention.

Source code in py3dinterpolations/modelling/models/idw.py

def predict(
    self,
    grid_x: np.ndarray,
    grid_y: np.ndarray,
    grid_z: np.ndarray,
    **kwargs: object,
) -> InterpolationResult:
    """Predict on a regular grid defined by 1D arrays.

    Returns:
        InterpolationResult with shape (len(grid_z), len(grid_y), len(grid_x))
        to match pykrige's output convention.
    """
    if self._points is None:
        msg = "Model must be fit before predicting"
        raise RuntimeError(msg)

    # Build meshgrid in ij (XYZ) indexing for computation
    mx, my, mz = np.meshgrid(grid_x, grid_y, grid_z, indexing="ij")
    query_points = np.column_stack([mx.ravel(), my.ravel(), mz.ravel()])

    # Batch processing for memory safety
    n_points = len(query_points)
    result = np.empty(n_points)
    for start in range(0, n_points, _BATCH_SIZE):
        end = min(start + _BATCH_SIZE, n_points)
        result[start:end] = self._predict_batch(query_points[start:end])

    # Reshape to (X, Y, Z) then transpose to (Z, Y, X) to match pykrige
    interpolated = result.reshape(mx.shape)
    interpolated = np.einsum("xyz->zyx", interpolated)

    return InterpolationResult(interpolated=interpolated, variance=None)

KrigingModel(**kriging_params)

Bases: BaseModel

Ordinary Kriging 3D wrapper around pykrige.

pykrige fits at construction time, so fit() constructs the OrdinaryKriging3D instance.

Parameters:

Name	Type	Description	Default
**kriging_params	object	Parameters passed to OrdinaryKriging3D constructor.	{}

Source code in py3dinterpolations/modelling/models/kriging.py

def __init__(self, **kriging_params: object):
    self._params = kriging_params
    self._model: OrdinaryKriging3D | None = None

fit(x, y, z, v)

Fit by constructing the OrdinaryKriging3D model.

Source code in py3dinterpolations/modelling/models/kriging.py

def fit(self, x: np.ndarray, y: np.ndarray, z: np.ndarray, v: np.ndarray) -> None:
    """Fit by constructing the OrdinaryKriging3D model."""
    self._model = OrdinaryKriging3D(x, y, z, v, **self._params)

predict(grid_x, grid_y, grid_z, **kwargs)

Execute kriging on the given grid arrays.

Returns:

Type	Description
InterpolationResult	InterpolationResult with interpolated and variance arrays.
InterpolationResult	Shape is (len(grid_z), len(grid_y), len(grid_x)) per pykrige convention.

Source code in py3dinterpolations/modelling/models/kriging.py

def predict(
    self,
    grid_x: np.ndarray,
    grid_y: np.ndarray,
    grid_z: np.ndarray,
    **kwargs: object,
) -> InterpolationResult:
    """Execute kriging on the given grid arrays.

    Returns:
        InterpolationResult with interpolated and variance arrays.
        Shape is (len(grid_z), len(grid_y), len(grid_x)) per pykrige convention.
    """
    if self._model is None:
        msg = "Model must be fit before predicting"
        raise RuntimeError(msg)
    interpolated, variance = self._model.execute(
        style="grid",
        xpoints=grid_x,
        ypoints=grid_y,
        zpoints=grid_z,
        **kwargs,
    )
    return InterpolationResult(
        interpolated=interpolated,
        variance=variance,
    )

SklearnModel(estimator, model_name='sklearn')

Bases: BaseModel

Wrapper for any sklearn estimator with fit/predict interface.

Handles classifiers (predict_proba) and regressors (predict).

Parameters:

Name	Type	Description	Default
estimator	SklearnEstimator	A sklearn estimator instance.	required
model_name	str	Human-readable name for this model.	'sklearn'

Source code in py3dinterpolations/modelling/models/sklearn_model.py

def __init__(self, estimator: SklearnEstimator, model_name: str = "sklearn"):
    self._estimator = estimator
    self._model_name = model_name

fit(x, y, z, v)

Fit the sklearn estimator.

Source code in py3dinterpolations/modelling/models/sklearn_model.py

def fit(self, x: np.ndarray, y: np.ndarray, z: np.ndarray, v: np.ndarray) -> None:
    """Fit the sklearn estimator."""
    X = np.column_stack([x, y, z])
    self._estimator.fit(X, v)

predict(grid_x, grid_y, grid_z, **kwargs)

Predict on a regular grid.

Returns:

Type	Description
InterpolationResult	InterpolationResult with shape (len(grid_z), len(grid_y), len(grid_x))
InterpolationResult	to match the convention of other models.

Source code in py3dinterpolations/modelling/models/sklearn_model.py

def predict(
    self,
    grid_x: np.ndarray,
    grid_y: np.ndarray,
    grid_z: np.ndarray,
    **kwargs: object,
) -> InterpolationResult:
    """Predict on a regular grid.

    Returns:
        InterpolationResult with shape (len(grid_z), len(grid_y), len(grid_x))
        to match the convention of other models.
    """
    mx, my, mz = np.meshgrid(grid_x, grid_y, grid_z, indexing="ij")
    X = np.column_stack([mx.ravel(), my.ravel(), mz.ravel()])

    predictions = self._estimator.predict(X)
    interpolated = predictions.reshape(mx.shape)
    # Transpose from XYZ to ZYX to match pykrige convention
    interpolated = np.einsum("xyz->zyx", interpolated)

    probability = None
    if isinstance(self._estimator, SklearnClassifier):
        proba = self._estimator.predict_proba(X)
        probability = proba.reshape((*mx.shape, -1))

    return InterpolationResult(
        interpolated=interpolated,
        probability=probability,
    )

PreprocessingKwargs

Bases: TypedDict

Type-safe kwargs for Preprocessor construction.

Preprocessor(griddata, downsampling_res=None, downsampling_method=DownsamplingStatistic.MEAN, normalize_xyz=True, standardize_v=True)

Preprocess GridData before interpolation.

Supports downsampling, normalization of XYZ, and standardization of V. Returns a new GridData with preprocessing params attached.

Parameters:

Name	Type	Description	Default
griddata	GridData	Source data to preprocess.	required
downsampling_res	float | None	Block resolution for downsampling. None to skip.	None
downsampling_method	DownsamplingStatistic | str | Callable[..., DataFrame]	Statistic for downsampling, or a custom callable.	MEAN
normalize_xyz	bool	Whether to normalize XYZ to [0, 1].	True
standardize_v	bool	Whether to standardize V to mean=0, std=1.	True

Source code in py3dinterpolations/modelling/preprocessor.py

def __init__(
    self,
    griddata: GridData,
    downsampling_res: float | None = None,
    downsampling_method: (
        DownsamplingStatistic | str | Callable[..., pd.DataFrame]
    ) = DownsamplingStatistic.MEAN,
    normalize_xyz: bool = True,
    standardize_v: bool = True,
):
    self.griddata = griddata
    self.downsampling_res = downsampling_res
    self.downsampling_method = downsampling_method
    self.normalize_xyz = normalize_xyz
    self.standardize_v = standardize_v

preprocess()

Execute the preprocessing pipeline.

Returns:

Type	Description
GridData	New GridData with preprocessed data and params attached.

Source code in py3dinterpolations/modelling/preprocessor.py

def preprocess(self) -> GridData:
    """Execute the preprocessing pipeline.

    Returns:
        New GridData with preprocessed data and params attached.
    """
    logger.info("Starting preprocessing")
    data = self.griddata.data.copy().reset_index()[["ID", "X", "Y", "Z", "V"]]

    downsampling_params: DownsamplingParams | None = None
    normalization_params: dict[Axis, NormalizationParams] | None = None
    standardization_params: StandardizationParams | None = None

    if self.downsampling_res is not None:
        data = self._downsample_data(data, statistic=self.downsampling_method)
        downsampling_params = DownsamplingParams(resolution=self.downsampling_res)

    if self.normalize_xyz:
        data, normalization_params = self._normalize_xyz(data)

    if self.standardize_v:
        data, standardization_params = self._standardize_v(data)

    params = PreprocessingParams(
        downsampling=downsampling_params,
        normalization=normalization_params,
        standardization=standardization_params,
    )
    logger.info("Preprocessing complete: %s", params)
    return GridData(data, preprocessing_params=params)

get_model(model_type, **kwargs)

Instantiate a model by type.

Parameters:

Name	Type	Description	Default
model_type	ModelType | str	Model identifier, either a ModelType enum or its string value.	required
**kwargs	object	Parameters passed to the model constructor.	{}

Returns:

Type	Description
BaseModel	An instantiated model ready for fit().

Raises:

Type	Description
ValueError	If model_type is not in the registry.

Source code in py3dinterpolations/modelling/models/__init__.py

def get_model(model_type: ModelType | str, **kwargs: object) -> BaseModel:
    """Instantiate a model by type.

    Args:
        model_type: Model identifier, either a ModelType enum or its string value.
        **kwargs: Parameters passed to the model constructor.

    Returns:
        An instantiated model ready for fit().

    Raises:
        ValueError: If model_type is not in the registry.
    """
    model_type = ModelType(model_type)
    cls = MODEL_REGISTRY.get(model_type)
    if cls is None:
        available = list(MODEL_REGISTRY.keys())
        msg = f"Model {model_type!r} not in registry. Available: {available}"
        raise ValueError(msg)
    return cls(**kwargs)

reverse_preprocessing(griddata)

Reverse all reversible preprocessing transformations.

Reverses normalization of XYZ and standardization of V. Downsampling cannot be reversed.

Parameters:

Name	Type	Description	Default
griddata	GridData	GridData with preprocessing_params set.	required

Returns:

Type	Description
GridData	New GridData with reversed transformations.

Raises:

Type	Description
ValueError	If no preprocessing params are present.

Source code in py3dinterpolations/modelling/preprocessor.py

def reverse_preprocessing(griddata: GridData) -> GridData:
    """Reverse all reversible preprocessing transformations.

    Reverses normalization of XYZ and standardization of V.
    Downsampling cannot be reversed.

    Args:
        griddata: GridData with preprocessing_params set.

    Returns:
        New GridData with reversed transformations.

    Raises:
        ValueError: If no preprocessing params are present.
    """
    params = griddata.preprocessing_params
    if params is None:
        msg = "No preprocessing has been applied to the data"
        raise ValueError(msg)

    data = griddata.data.copy().reset_index()

    if params.normalization is not None:
        for axis in [Axis.X, Axis.Y, Axis.Z]:
            norm = params.normalization[axis]
            data[axis.value] = data[axis.value] * (norm.max - norm.min) + norm.min

    if params.standardization is not None:
        std = params.standardization
        data["V"] = data["V"] * std.std + std.mean

    return GridData(data)