Utilities

Internal helpers used by runners, timing export, and logging. Most users interact with these indirectly via run_* and preset dicts; the entries below are for custom workflows and debugging.

Parameter resolution

Helper functions for running SCGO campaigns.

This module provides utility functions used by the high-level API in scgo.runner_api to eliminate code duplication and improve maintainability.

scgo.utils.run_helpers.initialize_params(params)[source]

Initialize and merge params with defaults.

Handles None check and deep merge with default parameters.

Parameters:

params (dict[str, Any] | None) – User-provided parameters dict or None.

Return type:

dict[str, Any]

Returns:

Deep copy of params merged with defaults.

scgo.utils.run_helpers.initialize_ts_params(ts_params, *, system_type, surface_config=None, go_params=None)[source]

Initialize and merge TS params with get_ts_search_params() defaults.

When go_params is provided, calculator settings are aligned with the merged GO dict unless overridden in ts_params.

Return type:

dict[str, Any]

scgo.utils.run_helpers.diff_param_overrides(base, merged, *, prefix='')[source]

Return flat path -> value entries where merged differs from base.

Return type:

dict[str, Any]

scgo.utils.run_helpers.log_params_resolution(context, *, source_label, user_params, merged, base, verbosity)[source]

Log how user params were merged onto preset defaults.

Return type:

None

scgo.utils.run_helpers.get_calculator_class(calculator_name)[source]

Get calculator class by name.

Parameters:

calculator_name (str) – Name of the calculator (e.g., “MACE”, “EMT”).

Return type:

type

Returns:

Calculator class.

Raises:

ValueError – If calculator name is unknown or not available.

scgo.utils.run_helpers.validate_algorithm_params(algo_params, chosen_go, verbosity)[source]

Validate algorithm-specific parameters.

Parameters:
  • algo_params (dict[str, Any]) – Dictionary of algorithm-specific parameters.

  • chosen_go (str) – Name of chosen algorithm (‘simple’, ‘bh’, or ‘ga’).

  • verbosity (int) – Logging verbosity level (0=quiet, 1=normal, 2=debug, 3=trace).

Return type:

None

scgo.utils.run_helpers.resolve_auto_params(algo_params, composition, chosen_go)[source]

Resolve ‘auto’ parameter values.

Parameters:
  • algo_params (dict[str, Any]) – Dictionary of algorithm-specific parameters.

  • composition (list[str]) – List of atomic symbols.

  • chosen_go (str) – Name of chosen algorithm.

Return type:

dict[str, Any]

Returns:

Dictionary with resolved values to merge into global_optimizer_kwargs.

scgo.utils.run_helpers.resolve_diversity_params(algo_params, params, chosen_go)[source]

Resolve diversity parameters for fitness strategy.

Extracts diversity parameters from algorithm-specific params or top-level params, with algorithm-specific taking precedence. Raises ValueError if required diversity_reference_db is missing.

Parameters:
  • algo_params (dict[str, Any]) – Algorithm-specific parameter dictionary.

  • params (dict[str, Any]) – Top-level parameter dictionary.

  • chosen_go (str) – Name of chosen algorithm (for error messages).

Returns:

  • diversity_reference_db (required)

  • diversity_max_references (default: 100)

  • diversity_update_interval (default: 5)

Return type:

dict[str, Any]

Raises:

ValueError – If diversity_reference_db is not provided.

scgo.utils.run_helpers.prepare_algorithm_kwargs(algo_params, params, composition, chosen_go, *, system_type)[source]

Unified parameter preparation for algorithm execution.

Resolves “auto” parameter values, converts optimizer string names to classes, resolves fitness strategy, and filters out top-level keys that shouldn’t be passed to algorithms.

Parameters:
  • algo_params (dict[str, Any]) – Dictionary of algorithm-specific parameters from optimizer_params.

  • params (dict[str, Any]) – Full top-level parameter dictionary (for fitness strategy and diversity resolution).

  • composition (list[str]) – List of atomic symbols.

  • chosen_go (str) – Name of chosen algorithm (‘simple’, ‘bh’, or ‘ga’).

