Rigorous.CurrentStateUtils

Rigorous.CurrentStateUtils#

Rigorous.CurrentStateUtils.py

class JobInfo(id, iterations, best_fv, timestamp)#

Bases: tuple

Metadata for a single rigorous optimization job.

Attributes: id, iterations, best_fv, timestamp.

Create new instance of JobInfo(id, iterations, best_fv, timestamp)

best_fv#: Alias for field number 2

id#: Alias for field number 0

iterations#: Alias for field number 1

timestamp#: Alias for field number 3

class JobConvergence(id, evals, fvs, best_fv, best_sv, timestamps)#

Bases: tuple

Per-job convergence detail with full fv trajectory.

Attributes: id, evals (list of eval counts), fvs (list of fv values), best_fv, best_sv, timestamps (list of datetimes).

Create new instance of JobConvergence(id, evals, fvs, best_fv, best_sv, timestamps)

best_fv#: Alias for field number 3

best_sv#: Alias for field number 4

evals#: Alias for field number 1

fvs#: Alias for field number 2

id#: Alias for field number 0

timestamps#: Alias for field number 5

class ConvergenceInfo(jobs, best_fv, best_sv, best_job_id, spread, trend, n_jobs)#

Bases: tuple

Summary of convergence across all jobs.

Attributes: jobs (list of JobConvergence), best_fv, best_sv, best_job_id, spread, trend (‘improving’/’worsening’/’stable’), n_jobs.

Create new instance of ConvergenceInfo(jobs, best_fv, best_sv, best_job_id, spread, trend, n_jobs)

best_fv#: Alias for field number 1

best_job_id#: Alias for field number 3

best_sv#: Alias for field number 2

jobs#: Alias for field number 0

n_jobs#: Alias for field number 6

spread#: Alias for field number 4

trend#: Alias for field number 5

fv_to_sv(fv)#

Convert objective function value (fv) to Score Value (SV, 0-100 scale).

The raw optimizer objective fv is negative and nonlinear, making it hard to interpret. SV maps it to a human-readable 0-100 quality scale via:

SV = -200 / (1 + exp(-1.5 * fv)) + 100

Quality thresholds:: SV >= 80 : Good (green in plot_convergence) SV 60-80 : Fair (orange) SV < 60 : Poor (red)
Reference points:: fv = -3 -> SV ~= 98 fv = -1 -> SV ~= 64 fv = 0 -> SV = 0

Can also be applied to individual score values (not just the total fv) to identify which objectives are bottlenecks.

Parameters:: fv (float or array-like) – Objective function value(s).
Returns:: Score Value(s) on 0-100 scale.
Return type:: float or ndarray

construct_decomposition_from_results(run_info, **kwargs)#

load_rigorous_result(decomp, analysis_folder, jobid=None, rgcurve=None, debug=False)#

Load a rigorous optimization result from disk without launching a subprocess.

This is the static result viewer: it reads the saved parameter vector from a completed job’s callback.txt and reconstructs the Decomposition using a lightweight (for_split_only) optimizer that only knows how to split parameters.

Parameters:

decomp (Decomposition) – The initial (quick) decomposition that was used as the starting point for rigorous optimization. Provides ssd, num_components, model type, and baseline information.
analysis_folder (str) – Path to the analysis folder used during optimization (same value passed to optimize_rigorously(analysis_folder=...)).
jobid (str, optional) – Specific job id to load (subfolder name under optimized/jobs/). If None, loads the latest (last sorted) job.
rgcurve (RgCurve, optional) – Pre-computed Rg curve. When provided, skips the expensive per-frame Guinier fitting that ssd.xr.compute_rgcurve() would perform.
debug (bool, optional) – If True, reload modules from disk.

Returns:

A new Decomposition built from the optimized parameters.

Return type:

Decomposition

Examples

After re-running cells 1-4 to restore corrected, decomp, rgcurve:

from molass.Rigorous.CurrentStateUtils import load_rigorous_result

result = load_rigorous_result(decomp, "temp_analysis_scaffolded")
result.plot_components(rgcurve=rgcurve)

list_rigorous_jobs(analysis_folder)#

List completed rigorous optimization jobs on disk.

Scans the analysis_folder/optimized/jobs/ directory and returns metadata for each job that has a callback.txt file.

Parameters:: analysis_folder (str) – The same analysis_folder passed to optimize_rigorously().
Returns:: Each entry is a JobInfo(id, iterations, best_fv, timestamp) namedtuple. The list is sorted by job id.
Return type:: list of JobInfo

Examples

jobs = decomp.list_rigorous_jobs("temp_analysis_scaffolded")
for job in jobs:
    print(f"Job {job.id}: {job.iterations} iters, best fv={job.best_fv:.4f}")

has_rigorous_results(analysis_folder)#

Check whether any rigorous optimization results are available.

This is a lightweight filesystem check — it does not parse results. Use this to poll readiness before calling load_rigorous_result() or list_rigorous_jobs().

Parameters:: analysis_folder (str) – The same analysis_folder passed to optimize_rigorously().
Returns:: True if at least one job has written a callback.txt file.
Return type:: bool

Examples

if decomp.has_rigorous_results("temp_analysis"):
    result = decomp.load_rigorous_result("temp_analysis")

wait_for_rigorous_results(analysis_folder, timeout=600, poll_interval=5)#

