API reference

The primary public API is the scikit-learn-compatible estimator.

SpatialDeformer

SpatialDeformer(*, backend: str = 'mds', cost: str = 'travel_cost', missing_cost: str | Callable[[GeoDataFrame, ndarray], Any] = 'error', fallback_speed: float = 8.0, impute_by: str | None = None, source: str | None = 'source', target: str | None = 'target', scale: str | float = 'auto', geo_weight: float = 0.1, spatial_weight: str = 'uniform', spatial_bandwidth: float | None = None, directed: bool = False, directed_policy: str = 'mean', auto_project: bool = True, n_neighbors: int = 8, displacement_power: float = 2.0, node_precision: int = 9, max_iter: int = 300, tol: float = 1e-06, n_components: int = 2)

Bases: TransformerMixin, BaseEstimator

Deform a line network so map distance represents network travel cost.

Parameters:

Name	Type	Description	Default
backend	('mds', 'isomap')	"mds" minimizes spatially weighted metric stress with a geographic anchoring penalty. "isomap" performs classical scaling on the graph shortest-path distances and aligns the result to the map.	"mds"
cost	str	Edge column used as travel cost when y is not passed to fit.	"travel_cost"
missing_cost	('error', 'drop', 'length', 'median')	Policy for NaN or infinite edge costs. A callable receives (X, costs) and must return one resolved value per input edge.	"error"
fallback_speed	float	Working-CRS distance units per cost unit used by missing_cost="length". For a metric CRS and costs in seconds this is metres per second.	8.0
impute_by	str or None	Optional column used for groupwise median imputation. Groups without a finite value fall back to the global median.	None
source	str or None	Edge endpoint identifier columns. When absent, rounded geometry endpoints define node identity.	'source'
target	str or None	Edge endpoint identifier columns. When absent, rounded geometry endpoints define node identity.	'source'
scale	'auto' or float	Conversion from cost units to working-CRS units. Reuse a fitted scale_ across scenarios to retain contraction/expansion magnitude.	"auto"
geo_weight	float	Strength of preservation of the original node coordinates.	0.1
spatial_weight	('uniform', 'gaussian', 'inverse')	Pair weighting in the MDS stress objective.	"uniform"
spatial_bandwidth	float or None	Working-CRS bandwidth for non-uniform pair weights. The median geographic node distance is used when omitted.	None
directed	bool	Whether edge costs form a directed graph.	False
directed_policy	('mean', 'min', 'max')	How reciprocal directed shortest-path costs become a symmetric metric.	"mean"
auto_project	bool	Automatically project a geographic CRS to an estimated local UTM CRS.	True
n_neighbors	int	Nearby graph nodes used to interpolate displacement at line vertices.	8
displacement_power	float	Inverse-distance interpolation power for non-node vertices.	2.0
node_precision	int	Decimal precision used when geometry endpoints define node identity.	9
max_iter	int	Maximum SMACOF iterations.	300
tol	float	Relative stress convergence tolerance.	1e-6

Attributes:

Name	Type	Description
embedding_	ndarray of shape (n_nodes, 2)	Deformed node coordinates in working_crs_.
geographic_coordinates_	ndarray of shape (n_nodes, 2)	Original node coordinates in working_crs_.
graph_distances_	ndarray of shape (n_nodes, n_nodes)	Symmetric shortest travel-cost distances.
node_ids_	ndarray of shape (n_nodes,)	Node identifiers in embedding order.
scale_	float	Fitted cost-to-map-unit scale.
stress_	float	Final raw weighted stress.
normalized_stress_	float	Stress divided by weighted squared target distances.
costs_	ndarray of shape (n_edges,)	Resolved costs aligned with the input. Dropped positions remain NaN or infinite and are excluded by edge_mask_.
edge_mask_	ndarray of shape (n_edges,)	Boolean mask of input edges used to fit the graph.
imputed_edges_	list	Input index labels whose missing costs were imputed.
dropped_edges_	list	Input index labels excluded by missing_cost="drop".

Source code in src/spatialdeform/_estimator.py

