carto_flow.flow_cartogram.history¶

History management system for iterative algorithms.

This module provides a flexible and extensible system for tracking algorithm state across iterations. It supports any iterative algorithm through the BaseSnapshot interface, making it reusable across different domains. The system provides both access to all data for a single snapshot (by index or iteration), and access to all snapshots for a single variable (by variable name).

Classes:

BaseSnapshot –

Abstract base class for algorithm state snapshots.
CartogramSnapshot –

Snapshot class for cartogram algorithm statistics.
CartogramInternalsSnapshot –

Snapshot class for cartogram internal state.
History –

Container class for managing collections of snapshots.
ErrorRecord –

Lightweight scalar error metrics for a single iteration.
ConvergenceHistory –

Lightweight history of scalar convergence metrics for all iterations.

Notes

To create a custom snapshot class, inherit from BaseSnapshot and add your algorithm-specific attributes:

from carto_flow.flow_cartogram.history import BaseSnapshot from dataclasses import dataclass import numpy as np

@dataclass ... class CustomSnapshot(BaseSnapshot): ... iteration: int ... loss: float = None ... accuracy: float = None ... weights: np.ndarray = None

Examples:

>>> from carto_flow.flow_cartogram.history import History, CartogramSnapshot
>>> from carto_flow.flow_cartogram.errors import MorphErrors
>>> import numpy as np
>>> history = History()
>>> errors = MorphErrors(np.array([0.1]), 0.1, 0.1, np.array([7.2]), 7.2, 7.2)
>>> snapshot = CartogramSnapshot(iteration=0, errors=errors)
>>> history.add_snapshot(snapshot)
>>> error_history = history.get_variable_history('errors')

BaseSnapshot `dataclass` ¶

BaseSnapshot(iteration: int)

Bases: ABC

Abstract base class for algorithm snapshots.

Defines the minimum interface required for any snapshot to work with :class:History. Subclass this and add algorithm-specific attributes.

Attributes:

iteration (int) –

The iteration number this snapshot represents.

Methods:

get_all_variables –

Return a dict of all non-None, non-private attributes.
get_variable –

Return the value of attribute name.
has_variable –

Return True if the attribute exists and is not None.

get_all_variables ¶

get_all_variables() -> dict[str, Any]

Return a dict of all non-None, non-private attributes.

get_variable ¶

get_variable(name: str) -> Any

Return the value of attribute name.

Raises AttributeError if the attribute does not exist.

has_variable ¶

has_variable(name: str) -> bool

Return True if the attribute exists and is not None.

CartogramInternalsSnapshot `dataclass` ¶

CartogramInternalsSnapshot(
    iteration: int,
    rho: ndarray | None = None,
    vx: ndarray | None = None,
    vy: ndarray | None = None,
    geometry_mask: ndarray | None = None,
)

Bases: BaseSnapshot

Snapshot of cartogram internal state at one iteration.

This class holds all the internal data for one iteration of the cartogram algorithm, making it easy to access all variables for a specific time step.

Attributes:

iteration (int) –

The iteration number this snapshot represents.
rho (Optional[ndarray]) –

Density field array.
vx (Optional[ndarray]) –

X-component of the effective velocity field used for displacement (after any anisotropy modulation and smoothing).
vy (Optional[ndarray]) –

Y-component of the effective velocity field used for displacement (after any anisotropy modulation and smoothing).
geometry_mask (Optional[ndarray]) –

Geometry index mask where: - -1 = outside all geometries - k = inside geometry k (0 <= k < number of geometries)

Methods:

__repr__ –

Concise string representation for terminal display.

repr ¶

__repr__() -> str

Concise string representation for terminal display.

CartogramSnapshot `dataclass` ¶

CartogramSnapshot(
    iteration: int,
    geometry: Any | None = None,
    landmarks: Any | None = None,
    coords: Any | None = None,
    errors: MorphErrors | None = None,
    density: ndarray | None = None,
)

Bases: BaseSnapshot

Snapshot of cartogram statistics at one iteration.

This class holds statistical data for one iteration of the cartogram algorithm, making it easy to access all variables for a specific time step.

Attributes:

iteration (int) –

The iteration number this snapshot represents.
geometry (Optional[Any]) –

List of shapely geometries or GeoDataFrame containing the current polygon geometries.
landmarks (Optional[Any]) –

List of morphed landmark geometries if landmarks were provided.
coords (Optional[Any]) –

Displaced coordinates for displacement field computation. Format matches the input coordinates format (points, grid, or mesh).
errors (Optional[MorphErrors]) –

