vibespin.utils package

Submodules

utils.cli_helpers module

CLI helper utilities shared by script entry points.

utils.cli_helpers.parse_args_compat(parser: ArgumentParser) → Namespace

Parse CLI args with compatibility for parser wrappers used by some runners.

utils.exceptions module

Project-specific exception types for analysis and runtime failures.

exception utils.exceptions.NumericalAnalysisError

Bases: VibeSpinError, RuntimeError

Base class for mathematically undefined or failed analysis results.

exception utils.exceptions.VibeSpinError

Bases: Exception

Base class for project-specific exceptions.

exception utils.exceptions.ZeroVarianceAutocorrelationError

Bases: NumericalAnalysisError

Raised when autocorrelation is undefined because the series variance is zero.

utils.physics_helpers module

Physics-related utility functions for calculating thermodynamic observables and correlations.

utils.physics_helpers.calculate_autocorr(*, time_series: ndarray, max_lag: int | None = None, window_constant: float = 6.0) → tuple[ndarray, float]

Compute the normalized autocorrelation function and integrated autocorrelation time.

Uses an FFT-based estimator for C(t): zero-pads to 2N, computes the power spectrum, IFFTs, and normalizes by variance and the number of pairs at each lag. Applies the Madras-Sokal automatic windowing rule to stop integration at the lag W where W = window_constant * tau_int, preventing the estimator from accumulating noise at large lags.

Parameters:

• time_series (1-D array of sequential measurements (e.g. magnetization).)