Return type:

dict[str, Any]

Returns:

Dictionary ready for direct algorithm execution.

scgo.utils.run_helpers.log_ts_configuration(ts_params, coerced_kwargs, *, verbosity, user_params=None, base=None)[source]

Log resolved transition-state search configuration.

Return type:

None

scgo.utils.run_helpers.log_configuration(params, chosen_go, cluster_formula, n_atoms, global_optimizer_kwargs, verbosity, *, user_params=None, params_base=None)[source]

Log final configuration.

Parameters:
  • params (dict[str, Any]) – Full parameter dictionary.

  • chosen_go (str) – Name of chosen algorithm.

  • cluster_formula (str) – Chemical formula string.

  • n_atoms (int) – Number of atoms.

  • global_optimizer_kwargs (dict[str, Any]) – Resolved algorithm parameters.

  • verbosity (int) – Logging verbosity level.

  • user_params (dict[str, Any] | None) – Original user dict before merge (for provenance logging).

  • params_base (dict[str, Any] | None) – Base defaults used for merge (defaults to get_default_params()).

Return type:

None

scgo.utils.run_helpers.cleanup_torch_cuda(logger=None)[source]

Release PyTorch CUDA caches when available, then run GC.

Return type:

None

Config aliases

Helpers for aliased dataclass construction kwargs.

scgo.utils.config_aliases.resolve_aliased_float(canonical_name, canonical, alias_name, alias, default)[source]

Pick canonical or alias (or default); error if both disagree.

Return type:

float

Combine atoms

Shared Atoms combine / surface-geometry helpers.

scgo.utils.combine_atoms.concatenate_inherit_cell_pbc(base, mobile)[source]

Concatenate base then mobile; inherit cell and PBC from base.

Return type:

Atoms

scgo.utils.combine_atoms.random_rotation_matrix(rng)[source]

Uniform random rotation (Haar on SO(3)) via random quaternion.

Return type:

ndarray

scgo.utils.combine_atoms.slab_surface_extreme(positions_or_atoms, axis, *, upper=True)[source]

Return max (or min) Cartesian coordinate along axis.

Return type:

float

scgo.utils.combine_atoms.top_layer_indices(positions, axis, *, thickness=2.5)[source]

Indices of atoms within thickness Å of the upper surface extreme.

Return type:

list[int]

Logging

Root logging setup, verbosity levels, and TRACE support for SCGO.

scgo.utils.logging.get_logger(name)[source]

Return a logger for name (typically __name__).

Return type:

Logger

scgo.utils.logging.configure_logging(verbosity=1, format_string=None, hpc_mode=None)[source]

Configure the root logger for the SCGO package.

SCGO targets HPC batch jobs: by default third-party loggers are suppressed aggressively. Set environment variable SCGO_LOCAL_DEV=1 to use milder suppression when hpc_mode is omitted, or pass hpc_mode=False explicitly.

Parameters:
  • verbosity (int) – Verbosity level (0=quiet, 1=normal, 2=debug, 3=trace).

  • format_string (str | None) – Custom format string. If None, uses default format.

  • hpc_mode (bool | None) – If True, suppresses more third-party logs (default when None, unless SCGO_LOCAL_DEV=1). If False, only WARNING+ for most libs.

Return type:

None

scgo.utils.logging.should_show_progress(verbosity)[source]

True when verbosity >= 1 (progress bars enabled for normal+).

Return type:

bool

scgo.utils.logging.log_debug_v(logger, message, *args, verbosity=1, min_verbosity=2)[source]

Log debug message if verbosity >= min_verbosity (default 2).

Uses lazy %-style formatting. Message is only formatted if it will be logged.

Parameters:
  • logger (Logger) – The logger instance.

  • message (str) – Format string for the message.

  • *args (object) – Arguments for the format string.

  • verbosity (int) – Current verbosity level (0-3).

  • min_verbosity (int) – Minimum verbosity to log (default 2 = DEBUG).

Return type:

None

scgo.utils.logging.log_info_v(logger, message, *args, verbosity=1, min_verbosity=1)[source]