Block until rigorous optimization results become available.

Polls the filesystem at regular intervals until at least one job has a parseable callback.txt file, or the timeout is reached.

Note: this checks list_rigorous_jobs() (not just file existence) to avoid a race condition where the file is created but not yet written.

Parameters:

analysis_folder (str) – The same analysis_folder passed to optimize_rigorously().
timeout (float, optional) – Maximum seconds to wait (default 600 = 10 minutes). Use 0 or None for no timeout.
poll_interval (float, optional) – Seconds between filesystem checks (default 5).

Returns:

True if results appeared before timeout, False if timed out.

Return type:

bool

Examples

decomp.optimize_rigorously(analysis_folder="temp", ...)
if Decomposition.wait_for_rigorous_results("temp"):
    result = decomp.load_rigorous_result("temp")
else:
    print("Timed out waiting for results.")

parse_sv_history(analysis_folder)#

Parse all callback.txt files and return the SV best-so-far trajectory.

This is the pure data-extraction layer shared by check_progress() and RunInfo.sv_history. No printing, no file writes.

Parameters:: analysis_folder (str) – Root folder for optimizer output (contains optimized/jobs/).
Returns:: sv_best_so_far: one SV value per accepted evaluation, accumulated as the running minimum. Empty list if no evaluations are recorded yet.
Return type:: list of float

parse_sv_history_per_job(analysis_folder)#

Parse callback.txt files and return per-job SV best-so-far trajectories.

Like parse_sv_history() but preserves job boundaries. Each job’s SV sequence starts fresh from its own first accepted evaluation but inherits the global running minimum from all preceding jobs.

Parameters:: analysis_folder (str) – Root folder for optimizer output (contains optimized/jobs/).
Returns:: Mapping from job id (e.g. '000', '001') to the list of global best-SV-so-far values recorded within that job. Jobs with no accepted evaluations are omitted. Empty dict if no evaluations are recorded yet.
Return type:: dict[str, list of float]

Examples

per_job = parse_sv_history_per_job(run_info.analysis_folder)
for job_id, svs in per_job.items():
    print(f"job {job_id}: {len(svs)} evals, best SV = {svs[-1]:.1f}")

check_progress(run_info_or_folder, label=None, write_snapshot=False)#

Read callback.txt(s) from all jobs and report best SV so far.

Re-runnable at any time while the optimizer is running or after it completes. Accepts either a RunInfo object or a plain folder path string, so it works even when the caller holds an older RunInfo instance from a previous session.

Parameters:

run_info_or_folder (RunInfo or str) – A RunInfo object (uses its analysis_folder attribute) or a plain analysis_folder path string.
label (str, optional) – Display label prefix. Defaults to the basename of the analysis folder.
write_snapshot (bool, optional) –
If True, write a compact JSON file to <analysis_folder>/optimized/progress_snapshot.json in addition to printing. The file can be read back later — even from a new AI session — via RunInfo.load_progress_snapshot() or json.load(open(...)):
```
_run_sub.check_progress(write_snapshot=True)
snap = _run_sub.load_progress_snapshot()  # dict
```
Default False (print-only, no side effects).

Returns:

Progress data dict when there are evaluations to report; None otherwise. Keys: label, n_evals, best_fv, best_sv, sv_best_so_far (full list), timestamp (ISO 8601 UTC).

Return type:

dict or None

Examples

# Via the RunInfo method (requires current class):
_run_sub.check_progress(label="subprocess")

# Persist for later AI-session retrieval:
snap = _run_sub.check_progress(write_snapshot=True)

# Via the standalone function (works with any RunInfo or a folder path):
from molass.Rigorous import check_progress
check_progress(_run_sub, label="subprocess")
check_progress("/path/to/analysis_folder")  # plain string also accepted

read_convergence_data(analysis_folder)#

Read convergence data from all completed rigorous optimization jobs.

Parameters:

analysis_folder (str) – The same analysis_folder passed to optimize_rigorously().

Returns:

Summary with per-job trajectories and cross-job diagnostics. Key fields for programmatic assessment:

best_fv — global best objective value
spread — difference between worst and best job’s best_fv
trend — 'improving', 'worsening', or 'stable'
jobs — list of JobConvergence with full fv trajectories

Return type:

ConvergenceInfo

Examples

info = Decomposition.plot_convergence("temp_analysis")
print(f"Best fv: {info.best_fv:.4f}, spread: {info.spread:.6f}")
print(f"Trend: {info.trend}")

plot_convergence(analysis_folder, ax=None, title=None)#

Plot convergence across rigorous optimization jobs.

Left subplot: best fv per job (cross-job trend). Right subplot: fv trajectory within each job (overlaid).

Parameters:

analysis_folder (str) – The same analysis_folder passed to optimize_rigorously().
ax (matplotlib Axes or array of Axes, optional) – If provided, plot into these axes (expects 2 axes). If None, creates a new figure with 2 subplots.
title (str, optional) – Figure title. If None, uses a default.

Returns:

The convergence data (same as read_convergence_data()), so the caller can inspect values programmatically.

Return type:

ConvergenceInfo

Examples

info = Decomposition.plot_convergence("temp_analysis")
print(f"Best: {info.best_fv:.4f}, trend: {info.trend}")

Rigorous.CurrentStateUtils

Contents

Rigorous.CurrentStateUtils#