• max_lag (Maximum lag to compute. Defaults to len(time_series) // 2.)

• window_constant (Multiplier for the Madras-Sokal window (default 6.0).)

Returns:

• A tuple (C_t, tau_int) where C_t is the normalized autocorrelation

• array truncated at the window endpoint, and tau_int is the integrated

• autocorrelation time (scalar).

Raises:

• ValueError – If time_series has fewer than 3 elements.:

• ZeroVarianceAutocorrelationError – If time_series has zero variance.:

utils.physics_helpers.calculate_entropy(*, temperatures: ndarray, specific_heat: ndarray, s_ref: float = 0.0) → ndarray

Compute entropy by integrating specific heat over a temperature sweep.

Integrates C(T)/T from high temperature downward so that S(T_max) = s_ref and S(T) = s_ref - integral from T to T_max of C/T dT’. This avoids the C/T divergence near T = 0 and anchors the result at the high-temperature limit where the entropy is analytically known.

Parameters:

• temperatures (1-D array of temperature values. Need not be sorted.)

• specific_heat (Specific heat C_v at each temperature (same length).)

• s_ref (Reference entropy at the highest temperature. Pass 0.0) – for relative entropy or np.log(q) (per site, in units of k_B) for absolute entropy of a q-state model.

Returns:

• 1-D array of entropy values, one per temperature, in the original

• unsorted order of temperatures.

Raises:

ValueError – If arrays differ in length, contain fewer than 2 points,: or include non-positive temperatures.

utils.physics_helpers.calculate_thermodynamics(*, mags: ndarray, engs: ndarray, T: float, L: int) → tuple[float, float, float, float]

Calculate average magnetization, energy, susceptibility, and specific heat.

Parameters:

• mags (Array of magnetization measurements.)

• engs (Array of energy measurements.)

• T (Temperature.)

• L (Linear lattice size.)

Return type:

A tuple of (avg_mag, avg_eng, susceptibility, specific_heat).

Raises:

ValueError – If T is not positive or L is not a positive integer.:

utils.physics_helpers.compute_kinetics_metrics(*, sim: MonteCarloSimulation) → dict[str, float]

Calculate common kinetics metrics (R_sk, xi) for a simulation state.

Parameters:

sim (MonteCarloSimulation instance.)

Return type:

Dictionary containing ‘R_sk’ and ‘xi’.

utils.physics_helpers.estimate_relaxation_time_two_start(trace_random: ndarray, trace_ordered: ndarray, *, k: float = 1.0, smooth_window: int = 30, dwell_window: int = 30, min_fraction_inside: float = 0.85) → int

Estimate thermalization time from convergence of random- and ordered-start traces.

Compares a trajectory started from a random spin configuration against one started from a fully ordered configuration. Returns the first step at which both smoothed traces enter and sustain a common equilibrium band.

Parameters:

• trace_random (1-D magnetization trace from a random initial state.)

• trace_ordered (1-D magnetization trace from an ordered initial state.)

• k (Half-width of the convergence band in units of the tail standard) – deviation. Larger values are more permissive.

• smooth_window (Window length (steps) for the moving-average smoother.)

• dwell_window (Window length (steps) for the sustained-convergence test.)

• min_fraction_inside (Fraction of steps within dwell_window that must) – lie inside the band to declare convergence.

Returns:

• Estimated relaxation time in MC steps. Returns the full trace length

• if no convergence is detected, or 0 for very short traces.

utils.physics_helpers.get_averaged_correlation(*, sim: MonteCarloSimulation, total_steps: int, sample_interval: int) → tuple[ndarray, ndarray]

Run simulation and average the correlation function over multiple configurations.

Parameters:

• sim (An instance of MonteCarloSimulation.)

• total_steps (Total number of MC steps to run.)

• sample_interval (Interval between correlation samples. Must be ≥ 1.)

Returns:

• r (Radial distances.)

• G_r_avg (Averaged correlation values.)

Raises:

ValueError – If sample_interval is less than 1 or total_steps is negative.:

utils.physics_helpers.pair_correlation_x(*, spins: ndarray) → tuple[ndarray, ndarray]

Compute the real-space spin-spin pair correlation G(r) along x.

Uses the Wiener-Khinchin theorem: the autocorrelation of each row is the inverse FFT of the row’s power spectrum. Results are averaged over all rows (y positions) and normalised so G(0) = 1.

Parameters:

spins ((N, N) or (N, N, 2) spin array.)

Returns:

• r_vals (Lag distances r = 0 … N//2 in lattice units.)

• G (Normalised pair correlation G(r) / G(0).)

utils.physics_helpers.power_fit(*, t_arr: ndarray, y_arr: ndarray, mask: ndarray) → tuple[float, float] | tuple[None, None]

Fit a power law y = prefactor * t^exponent via log-log linear regression.

Parameters:

• t_arr (Independent variable array (e.g., time or length scale).)

• y_arr (Dependent variable array; only positive values are used in the fit.)

• mask (Boolean mask selecting the subset of points to include.)

Returns:

• exponent (Fitted power-law exponent, or None if fewer than 3 valid points.)

• prefactor (Fitted prefactor, or None if fewer than 3 valid points.)

utils.physics_helpers.radial_average_sk(*, spins: ndarray) → tuple[ndarray, ndarray]

Compute the circularly averaged structure factor \(S(|k|)\).

Bins \(S(k)\) by integer pixel radius from the DC centre of the shifted FFT, then averages within each annular bin.

Parameters:

spins ((N, N) or (N, N, 2) spin array.)

Returns:

• k_vals (Wavevector magnitudes in units of 2π/N (reciprocal lattice).)

• S_radial (Mean S(k) value for each annular bin.)

utils.system_helpers module

Technical utility functions for file system operations, plotting, and parallel execution.

utils.system_helpers.adaptive_equilibrate(sim: _Sim, *, min_steps: int, probe_steps: int = 500, factor: float = 50.0, max_steps: int = 200000) → int

Equilibrate a simulation adaptively, extending the burn-in until the probe window covers factor integrated autocorrelation times.

After the mandatory min_steps burn-in, the function repeatedly runs a probe_steps measurement via sim.run() and computes tau_int from the resulting magnetization series. If probe_steps >= factor * tau_int, the probe spans enough correlation times for the initial state to be forgotten and equilibration is declared complete. Otherwise the probe has advanced the system state and the loop continues. In the ordered phase the magnetization series has zero variance; this is treated as full equilibration and the function returns immediately.

Parameters:

• sim (Any simulation object implementing equilibrate and run.)

• min_steps (Mandatory burn-in passed to sim.equilibrate before probing.)

• probe_steps (MC steps per probe run (default 500).)

• factor (Required ratio probe_steps / tau_int (default 50.0).)

• max_steps (Hard cap on total steps to prevent unbounded runtime near) – criticality (default 200 000).

Return type:

Total number of MC steps run (burn-in + probes).

Raises:

ValueError – If the adaptive-equilibration parameters are invalid.:

utils.system_helpers.convergence_equilibrate(sim_random: _Sim, sim_ordered: _Sim, *, chunk_size: int = 500, max_steps: int = 200000, **kwargs: Any) → int

Equilibrate two simulations (random- and ordered-start) until they converge.

Uses estimate_relaxation_time_two_start to detect when the two trajectories have entered the same equilibrium band. This is more robust than one-start adaptive methods for complex energy landscapes.

Parameters:

• sim_random (Simulation instance started from a random state.)

• sim_ordered (Simulation instance started from an ordered state.)

• chunk_size (Number of steps to run between convergence checks.)

• max_steps (Hard cap on total steps.)

• **kwargs (Passed to estimate_relaxation_time_two_start (k, smooth, dwell, etc.).)

Return type:

Total number of MC steps run per simulation.

utils.system_helpers.ensure_results_dir(*, directory: str = 'results') → str

Ensure the results directory exists.

Parameters:

directory (Name of the directory to create.)

Return type:

The path to the directory.

utils.system_helpers.parallel_sweep(*, worker_func: Callable, params: Iterable, num_processes: int | None = None) → list

Run a parallel sweep over a set of parameters using a worker function. Uses multiprocessing.Pool and tqdm for progress tracking.

Parameters:

• worker_func (Function to execute in parallel.)

• params (Iterable of parameters to pass to the worker function.)

• num_processes (Number of processes to use. Defaults to CPU count.)

Return type:

List of results from the worker function.

utils.system_helpers.plot_ordering_evolution(*, targets: Sequence[int], snapshots: Sequence[ndarray], gr_data: Sequence[tuple[ndarray, ndarray]], vorticity_data: Sequence[ndarray] | None, title: str, filename: str, directory: str, is_vector: bool = False) → None

Generate and save a multi-row figure showing order evolution over time.

Row 0: Spin configurations (binary for Ising, HSV for XY/Clock). Row 1: Circularly-averaged structure factor \(S(|k|)\) or vorticity maps. Row 2: Radially averaged correlation functions G(r) with xi estimates.

Parameters:

• targets (List of MC steps for each snapshot.)

• snapshots (List of spin arrays.)

• gr_data (List of (r, G) tuples.)

• vorticity_data (Optional list of vorticity arrays.)

• title (Figure-level suptitle.)

• filename (Output filename.)

• directory (Output directory.)

• is_vector (Whether the spins are 2D vectors (True) or scalars (False).)

utils.system_helpers.plot_ordering_kinetics(*, t: ndarray, R_sk: ndarray, R_xi: ndarray, third_metric: ndarray | None, third_metric_label: str | None, exponents: dict[str, float | None], prefactors: dict[str, float | None], fit_mask: ndarray, title: str, filename: str, directory: str, y_label: str = 'Characteristic Length Scale $L(t)$ (lattice units)', left_title: str = 'Domain Coarsening', right_title: str = 'Defect/Boundary Evolution') → None

Generate and save a 2-panel figure showing ordering kinetics.

Left Panel: Growth of length scales R_sk and xi. Right Panel: Growth/Decay of a third metric (Vortex density or MIL).

Parameters:

• t (Time array (Monte Carlo sweeps).)

• R_sk (Domain size from structure factor.)

• R_xi (Correlation length from G(r).)

• third_metric (Optional third metric array.)

• third_metric_label (Label for the third metric.)

• exponents (Dict of fitted exponents.)

• prefactors (Dict of fitted prefactors.)

• fit_mask (Mask used for fitting.)

• title (Figure-level suptitle.)

• filename (Output filename.)

• directory (Output directory.)

• y_label (Y-axis label for the left panel.)

• left_title (Title for the left subplot.)

• right_title (Title for the right subplot.)

utils.system_helpers.plot_temperature_sweep(*, temperatures: ndarray, avg_m: Sequence[float], avg_e: Sequence[float], susc: Sequence[float], spec_h: Sequence[float], title: str, filename: str, directory: str, entropy: ndarray | Sequence[float] | None = None, tau_int: ndarray | Sequence[float] | None = None) → None

Generate and save a standardized temperature sweep plot.

Displays magnetization, energy, susceptibility, and specific heat as functions of temperature. When entropy or tau_int are provided the layout expands to 3×2; otherwise the classic 2×2 is used. Saves the figure via save_plot().

Parameters:

• temperatures (Array of temperature values (x-axis).)

• avg_m (Average absolute magnetization per temperature point.)

• avg_e (Average energy per temperature point.)

• susc (Magnetic susceptibility per temperature point.)

• spec_h (Specific heat per temperature point.)

• title (Figure-level suptitle string (e.g. '2D Ising Model: Temperature Sweep (L=50)').)

• filename (Output filename passed to save_plot() (e.g. ‘temperature_sweep.png’).)

• directory (Output directory passed to save_plot() (e.g. ‘results/ising’).)

• entropy (Optional entropy per temperature point.)

• tau_int (Optional integrated autocorrelation time per temperature point.) – Reveals critical slowing down as a peak near T_c.

utils.system_helpers.save_plot(*, filename: str, directory: str = 'results', tight_layout: bool = True) → None

Save the current matplotlib plot to the results directory.

Parameters:

• filename (Name of the output file (e.g., 'plot.png').)

• directory (Output directory name.)

• tight_layout (Whether to apply plt.tight_layout() before saving.)

utils.system_helpers.setup_logging(*, level: int = 20, log_file: str | None = None) → Logger

Configure project-wide logging.

Parameters:

• level (Logging level (e.g., logging.INFO).)

• log_file (Optional path to a file to save logs to.)

Return type:

The configured logger instance.

Module contents