Log info message if verbosity >= min_verbosity (default 1).

Uses lazy %-style formatting. Message is only formatted if it will be logged.

Parameters:
  • logger (Logger) – The logger instance.

  • message (str) – Format string for the message.

  • *args (object) – Arguments for the format string.

  • verbosity (int) – Current verbosity level (0-3).

  • min_verbosity (int) – Minimum verbosity to log (default 1 = INFO).

Return type:

None

scgo.utils.logging.log_warning_v(logger, message, *args, verbosity=1, min_verbosity=1)[source]

Log warning message if verbosity >= min_verbosity (default 1).

Warnings are typically always shown, but this allows conditional suppression.

Parameters:
  • logger (Logger) – The logger instance.

  • message (str) – Format string for the message.

  • *args (object) – Arguments for the format string.

  • verbosity (int) – Current verbosity level (0-3).

  • min_verbosity (int) – Minimum verbosity to log (default 1).

Return type:

None

scgo.utils.logging.log_error_v(logger, message, *args, verbosity=0, min_verbosity=0)[source]

Log error message if verbosity >= min_verbosity (default 0).

Errors are typically always shown, but this allows conditional suppression.

Parameters:
  • logger (Logger) – The logger instance.

  • message (str) – Format string for the message.

  • *args (object) – Arguments for the format string.

  • verbosity (int) – Current verbosity level (0-3).

  • min_verbosity (int) – Minimum verbosity to log (default 0 = always).

Return type:

None

scgo.utils.logging.log_exception_v(logger, message, *args, verbosity=1, min_verbosity=1, min_verbosity_for_traceback=2)[source]

Log exception with traceback if verbosity >= min_verbosity_for_traceback.

For unexpected errors, use logger.exception() directly instead. This helper is for handled exceptions where you want conditional traceback.

Parameters:
  • logger (Logger) – The logger instance.

  • message (str) – Format string for the message.

  • *args (object) – Arguments for the format string.

  • verbosity (int) – Current verbosity level (0-3).

  • min_verbosity (int) – Minimum verbosity to log error (default 1).

  • min_verbosity_for_traceback (int) – Minimum verbosity for traceback (default 2).

Return type:

None

Phase logging

Phase-oriented logging helpers for SCGO runs (headers, summaries, collectors).

scgo.utils.phase_logging.infer_verbosity(logger, explicit=None)[source]

Map an explicit verbosity or infer from the configured logger level.

Return type:

int

scgo.utils.phase_logging.log_phase_header(logger, title, *, verbosity, level=1)[source]

Emit a visible phase banner when verbosity >= level.

Return type:

None

scgo.utils.phase_logging.log_phase_subheader(logger, title, *, verbosity, level=1)[source]

Emit a lighter sub-phase banner (e.g. per generation).

Return type:

None

scgo.utils.phase_logging.format_count_summary(counts)[source]

Format outcome counts as label×N, ....

Return type:

str

class scgo.utils.phase_logging.InitDiagnosticsCollector[source]

Bases: object

Thread-safe accumulator for initialization fallbacks and placement failures.

classmethod reset()[source]
Return type:

None

classmethod record_fallback(used_strategy, from_strategy)[source]
Return type:

None

classmethod record_placement_failure(compact_line, detail_msg)[source]
Return type:

None

classmethod emit_summary(logger, *, verbosity, n_structures, prefix='Population initialization', extra='')[source]

Emit one INFO summary at v1+ and per-record DEBUG detail at v2+.

Return type:

None

scgo.utils.phase_logging.log_generation_offspring_summaries(logger, *, verbosity, job_results, total_jobs, created, n_offspring, attempts)[source]

Log v1 generation crossover/mutation/offspring summaries and v2 per-job detail.

Return type:

None

scgo.utils.phase_logging.format_offspring_outcome_line(index, *, failure_reason, desc, mutation_applied, validation_error)[source]

One-line DEBUG summary for a single offspring build attempt.

Return type:

str

Run directories and metadata

Run tracking and metadata management for SCGO campaigns.