Structured error metrics object containing all error fields. Provides consistent access to log_errors, percentage errors, etc.
density (Optional[ndarray]) –

Current density values for each geometry (values / current_areas).

Methods:

__repr__ –

Concise string representation for terminal display.

repr ¶

__repr__() -> str

Concise string representation for terminal display.

ConvergenceHistory ¶

ConvergenceHistory(capacity: int | None = None)

Lightweight history of scalar convergence metrics.

Stores scalar error metrics for every iteration of the morphing algorithm, enabling detailed convergence analysis without the memory overhead of full snapshots. Uses pre-allocated numpy arrays for efficiency.

Parameters:

capacity (int, default: None ) –

Maximum number of iterations to store. If not provided, uses dynamic resizing (less efficient but more flexible).

Attributes:

iterations (ndarray) –

Array of iteration numbers (int64).
mean_log_errors (ndarray) –

Array of mean log2 errors (float64).
max_log_errors (ndarray) –

Array of max log2 errors (float64).
mean_errors_pct (ndarray) –

Array of mean percentage errors (float64).
max_errors_pct (ndarray) –

Array of max percentage errors (float64).

Examples:

>>> from carto_flow.flow_cartogram.history import ConvergenceHistory
>>> from carto_flow.flow_cartogram.errors import MorphErrors
>>> import numpy as np
>>> convergence = ConvergenceHistory(capacity=100)
>>> errors = MorphErrors(np.array([0.1]), 0.1, 0.2, np.array([7.2]), 7.2, 14.9)
>>> convergence.add(1, errors)
>>> print(len(convergence))
1
>>> record = convergence[0]
>>> print(record.iteration)
1

Methods:

__getitem__ –

Get error record by index.
__iter__ –

Iterate over error records.
__len__ –

Return number of recorded iterations.
__repr__ –

Concise string representation.
add –

Add error metrics for an iteration.
finalize –

Trim arrays to actual size, freeing unused memory.
get_by_iteration –

Get error record for a specific iteration.
to_dict –

Convert to dictionary of arrays for easy export.

iterations `property` ¶

iterations: ndarray

Array of iteration numbers.

max_errors_pct `property` ¶

max_errors_pct: ndarray

Array of max percentage errors.

max_log_errors `property` ¶

max_log_errors: ndarray

Array of max log2 errors.

mean_errors_pct `property` ¶

mean_errors_pct: ndarray

Array of mean percentage errors.

mean_log_errors `property` ¶

mean_log_errors: ndarray

Array of mean log2 errors.

getitem ¶

__getitem__(index: int) -> ErrorRecord

Get error record by index.

Parameters:

index (int) –

Index into the history (supports negative indexing).

Returns:

ErrorRecord –

Error metrics for the requested index.

iter ¶

__iter__() -> Iterator[ErrorRecord]

Iterate over error records.

len ¶

__len__() -> int

Return number of recorded iterations.

repr ¶

__repr__() -> str

Concise string representation.

add ¶

add(iteration: int, errors: MorphErrors) -> None

Add error metrics for an iteration.

Parameters:

iteration (int) –

The iteration number.
errors (MorphErrors) –

Error metrics object (only scalar values are extracted).

finalize ¶

finalize() -> None

Trim arrays to actual size, freeing unused memory.

get_by_iteration ¶

get_by_iteration(iteration: int) -> ErrorRecord | None

Get error record for a specific iteration.

Parameters:

iteration (int) –

The iteration number to retrieve.

Returns:

ErrorRecord or None –

The error record, or None if iteration not found.

to_dict ¶

to_dict() -> dict[str, np.ndarray]

Convert to dictionary of arrays for easy export.

Returns:

dict –

Dictionary with keys 'iteration', 'mean_log_error', 'max_log_error', 'mean_error_pct', 'max_error_pct'.

ErrorRecord `dataclass` ¶

ErrorRecord(
    iteration: int,
    mean_log_error: float,
    max_log_error: float,
    mean_error_pct: float,
    max_error_pct: float,
)

Scalar error metrics for a single iteration.

This lightweight class stores only the 4 scalar error metrics, unlike MorphErrors which also stores per-geometry error arrays. Memory footprint: ~40 bytes per instance.

Attributes:

iteration (int) –

The iteration number.
mean_log_error (float) –

Mean of absolute log2 errors across all geometries.
max_log_error (float) –

Maximum of absolute log2 errors across all geometries.
mean_error_pct (float) –