def __init__(
    self,
    *,
    backend: str = "mds",
    cost: str = "travel_cost",
    missing_cost: str | Callable[[gpd.GeoDataFrame, np.ndarray], Any] = "error",
    fallback_speed: float = 8.0,
    impute_by: str | None = None,
    source: str | None = "source",
    target: str | None = "target",
    scale: str | float = "auto",
    geo_weight: float = 0.1,
    spatial_weight: str = "uniform",
    spatial_bandwidth: float | None = None,
    directed: bool = False,
    directed_policy: str = "mean",
    auto_project: bool = True,
    n_neighbors: int = 8,
    displacement_power: float = 2.0,
    node_precision: int = 9,
    max_iter: int = 300,
    tol: float = 1e-6,
    n_components: int = 2,
) -> None:
    self.backend = backend
    self.cost = cost
    self.missing_cost = missing_cost
    self.fallback_speed = fallback_speed
    self.impute_by = impute_by
    self.source = source
    self.target = target
    self.scale = scale
    self.geo_weight = geo_weight
    self.spatial_weight = spatial_weight
    self.spatial_bandwidth = spatial_bandwidth
    self.directed = directed
    self.directed_policy = directed_policy
    self.auto_project = auto_project
    self.n_neighbors = n_neighbors
    self.displacement_power = displacement_power
    self.node_precision = node_precision
    self.max_iter = max_iter
    self.tol = tol
    self.n_components = n_components

fit

fit(X: GeoDataFrame, y: Any = None)

Fit the graph embedding from edge geometries and travel costs.

Source code in src/spatialdeform/_estimator.py

def fit(self, X: gpd.GeoDataFrame, y: Any = None):
    """Fit the graph embedding from edge geometries and travel costs."""
    self._validate_hyperparameters()
    working, working_crs = prepare_working_edges(X, auto_project=self.auto_project)
    costs, edge_mask, imputed, dropped = self._prepare_costs(X, working, y)
    fit_working = working.loc[edge_mask].copy()
    graph = extract_graph(
        fit_working,
        source=self.source,
        target=self.target,
        node_precision=self.node_precision,
    )
    graph_distances = shortest_path_distances(
        len(graph.node_ids),
        graph.edge_indices,
        costs[edge_mask],
        directed=self.directed,
        directed_policy=self.directed_policy,
    )
    result = embed_graph_distances(
        graph_distances,
        graph.coordinates,
        backend=self.backend,
        n_components=self.n_components,
        scale=self.scale,
        geo_weight=self.geo_weight,
        spatial_weight=self.spatial_weight,
        spatial_bandwidth=self.spatial_bandwidth,
        max_iter=self.max_iter,
        tol=self.tol,
    )

    self.node_ids_ = graph.node_ids
    self.geographic_coordinates_ = graph.coordinates
    self.embedding_ = result.coordinates
    self.displacement_ = self.embedding_ - self.geographic_coordinates_
    self.graph_distances_ = graph_distances
    self.scale_ = result.scale
    self.stress_ = result.stress
    self.normalized_stress_ = result.normalized_stress
    self.n_iter_ = result.n_iter
    self.costs_ = costs
    self.edge_mask_ = edge_mask
    self.imputed_edges_ = list(X.index[imputed])
    self.dropped_edges_ = list(X.index[dropped])
    self.n_edges_in_ = len(X)
    self.n_edges_used_ = int(edge_mask.sum())
    self.working_crs_ = working_crs
    self.input_crs_ = X.crs
    self.n_features_in_ = len(X.columns)
    if all(isinstance(column, str) for column in X.columns):
        self.feature_names_in_ = np.asarray(X.columns, dtype=object)
    self._displacement_transform = make_displacement_transform(
        self.geographic_coordinates_,
        self.embedding_,
        n_neighbors=self.n_neighbors,
        power=self.displacement_power,
    )
    return self

transform

transform(X: GeoDataFrame) -> gpd.GeoDataFrame

Apply the fitted smooth displacement field to line geometries.

Source code in src/spatialdeform/_estimator.py

def transform(self, X: gpd.GeoDataFrame) -> gpd.GeoDataFrame:
    """Apply the fitted smooth displacement field to line geometries."""
    check_is_fitted(self, "embedding_")
    working = self._to_working_crs(X)
    transformed = working.copy()
    transformed.geometry = working.geometry.map(
        lambda geometry: deform_geometry(geometry, self._displacement_transform)
    )
    if X.crs is not None and transformed.crs != X.crs:
        transformed = transformed.to_crs(X.crs)
    return transformed

get_nodes