This module provides functions for generating run IDs, saving and loading run metadata, and managing run-specific directory structures.

class scgo.utils.run_tracking.RunMetadataJSONEncoder(*, skipkeys=False, ensure_ascii=True, check_circular=True, allow_nan=True, sort_keys=False, indent=None, separators=None, default=None)[source]

Bases: JSONEncoder

JSON encoder: type objects become their __name__ (for params snapshots).

default(obj)[source]

Implement this method in a subclass such that it returns a serializable object for o, or calls the base implementation (to raise a TypeError).

For example, to support arbitrary iterators, you could implement default like this:

def default(self, o):
    try:
        iterable = iter(o)
    except TypeError:
        pass
    else:
        return list(iterable)
    # Let the base class default method raise the TypeError
    return super().default(o)
Return type:

Any

class scgo.utils.run_tracking.RunMetadata(run_id, timestamp, composition=None, formula=None, params=None)[source]

Bases: object

Metadata for a single run.

run_id: str
timestamp: str
composition: list[str] | None = None
formula: str | None = None
params: dict[str, Any] | None = None
to_dict()[source]

Convert metadata to dictionary.

Return type:

dict[str, Any]

classmethod from_dict(data)[source]

Create metadata from dictionary (ignores TS-parity provenance keys).

Return type:

RunMetadata

scgo.utils.run_tracking.generate_run_id()[source]

Generate timestamp-based run ID with microsecond granularity.

Returns:

run_YYYYMMDD_HHMMSS_ffffff Example: “run_20250124_143022_123456”

Return type:

str

scgo.utils.run_tracking.ensure_run_id(run_id, verbosity=0, logger=None)[source]

Ensure a run_id exists, generating one if needed and logging if appropriate.

This helper consolidates the common pattern of generating a run_id if None and logging it when verbosity is sufficient.

Parameters:
  • run_id (str | None) – Existing run_id or None to generate a new one.

  • verbosity (int) – Logging verbosity level (0=quiet, 1=normal, 2=debug, 3=trace).

  • logger – Optional logger instance. If None, will create one if needed.

Return type:

str

Returns:

Run ID (existing or newly generated).

scgo.utils.run_tracking.save_run_metadata(run_dir, run_id, metadata=None)[source]

Save run metadata to metadata.json file.

Parameters:
  • run_dir (str) – Directory where metadata file will be saved.

  • run_id (str) – Run ID for this run.

  • metadata (dict[str, Any] | None) – Optional dictionary of additional metadata to store. Common keys: composition, params, etc.

Return type:

None

scgo.utils.run_tracking.load_run_metadata(run_dir)[source]

Load run metadata from metadata.json file.

Parameters:

run_dir (str) – Directory containing metadata.json file.

Return type:

RunMetadata | None

Returns:

RunMetadata object if metadata file exists and is valid, None otherwise.

scgo.utils.run_tracking.get_run_directories(base_output_dir)[source]

Get list of all run directories in base output directory.

Parameters:

base_output_dir (str) – Base directory to search for run_* subdirectories.

Return type:

list[str]

Returns:

List of run directory paths (full paths), sorted by name.

scgo.utils.run_tracking.resolve_run_id_from_db_path(db_path, *, base_dir=None)[source]

Resolve GO run ID from a database path (run_* segment when present).

Return type:

str

scgo.utils.run_tracking.get_run_id_from_dir(run_dir)[source]

Extract run ID from directory name.

Parameters:

run_dir (str) – Directory path (may be full path or just name).

Return type:

str | None

Returns:

Run ID if directory name matches pattern, None otherwise.

Timing JSON

Timing summary logging and timing.json for GO, basin hopping, NEB/TS, and GO+TS.

GA/BH: set write_timing_json and detailed_timing in optimizer_params['ga'] (or bh) inside params/go_params. TS: set write_timing_json in ts_params.

GO timing is written at run level: {run_dir}/timing.json (alongside metadata.json and the optimizer database).

GO+TS pipeline rollup timing is written at the campaign root as go_ts_timing.json when write_timing_json=True in go_params and/or ts_params.

