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.

This class defines the minimum interface required for any snapshot to work with the History class. Users can extend this class to add their own algorithm-specific variables.

The History class only requires: 1. An 'iteration' attribute (int) 2. A 'has_variable(name: str)' method 3. A 'get_variable(name: str)' method

All other attributes and methods are optional and algorithm-specific.

Attributes:

iteration (int) –

The iteration number this snapshot represents.

Methods:

get_all_variables –

Get all non-None variables in this snapshot.
get_variable –

Get the value of a specific variable.
has_variable –

Check if this snapshot contains a specific variable.

iteration `instance-attribute` ¶

iteration: int

The iteration number this snapshot represents.

get_all_variables ¶

get_all_variables() -> dict[str, Any]

Get all non-None variables in this snapshot.

Returns:

Dict[str, Any] –

Dictionary mapping variable names to their values.

get_variable ¶

get_variable(name: str) -> Any

Get the value of a specific variable.

Parameters:

name (str) –

Name of the variable to retrieve.

Returns:

Any –

The value of the variable.

Raises:

AttributeError –

If the variable doesn't exist in this snapshot.

has_variable ¶

has_variable(name: str) -> bool

Check if this snapshot contains a specific variable.

Parameters:

name (str) –

Name of the variable to check.

Returns:

bool –

True if the variable exists and is not None, False otherwise.

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[BaseSnapshot] = list())

Manages a collection of snapshots with convenient access patterns.

This class provides: - List-like index access via history[index] or history[start:stop] - Iteration-based lookup via get_snapshot(iteration) - Variable history across all snapshots via get_variable_history(name)

It works with any snapshot class that inherits from BaseSnapshot, making it reusable for any iterative algorithm.

Attributes:

snapshots (List[BaseSnapshot]) –

List of snapshot objects in chronological order.

Examples:

>>> from carto_flow.flow_cartogram.history import History, CartogramSnapshot
>>> history = History()
>>> snapshot = CartogramSnapshot(iteration=0)
>>> history.add_snapshot(snapshot)
>>> latest = history.latest()
>>> geometries = history.get_variable_history('geometry')

Methods:

__getitem__ –

Get snapshot(s) by index, like a list.
__iter__ –

Iterate over snapshots in chronological order.
__len__ –

Return number of snapshots.
__repr__ –

Concise string representation for terminal display.
add_snapshot –

Add a new snapshot to the history.
get_iterations –

Get list of all iteration numbers.
get_snapshot –

Get snapshot for a specific iteration.
get_variable_at_iteration –

Get a specific variable value at a specific iteration.
get_variable_history –

Get the history of a specific variable across all iterations.
latest –

Get the most recent snapshot.
variable_summary –

Get summary statistics for a variable across all iterations.

getitem ¶

__getitem__(
    index: int | slice,
) -> BaseSnapshot | list[BaseSnapshot]

Get snapshot(s) by index, like a list.

This provides intuitive list-like access to snapshots. Use get_snapshot(iteration) if you need to look up by iteration number.

Parameters:

index (int | slice) –

The index or slice to retrieve.

Returns:

BaseSnapshot | list[BaseSnapshot] –

The snapshot at the given index, or a list of snapshots for slices.

Raises:

IndexError –

If the index is out of range.

Examples:

>>> history[0]      # First snapshot
>>> history[-1]     # Last snapshot (same as history.latest())
>>> history[1:3]    # Snapshots at indices 1 and 2

iter ¶

__iter__() -> Iterator[CartogramSnapshot]

Iterate over snapshots in chronological order.

Returns:

Iterator[CartogramSnapshot] –

Iterator over snapshots in chronological order.

len ¶

__len__() -> int

Return number of snapshots.

Returns:

int –

The number of snapshots in the history.

repr ¶

__repr__() -> str

Concise string representation for terminal display.

add_snapshot ¶

add_snapshot(snapshot: BaseSnapshot) -> None

Add a new snapshot to the history.

Parameters:

snapshot (BaseSnapshot) –

The snapshot object to add to the collection.

get_iterations ¶

get_iterations() -> list[int]

Get list of all iteration numbers.

Returns:

List[int] –

List of all iteration numbers in chronological order.

get_snapshot ¶

get_snapshot(iteration: int) -> BaseSnapshot | None

Get snapshot for a specific iteration.

Parameters:

iteration (int) –

The iteration number to retrieve.

Returns:

Optional[BaseSnapshot] –

The snapshot for the given iteration, or None if not found.

get_variable_at_iteration ¶

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

Get a specific variable value at a specific iteration.

Parameters:

variable_name (str) –

Name of the variable to retrieve.
iteration (int) –

The iteration number to retrieve the variable from.

Returns:

Any –

The value of the variable at the specified iteration.

Raises:

ValueError –

If no snapshot exists for the given iteration.

get_variable_history ¶

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

Get the history of a specific variable across all iterations.

Parameters:

variable_name (str) –

Name of the variable to retrieve history for.

Returns:

List[Any] –

List of values for the variable, in chronological order. Empty list if variable not found in any snapshot.

latest ¶

latest() -> CartogramSnapshot | None

Get the most recent snapshot.

Returns:

Optional[CartogramSnapshot] –

The most recent snapshot, or None if no snapshots exist.

variable_summary ¶

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

Get summary statistics for a variable across all iterations.

Parameters:

variable_name (str) –

Name of the variable to summarize.

Returns:

Dict[str, Any] –

Dictionary containing summary statistics. For arrays, includes shape information and data type. For scalars, includes min/max values and type information.

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 ¶

iteration instance-attribute ¶

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 ¶

__getitem__ ¶

__iter__ ¶

__len__ ¶

__repr__ ¶

add_snapshot ¶

get_iterations ¶

get_snapshot ¶

get_variable_at_iteration ¶

get_variable_history ¶

latest ¶

variable_summary ¶

MorphErrors dataclass ¶

__repr__ ¶

BaseSnapshot `dataclass` ¶

iteration `instance-attribute` ¶

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` ¶

getitem ¶

iter ¶

len ¶

repr ¶

MorphErrors `dataclass` ¶

repr ¶