get_nodes(crs: Any = None) -> gpd.GeoDataFrame

Return fitted deformed nodes as a GeoDataFrame.

By default nodes use the input CRS. Pass a CRS explicitly to override that behavior; pass working_crs_ to inspect optimizer coordinates.

Source code in src/spatialdeform/_estimator.py

def get_nodes(self, crs: Any = None) -> gpd.GeoDataFrame:
    """Return fitted deformed nodes as a GeoDataFrame.

    By default nodes use the input CRS. Pass a CRS explicitly to override
    that behavior; pass ``working_crs_`` to inspect optimizer coordinates.
    """
    check_is_fitted(self, "embedding_")
    nodes = gpd.GeoDataFrame(
        {"node_id": self.node_ids_},
        geometry=[Point(float(x), float(y)) for x, y in self.embedding_],
        crs=self.working_crs_,
    )
    destination = self.input_crs_ if crs is None else crs
    if destination is not None and nodes.crs != destination:
        nodes = nodes.to_crs(destination)
    return nodes

score

score(X: GeoDataFrame | None = None, y: Any = None) -> float

Return negative normalized stress (higher is better).

Source code in src/spatialdeform/_estimator.py

def score(self, X: gpd.GeoDataFrame | None = None, y: Any = None) -> float:
    """Return negative normalized stress (higher is better)."""
    check_is_fitted(self, "embedding_")
    return -float(self.normalized_stress_)

Low-level constrained SMACOF

Large pipelines that already compute their own dissimilarity matrices can use the same optimization core directly. This is how the repository's landmark street-network pipeline shares SpatialDeform without constructing an infeasible all-pairs matrix for every one of its 19,000+ nodes.

spatial_smacof

spatial_smacof(dissimilarities: Any, reference_coordinates: Any, *, pair_weights: Any | None = None, coordinate_weights: float | Any = 0.0, previous_coordinates: Any | None = None, temporal_weight: float = 0.0, init: Any | None = None, max_iter: int = 300, tol: float | None = 1e-06, backend: str = 'auto') -> tuple[np.ndarray, int]

Minimize weighted metric stress with coordinate constraints.

The optimized objective is

sum(i<j) w_ij (||x_i-x_j||-delta_ij)^2 + sum(i) a_i ||x_i-g_i||^2 + beta sum(i) ||x_i-p_i||^2.

backend="auto" uses the optional Rust kernel when installed and falls back to NumPy/SciPy otherwise. Pass "rust" to require acceleration or "python" for reproducible benchmarking. tol=None disables early stopping, which is useful for callers that need a fixed iteration count.

Source code in src/spatialdeform/_smacof.py