scgo.utils.timing_report.ga_relax_seconds_from_timings(timings)[source]

Total MLIP relax wall time for TorchSim GA (initial + offspring batches).

Return type:

float

scgo.utils.timing_report.relax_seconds_from_timings(timings)[source]

Return total relax/NEB wall time inferred from a timing payload.

Return type:

float

scgo.utils.timing_report.cpu_non_relax_seconds_from_timings(timings)[source]

Return non-relax CPU wall time (precomputed or total minus relax).

Return type:

float

scgo.utils.timing_report.log_timing_summary(logger, backend, timings_s, *, verbosity)[source]

Log a one-line timing summary when verbosity >= 1.

Return type:

None

scgo.utils.timing_report.write_timing_file(output_dir, payload, *, filename=None)[source]
Return type:

str

scgo.utils.timing_report.read_timing_file(path)[source]

Load a timing JSON file; return None if missing or unreadable.

Return type:

dict[str, Any] | None

scgo.utils.timing_report.resolve_run_timing_path(run_dir)[source]
Return type:

str

scgo.utils.timing_report.load_run_timing_payload(run_dir)[source]

Load timing.json from a run directory.

Return type:

dict[str, Any] | None

scgo.utils.timing_report.flatten_run_timing_payload(payload)[source]

Return a flat timing payload (legacy multi-trial documents are rejected).

Return type:

dict[str, Any]

scgo.utils.timing_report.build_timing_payload(*, backend, timings_s, run_id=None, extra=None)[source]

Build a structured timing document with provenance header and schema version.

Return type:

dict[str, Any]

scgo.utils.timing_report.build_run_timing_document(*, run_id, payload)[source]

Attach run_id to a single-run timing payload.

Return type:

dict[str, Any]

scgo.utils.timing_report.write_run_timing_file(run_dir, payload, *, run_id=None)[source]
Return type:

str

scgo.utils.timing_report.sum_neb_seconds_from_ts_results(ts_results)[source]

Sum per-pair neb_optimization_s values from TS result dicts.

Return type:

float

Output path helpers

Campaign output directory layout for global optimization and TS search.

scgo.utils.output_paths.formula_searches_dir(root, formula)[source]

Return {root}/{formula}_searches.

Return type:

Path

scgo.utils.output_paths.formula_ts_results_dir(root, formula)[source]

Return {root}/{formula}_ts_results.

Return type:

Path

scgo.utils.output_paths.resolve_campaign_root(output_dir)[source]

Resolve campaign root from output_dir.

When output_dir is None, use the current working directory. When output_dir ends with _searches, use its parent as the campaign root. Otherwise treat output_dir as the campaign root.

Return type:

Path

scgo.utils.output_paths.resolve_minima_dir(campaign_root, formula, *, searches_dir=None)[source]

Return the directory containing GO run_*/ minima databases.

When searches_dir is provided, minima are read from that path.

Return type:

Path

scgo.utils.output_paths.resolve_ts_campaign_paths(output_dir, path_key_formula, *, searches_dir=None)[source]

Return (campaign_root, minima_dir, ts_results_root) for TS search.

path_key_formula is the cluster/mobile formula used for sibling {formula}_searches and {formula}_ts_results directory names (without slab symbols for surface runs).

Return type:

tuple[Path, Path, Path]

scgo.utils.output_paths.resolve_go_searches_dir(output_dir, formula)[source]

Return the GO {formula}_searches/ directory for run_go.

When output_dir is provided, it is the searches directory itself. When output_dir is None, use {formula}_searches under CWD.

Return type:

Path

scgo.utils.output_paths.resolve_go_campaign_searches_dir(campaign_parent, formula)[source]

Return {parent}/{formula}_searches for run_go_campaign.

When campaign_parent is None, return None so run_go applies its own default searches path.

Return type:

Path | None

scgo.utils.output_paths.resolve_go_ts_pipeline_paths(campaign_root, formula)[source]

Return (searches_dir, ts_results_dir) under a GO+TS campaign root.

Return type:

tuple[Path, Path]