Mean approximate percentage error.
max_error_pct (float) –

Maximum approximate percentage error.

Methods:

__repr__ –

Concise string representation.

repr ¶

__repr__() -> str

Concise string representation.

History `dataclass` ¶

History(snapshots: list[S] = list())

Bases: Generic[S]

Manages a collection of :class:BaseSnapshot objects.

Provides list-like index access, iteration-based lookup, and per-variable history across all snapshots. Works with any snapshot class that inherits from :class:BaseSnapshot.

Attributes:

snapshots (list[S]) –

Snapshots in chronological order.

Methods:

add_snapshot –

Append snapshot to the history.
get_iterations –

Return the list of recorded iteration numbers.
get_snapshot –

Return the snapshot for iteration, or None if not found.
get_variable_at_iteration –

Return variable_name at iteration.
get_variable_history –

Return a list of variable_name values across all snapshots.
latest –

Return the most recent snapshot, or None if empty.
variable_summary –

Return summary statistics for variable_name across all snapshots.

add_snapshot ¶

add_snapshot(snapshot: S) -> None

Append snapshot to the history.

get_iterations ¶

get_iterations() -> list[int]

Return the list of recorded iteration numbers.

get_snapshot ¶

get_snapshot(iteration: int) -> S | None

Return the snapshot for iteration, or None if not found.

get_variable_at_iteration ¶

get_variable_at_iteration(
    variable_name: str, iteration: int
) -> Any

Return variable_name at iteration.

Raises ValueError if no snapshot exists for that iteration.

get_variable_history ¶

get_variable_history(variable_name: str) -> list[Any]

Return a list of variable_name values across all snapshots.

Missing values (attribute absent or None) are represented as None.

latest ¶

latest() -> S | None

Return the most recent snapshot, or None if empty.

variable_summary ¶

variable_summary(variable_name: str) -> dict[str, Any]

Return summary statistics for variable_name across all snapshots.

MorphErrors `dataclass` ¶

MorphErrors(
    log_errors: ndarray,
    mean_log_error: float,
    max_log_error: float,
    errors_pct: ndarray,
    mean_error_pct: float,
    max_error_pct: float,
)

Structured error metrics from morphing computation.

Attributes:

log_errors (ndarray) –

Array of log2 area errors for each geometry. Computed as log2(current_area / target_area). Positive values indicate oversized regions, negative values indicate undersized regions. This representation is symmetric: a value of +1 means 2x too large, -1 means 2x too small.
mean_log_error (float) –

Mean of absolute log2 errors across all geometries.
max_log_error (float) –

Maximum of absolute log2 errors across all geometries.
errors_pct (ndarray) –

Approximate percentage error for each geometry. Computed as sign(log_error) * (2^|log_error| - 1) * 100.
mean_error_pct (float) –

Mean approximate percentage error across all geometries.
max_error_pct (float) –

Maximum approximate percentage error across all geometries.

Methods:

__repr__ –

Concise string representation for terminal display.

repr ¶

__repr__() -> str

Concise string representation for terminal display.

carto_flow.flow_cartogram.history¶

BaseSnapshot dataclass ¶

get_all_variables ¶

get_variable ¶

has_variable ¶

CartogramInternalsSnapshot dataclass ¶

__repr__ ¶

CartogramSnapshot dataclass ¶

__repr__ ¶

ConvergenceHistory ¶

iterations property ¶

max_errors_pct property ¶

max_log_errors property ¶

mean_errors_pct property ¶

mean_log_errors property ¶

__getitem__ ¶

__iter__ ¶

__len__ ¶

__repr__ ¶

add ¶

finalize ¶

get_by_iteration ¶

to_dict ¶

ErrorRecord dataclass ¶

__repr__ ¶

History dataclass ¶

add_snapshot ¶

get_iterations ¶

get_snapshot ¶

get_variable_at_iteration ¶

get_variable_history ¶

latest ¶

variable_summary ¶

MorphErrors dataclass ¶

__repr__ ¶

BaseSnapshot `dataclass` ¶

CartogramInternalsSnapshot `dataclass` ¶

repr ¶

CartogramSnapshot `dataclass` ¶

repr ¶

iterations `property` ¶

max_errors_pct `property` ¶

max_log_errors `property` ¶

mean_errors_pct `property` ¶

mean_log_errors `property` ¶

getitem ¶

iter ¶

len ¶

repr ¶

ErrorRecord `dataclass` ¶

repr ¶

History `dataclass` ¶

MorphErrors `dataclass` ¶

repr ¶