def spatial_smacof(
    dissimilarities: Any,
    reference_coordinates: Any,
    *,
    pair_weights: Any | None = None,
    coordinate_weights: float | Any = 0.0,
    previous_coordinates: Any | None = None,
    temporal_weight: float = 0.0,
    init: Any | None = None,
    max_iter: int = 300,
    tol: float | None = 1e-6,
    backend: str = "auto",
) -> tuple[np.ndarray, int]:
    """Minimize weighted metric stress with coordinate constraints.

    The optimized objective is

    ``sum(i<j) w_ij (||x_i-x_j||-delta_ij)^2``
    ``+ sum(i) a_i ||x_i-g_i||^2``
    ``+ beta sum(i) ||x_i-p_i||^2``.

    ``backend="auto"`` uses the optional Rust kernel when installed and falls
    back to NumPy/SciPy otherwise. Pass ``"rust"`` to require acceleration or
    ``"python"`` for reproducible benchmarking. ``tol=None`` disables early
    stopping, which is useful for callers that need a fixed iteration count.
    """
    delta = np.asarray(dissimilarities, dtype=float)
    geographic = np.asarray(reference_coordinates, dtype=float)
    if delta.ndim != 2 or delta.shape[0] != delta.shape[1]:
        raise ValueError("dissimilarities must be a square matrix")
    n = len(delta)
    if geographic.shape != (n, 2):
        raise ValueError("reference_coordinates must have shape (n_samples, 2)")
    if not np.isfinite(delta).all() or np.any(delta < 0):
        raise ValueError("dissimilarities must be finite and non-negative")

    uniform_weights = pair_weights is None
    weights = None if uniform_weights else np.asarray(pair_weights, dtype=float).copy()
    if weights is not None:
        if weights.shape != delta.shape:
            raise ValueError("pair_weights must match dissimilarities")
        if not np.isfinite(weights).all() or np.any(weights < 0):
            raise ValueError("pair_weights must be finite and non-negative")
        np.fill_diagonal(weights, 0.0)

    anchors = np.asarray(coordinate_weights, dtype=float)
    if anchors.ndim == 0:
        anchors = np.full(n, float(anchors))
    if anchors.shape != (n,) or not np.isfinite(anchors).all() or np.any(anchors < 0):
        raise ValueError("coordinate_weights must be non-negative scalar or length n")
    if not np.isfinite(temporal_weight) or temporal_weight < 0:
        raise ValueError("temporal_weight must be non-negative and finite")
    if not isinstance(max_iter, int) or max_iter < 1:
        raise ValueError("max_iter must be a positive integer")
    if tol is not None and (not np.isfinite(tol) or tol < 0):
        raise ValueError("tol must be non-negative, finite, or None")
    if backend not in {"auto", "rust", "python"}:
        raise ValueError("backend must be one of {'auto', 'rust', 'python'}")
    if backend == "rust" and _rust_majorization is None:
        raise RuntimeError(
            "Rust acceleration is not installed; install spatialdeform[rust]"
        )
    use_rust = backend != "python" and _rust_majorization is not None

    previous = None
    if previous_coordinates is not None:
        previous = np.asarray(previous_coordinates, dtype=float)
        if previous.shape != (n, 2):
            raise ValueError("previous_coordinates must have shape (n_samples, 2)")

    x = geographic.copy() if init is None else np.asarray(init, dtype=float).copy()
    if x.shape != (n, 2):
        raise ValueError("init must have shape (n_samples, 2)")

    temporal = float(temporal_weight) if previous is not None else 0.0
    diagonal = anchors + temporal
    factor = None
    uniform_system_diagonal = None
    if uniform_weights:
        ridge = max(float(n - 1), 1.0) * 1e-12
        uniform_system_diagonal = n + diagonal + ridge
    else:
        laplacian = np.diag(weights.sum(axis=1)) - weights
        ridge = max(float(np.max(np.diag(laplacian))), 1.0) * 1e-12
        system = laplacian + np.diag(diagonal + ridge)
        factor = cho_factor(system, check_finite=False)
    constant = anchors[:, None] * geographic
    if previous is not None and temporal > 0:
        constant = constant + temporal * previous

    previous_stress = _stress(x, delta, weights, use_rust)
    unanchored = not np.any(diagonal > 0)
    for iteration in range(1, max_iter + 1):
        majorized = _majorization(x, delta, weights, use_rust)
        right_hand_side = majorized + constant
        if uniform_weights and unanchored:
            updated = right_hand_side / max(n, 1)
        elif uniform_weights:
            inverse_diagonal = 1.0 / uniform_system_diagonal
            scaled = inverse_diagonal[:, np.newaxis] * right_hand_side
            denominator = 1.0 - inverse_diagonal.sum()
            correction = scaled.sum(axis=0) / denominator
            updated = scaled + inverse_diagonal[:, np.newaxis] * correction
        else:
            updated = cho_solve(factor, right_hand_side, check_finite=False)
        if unanchored:
            updated += geographic.mean(axis=0) - updated.mean(axis=0)
        current_stress = _stress(updated, delta, weights, use_rust)
        relative_change = abs(previous_stress - current_stress) / max(
            previous_stress, 1e-12
        )
        x = updated
        if tol is not None and relative_change <= tol:
            return x, iteration
        previous_stress = current_stress
    return x, max_iter

rust_available

rust_available()

Return whether the optional Rust SMACOF kernel is installed.

Source code in src/spatialdeform/_smacof.py

def rust_available() -> bool:
    """Return whether the optional Rust SMACOF kernel is installed."""
    return _rust_majorization is not None

Fitted-state conventions

Calling fit replaces all learned attributes. Constructor parameters remain unchanged, so sklearn.base.clone returns an unfitted estimator with the same configuration.

from sklearn.base import clone

configured = SpatialDeformer(geo_weight=0.2, backend="mds")
fresh = clone(configured)

assert fresh.get_params() == configured.get_params()

transform and get_nodes call scikit-learn’s fitted-state validation and raise NotFittedError when used before fit.