coseda.ici package

Submodules

coseda.ici.build_early_break_from_log module

build_early_break_from_log.py

New implementation using the sidecar image_run_state.json created by propose_next_shifts.py.

Builds two streams:

  1. early_break.stream - For every (image,event) with at least one finite wRMSD, pick the run

    where that event reached its BEST (minimum) wRMSD.

    • Extract the corresponding chunk from stream_RRR.stream and write it.

    • This is what the orchestrator later renames to done.stream.

  2. only_done_events.stream - Same best-wRMSD rule, but only for events whose latest_status

    == (‘done’,’done’) in the sidecar.

coseda.ici.build_early_break_from_log.build_plans_from_state(events_state)[source]
From image_run_state[“events”], build:

  • plans_all: {HeaderKey -> best_run_n} for all events with ≥1 finite wRMSD

  • plans_done: {HeaderKey -> best_run_n} but only for events whose

    latest_status == (‘done’,’done’)

Here:

  • state key is “abs_path::event_id”

  • trials: [ [run, dx, dy, idx, wr], … ]

Parameters:

events_state (Dict[str, Dict])

Return type:

Tuple[Dict[Tuple[str, int], int], Dict[Tuple[str, int], int]]

coseda.ici.build_early_break_from_log.build_stream_for_plans(run_root, plans, out_path, workers, label)[source]

Given plans {HeaderKey -> best_run_n}, build a new stream file at out_path containing:

  • header cloned from any contributing run’s stream file

  • one chunk for each HeaderKey (from the selected run)

The file is rebuilt from scratch.

Parameters:

  • run_root (str)

  • plans (Dict[Tuple[str, int], int])

  • out_path (str)

  • workers (int)

  • label (str)

Return type:

str

coseda.ici.build_early_break_from_log.chunk_key_from_slice(lines_slice)[source]

Derive (abs_image_path, event_id) from a chunk’s lines.

Parameters:

lines_slice (List[str])

Return type:

Tuple[str, int] | None

coseda.ici.build_early_break_from_log.extract_chunks_for_run(run_root, rn, keys)[source]

Worker: open stream_{rn}.stream once, build index, and return {key: chunk_text}.

Parameters:

  • run_root (str)

  • rn (int)

  • keys (List[Tuple[str, int]])

Return type:

Dict[Tuple[str, int], str]

coseda.ici.build_early_break_from_log.load_state(state_path)[source]

Load image_run_state.json as produced by propose_next_shifts.py. Falls back to an empty state if missing or corrupted.

Parameters:

state_path (str)

Return type:

Dict

coseda.ici.build_early_break_from_log.main(argv=None)[source]
coseda.ici.build_early_break_from_log.map_chunk_ids_by_image_event(lines, bounds)[source]

Build mapping from (abs_path,event) to chunk index.

Parameters:

  • lines (List[str])

  • bounds (List[Tuple[int, int]])

Return type:

Dict[Tuple[str, int], int]

coseda.ici.build_early_break_from_log.parse_stream_chunks(stream_path)[source]

Return (bounds, lines, header) for a CrystFEL .stream file.

bounds: list[(start_idx, end_idx)] for each chunk lines: list of all lines in the file header: lines before the first “Begin chunk”

Parameters:

stream_path (str)

coseda.ici.build_early_break_from_log.run_stream_path(run_root, run_n)[source]
Parameters:

  • run_root (str)

  • run_n (int)

Return type:

str

coseda.ici.build_overlays_and_list module

build_overlays_and_list.py

From image_run_log.csv (latest run only), create per-source overlay HDF5 files (named with the next run number) and per-(source,event) lst files:

run_{next}/event_{EV6}_{HASH8}/lst_{EV6}_{HASH8}.lst -> “<overlay.h5> //EV”

Key fixes vs single-H5 version: - Disambiguate same event ids coming from different .h5 sources by appending a

short hash of the absolute source path to the event folder and lst file names.

  • Keep one overlay per source; write all proposed shifts for that source into it.

  • Persist overlay→original maps (JSON + TSV) for traceability.

Notes: - Skips entries whose next_* are “done”. - Overlays are created in runs/run_{next:03d}/ - Each overlay filename is <src_basename>_overlay_{next:03d}.h5

Requires overlay_elink.py (create_overlay, write_shifts_mm).

coseda.ici.build_overlays_and_list.collect_latest_numeric_proposals(entries, latest_run)[source]

From latest run only, collect numeric proposals per (real_path, event). Returns mapping: real_path -> { event_idx: (next_dx_float, next_dy_float) }

Parameters:

latest_run (int)

coseda.ici.build_overlays_and_list.ensure_overlay_for_run(src_path, run_dir, next_run)[source]

Create (or overwrite) an overlay for src_path under run_dir. Name: <basename>_overlay_{next:03d}.h5 Returns absolute overlay path.

Parameters:

  • src_path (str)

  • run_dir (str)

  • next_run (int)

Return type:

str

coseda.ici.build_overlays_and_list.load_state(state_path)[source]

Load the JSON sidecar produced by propose_next_shifts.py.

Expected structure:
{

“last_global_run”: int, “events”: {

“<abs_h5_path>::<event_id>”: {

“trials”: […], “latest_status”: [next_dx_str, next_dy_str], “last_run”: int

}

}

Parameters:

state_path (str)

Return type:

dict

coseda.ici.build_overlays_and_list.main(argv=None)[source]
Return type:

int

coseda.ici.build_overlays_and_list.parse_log(log_path)[source]
Returns:

list of tuples (type, payload)

type == ‘section’ -> payload = (real_path, event:int) type == ‘row’ -> payload = (run_n:int, dx:str, dy:str, idx:str, wr:str, ndx:str, ndy:str)

latest_run: int

Return type:

entries

Parameters:

log_path (str)

coseda.ici.build_overlays_and_list.tag_overlay_with_original(overlay_path, original_path)[source]

Optionally store the original path as an attribute in the overlay HDF5 file.

Parameters:

  • overlay_path (str)

  • original_path (str)

Return type:

None

coseda.ici.build_overlays_and_list.write_all_shifts(run_dir, next_run, proposals_by_src)[source]

For each source, write its proposed shifts to its overlay in run_dir. Returns: dict src_path -> overlay_path

Parameters:

  • run_dir (str)

  • next_run (int)

  • proposals_by_src (dict)

coseda.ici.build_overlays_and_list.write_event_lists(run_dir, proposals_by_src, overlay_paths)[source]
Create unique per-(source,event) folders + one-line .lst:

run_dir/event_{EV6}_{H8}/lst_{EV6}_{H8}.lst -> “<overlay.h5> //EV”

Returns number of lists written.

Parameters:

  • run_dir (str)

  • proposals_by_src (dict)

  • overlay_paths (dict)

Return type:

int

coseda.ici.build_overlays_and_list.write_overlay_mapping(run_dir, overlay_paths)[source]

Persist mapping files relating overlays to originals within the run folder. Writes:

  • overlay_to_original.json : {“/abs/overlay.h5”: “/abs/original.h5”, …}

  • overlay_to_original.tsv : tab-delimited table for quick inspection

Returns the path to the JSON file.

Parameters:

  • run_dir (str)

  • overlay_paths (dict)

Return type:

str

coseda.ici.create_run_sh module

coseda.ici.create_run_sh.main()[source]

coseda.ici.evaluate_stream module

evaluate_stream.py

Fast evaluator that preserves the correctness of your original parser: - Memory-maps the .stream and finds chunks via your Begin/End patterns. - Parses each chunk with your original regexes (incl. panel IDs). - Computes peak↔reflection matches per panel, then aggregates wRMSD. - Parallelizes per-chunk work (no heavy IPC; each worker receives only the chunk bytes).

Outputs (same names):

chunk_metrics_<run>.csv summary_<run>.txt parse_debug_<run>.txt

class coseda.ici.evaluate_stream.ChunkRow(image: 'str', event: 'str', det_dx_mm: 'float | None', det_dy_mm: 'float | None', indexed: 'int', wrmsd: 'float | None', n_matches: 'int', n_kept: 'int', reason: 'str')[source]

Bases: object

Parameters:

  • image (str)

  • event (str)

  • det_dx_mm (float | None)

  • det_dy_mm (float | None)

  • indexed (int)

  • wrmsd (float | None)

  • n_matches (int)

  • n_kept (int)

  • reason (str)

det_dx_mm: float | None
det_dy_mm: float | None
event: str
image: str
indexed: int
n_kept: int
n_matches: int
reason: str
wrmsd: float | None
coseda.ici.evaluate_stream.get_chunks(path)[source]
Parameters:

path (str)

coseda.ici.evaluate_stream.main(argv=None)[source]
coseda.ici.evaluate_stream.parse_chunk_text(b, mr, sg)[source]
Parameters:

  • b (bytes)

  • mr (float)

  • sg (float)

Return type:

ChunkRow

coseda.ici.extract_mille_shifts module

Extract per-frame estimates of two global parameters (e.g., x/y detector translations) from a Millepede-II C-binary file (e.g., produced by CrystFEL/indexamajig → mille).

Default global IDs: 1 (dx), 2 (dy). Override with –globals. Optionally scale results by a factor (e.g., –scale 17800).

Usage:

python extract_mille_shifts.py mille-data.bin -o per_frame_dx_dy.csv –scale 17800

coseda.ici.extract_mille_shifts.iter_records(path)[source]

Yield (rec_idx, glder, inder) for each event record in Millepede C-binary.

The file is organized as records, each starting with a 4-byte length word:

length_word = (#entries << 1) with sign indicating float size: >=0 → float32, <0 → float64

followed by nr floats (derivatives & residual/sigma values) and nr ints (labels), both aligned such that glder[k] pairs with inder[k].

Parameters:

path (Path)

coseda.ici.extract_mille_shifts.main()[source]
coseda.ici.extract_mille_shifts.per_frame_two_globals(records_iter, global_ids=(1, 2))[source]

For each record (frame), compute frame-wise estimates of two global parameters using local elimination via the Schur complement.

Returns a pandas.DataFrame with columns:

record, dx, dy, dx_err, dy_err, n_locals, n_meas

where dx/dy correspond to the first/second entries in global_ids.

coseda.ici.fix_stream_paths module

fix_stream_paths.py

Rewrite ‘Image filename:’ lines in an indexamajig .stream so they point to the original .h5 using the overlay_to_original mapping produced by build_overlays_and_list.py.

Now fully argparse-driven so it works for ANY run without editing the file.

USAGE EXAMPLES

# Preferred (explicit): python3 fix_stream_paths.py –run-dir /path/to/runs/run_007 –run 007

# Or, if you keep run-root and run number separately: python3 fix_stream_paths.py –run-root /path/to/exp_root –run 007

# Auto-detect the .stream inside a run folder (if exactly one stream_*.stream exists): python3 fix_stream_paths.py –run-dir /path/to/runs/run_007

# Operate in-place instead of writing *_fixed.stream: python3 fix_stream_paths.py –run-dir /path/to/runs/run_007 –run 007 –inplace

coseda.ici.fix_stream_paths.main(argv=None)[source]
Return type:

int

coseda.ici.no_run_prep_singlelist module

no_run_prep_singlelist.py Step 1 with “defaults-if-no-args”: - If launched with NO arguments, uses hardcoded defaults for root/geom/cell/h5 and a standard flag set. - If launched WITH arguments, behaves as a normal CLI and appends any flags after –. - Always writes:

<run-root>/runs_ts/run_000/lst_000.lst <run-root>/runs_ts/run_000/sh_000.sh

  • Does NOT execute indexamajig; only writes the exact command to sh_000.sh.

Default dataset (used only when no CLI args are provided):

default_root = “/Users/xiaodong/Desktop/simulations/MFM300-VIII_tI/sim_004” default_geom = default_root + “/MFM300-VIII.geom” default_cell = default_root + “/MFM300-VIII.cell” default_h5 = default_root + “/sim.h5”

coseda.ici.no_run_prep_singlelist.build_argparser()[source]
Return type:

ArgumentParser

coseda.ici.no_run_prep_singlelist.main(argv)[source]
Parameters:

argv (List[str])

Return type:

int

coseda.ici.orchestrator module

ici_orchestrator.py Orchestrate SerialED iterative runs using provided helper scripts. loop:

  • create overlays and list for next iteration

  • run indexamajig/xgandalf/integration

  • evaluate streams

  • update image_run_log.csv

  • propose next det shifts

  • summarize log

  • build early_break_from_log (one with all best index per image/event and one with only done image/events)

until done or max iterations reached

class coseda.ici.orchestrator.OrchestratorRunLogger(runs_dir)[source]

Bases: object

Context manager that:

  • creates runs/<timestamp>_orchestrator.log

  • tees stdout/stderr to that file

  • records start/end timestamps and total wall time

Parameters:

runs_dir (str)

class coseda.ici.orchestrator.TimestampingStream(real_stream)[source]

Bases: object

Wraps a stream and prefixes each full line with a wall-clock timestamp.

Works with arbitrary chunks (no assumption that writes end with ‘

‘).

flush()[source]
write(data)[source]
coseda.ici.orchestrator.all_next_done_for_latest(log_path, latest)[source]
Parameters:

  • log_path (str)

  • latest (int)

Return type:

bool

coseda.ici.orchestrator.detect_latest_run_from_log(log_path)[source]
Parameters:

log_path (str)

Return type:

int

coseda.ici.orchestrator.do_init_sequence(run_root, geom, cell, h5_sources, flags, params=None, jobs=2)[source]
Parameters:

  • run_root (str)

  • geom (str)

  • cell (str)

  • h5_sources (list)

  • flags (List[str])

  • params (dict | None)

coseda.ici.orchestrator.iterate_until_done(run_root, geom, cell, max_iters, flags, params=None, jobs=2, h5_sources=None)[source]
Parameters:

  • run_root (str)

  • geom (str)

  • cell (str)

  • max_iters (int)

  • flags (List[str])

  • params (dict | None)

  • h5_sources (list | None)

coseda.ici.orchestrator.latest_run(run_root)[source]
Parameters:

run_root (str)

Return type:

Tuple[int, str]

coseda.ici.orchestrator.list_run_numbers(run_root)[source]
Parameters:

run_root (str)

Return type:

List[int]

coseda.ici.orchestrator.main(argv=None, callbacks=None)[source]
coseda.ici.orchestrator.run_py(script, args, check=True)[source]
Parameters:

  • script (str)

  • args (List[str])

  • check (bool)

Return type:

int

coseda.ici.orchestrator.runs_dir(run_root)[source]
Parameters:

run_root (str)

Return type:

str

coseda.ici.propose_next_shifts module

propose_next_shifts.py

Implements a two-step proposal logic per event based on a global image_run_log.csv and a JSON sidecar (image_run_state.json).

Decision table (Option A):

IF indexed_this_run == 1:
IF dx/dy is refinable:

→ Step 2 (dxdy refinement)

ELSE:

→ Step 1 (Hillmap, weighted by previous successes/failures)

ELSE:
IF ever_indexed == 0:

→ Step 1 “pure” Hillmap (no positive hills; only exploration + penalties)

ELSE:

→ Step 1 weighted Hillmap (Boltzmann hills for successes, negative hills for fails)

The CSV schema (one section per image/event):

run_n,det_shift_x_mm,det_shift_y_mm,indexed,wrmsd,next_dx_mm,next_dy_mm,next_reason

where

  • indexed is an ever-indexed sticky flag (0/1)

  • wrmsd is per-run wRMSD (blank if not indexed for that run)

  • next_* are filled by this script for the NEXT run

coseda.ici.propose_next_shifts.load_state(state_path)[source]
Parameters:

state_path (str)

Return type:

Dict

coseda.ici.propose_next_shifts.main(argv=None)[source]
Return type:

int

coseda.ici.propose_next_shifts.parse_blocks(lines)[source]

Group into (header, block_lines) and detect latest run_n by scanning all data lines (not just from the end), because the log is grouped by event.

Parameters:

lines (List[str])

Return type:

Tuple[List[Tuple[str, List[str]]], int]

coseda.ici.propose_next_shifts.propose_event(step2algo, trials_sorted, R, rng, step1_A0, step1_hill_frac, step1_drop_frac, step1_candidates, step1_explore_floor, min_spacing, allow_spacing_relax, done_on_streak_successes=2, done_on_streak_length=5, noimprove_N=2, noimprove_eps=0.02, stability_N=3, stability_std=0.05, N_conv=3, recurring_tol=0.1, median_rel_tol=0.1, λ=0.8, event_abs_path='', beta=10.0)[source]

Return (next_dx_mm, next_dy_mm, reason). If done, returns (None, None, ‘done_*’).

trials_sorted: list of (run_n, dx, dy, idx_this_run, wrmsd{float|None})

Parameters:

  • step2algo (str)

  • trials_sorted (List[Tuple[int, float, float, int, float | None]])

  • R (float)

  • rng (numpy.random.Generator)

  • step1_A0 (float)

  • step1_hill_frac (float)

  • step1_drop_frac (float)

  • step1_candidates (int)

  • step1_explore_floor (float)

  • min_spacing (float)

  • allow_spacing_relax (bool)

  • done_on_streak_successes (int)

  • done_on_streak_length (int)

  • noimprove_N (int)

  • noimprove_eps (float)

  • stability_N (int)

  • stability_std (float)

  • N_conv (int)

  • recurring_tol (float)

  • median_rel_tol (float)

  • λ (float)

  • event_abs_path (str)

  • beta (float)

Return type:

Tuple[float | None, float | None, str]

coseda.ici.propose_next_shifts.read_log(path)[source]
Parameters:

path (str)

Return type:

List[str]

coseda.ici.propose_next_shifts.save_state(state_path, state)[source]
Parameters:

  • state_path (str)

  • state (Dict)

Return type:

None

coseda.ici.propose_next_shifts.update_csv_with_proposals(log_path, proposals)[source]
Modify the last row for each (h5,event) section in image_run_log.csv and fill:

next_dx_mm, next_dy_mm, next_reason

proposals[(abs_h5_path, ev)] = (next_dx, next_dy, reason) where next_dx/next_dy are strings (either numeric or “done”).

This version is O(N) in the number of lines in the log. It preserves the exact line ordering of the input file.

Parameters:

  • log_path (str)

  • proposals (Dict[Tuple[str, int], Tuple[str, str, str]])

Return type:

None

coseda.ici.propose_next_shifts.write_log(path, lines)[source]
Parameters:

  • path (str)

  • lines (List[str])

Return type:

None

coseda.ici.propose_next_shifts.wrmsd_median_convergence(successes_w, N=5, rel_tol=0.05)[source]

Detect convergence when median wRMSD stops changing significantly. Compares medians of two consecutive N-length windows; converged if relative change < rel_tol.

coseda.ici.propose_next_shifts.wrmsd_no_improvement(successes_w, N=5, eps=0.01)[source]

successes_w: list[(dx, dy, wrmsd)] for indexed trials.

Returns True if best wRMSD over the last N successes has not improved by at least eps relative to the best before that window.

coseda.ici.propose_next_shifts.wrmsd_recurring_convergence(successes_w, N=5, tol=0.1)[source]

Detects convergence when the best wRMSD is repeatedly reached. tol: allowed relative difference (e.g., 0.1 = 10%).

coseda.ici.propose_next_shifts.wrmsd_stability(successes_w, N=6, rel_std_tol=0.02)[source]

Returns True if the relative std of the last N successful wRMSDs is ≤ rel_std_tol. Uses std/mean.

coseda.ici.run_contract module

Canonical file layout helpers for COSEDA ICI runs.

coseda.ici.run_contract.canonical_stream_filename(run_root)[source]

Return the COSEDA-compatible final stream filename for an ICI run.

Parameters:

run_root (str)

Return type:

str

coseda.ici.run_contract.canonical_stream_path(run_root)[source]

Return the top-level stream path expected by COSEDA run consumers.

Parameters:

run_root (str)

Return type:

str

coseda.ici.run_contract.done_stream_path(run_root)[source]
Parameters:

run_root (str)

Return type:

str

coseda.ici.run_contract.early_break_stream_path(run_root)[source]
Parameters:

run_root (str)

Return type:

str

coseda.ici.run_contract.manifest_path(run_root)[source]
Parameters:

run_root (str)

Return type:

str

coseda.ici.run_contract.resolve_final_stream_path(run_root)[source]

Resolve the best available final stream for an ICI run.

Preference is converged output, then current best early-break output, then the canonical alias for backwards compatibility with existing COSEDA code.

Parameters:

run_root (str)

Return type:

str | None

coseda.ici.run_contract.run_name(run_root)[source]

Return the ICI run folder name.

Parameters:

run_root (str)

Return type:

str

coseda.ici.run_contract.run_root_path(run_root)[source]

Return the normalized top-level ICI run directory.

Parameters:

run_root (str)

Return type:

str

coseda.ici.run_contract.update_final_stream_alias(run_root)[source]

Maintain the canonical <ici_run_folder>.stream alias.

ICI keeps its native streams (early_break.stream and done.stream), while this alias makes the run look like a normal COSEDA indexing run to existing downstream tools.

Parameters:

run_root (str)

Return type:

str | None

coseda.ici.run_contract.write_run_manifest(run_root, *, geom=None, cell=None, h5_sources=None, flags=None, params=None, argv=None, phase=None, status=None, exit_code=None)[source]

Write or refresh the top-level ICI run manifest.

Parameters:

  • run_root (str)

  • geom (str | None)

  • cell (str | None)

  • h5_sources (list[str] | tuple[str, ...] | None)

  • flags (list[str] | tuple[str, ...] | None)

  • params (dict | None)

  • argv (list[str] | tuple[str, ...] | None)

  • phase (str | None)

  • status (str | None)

  • exit_code (int | None)

Return type:

str

coseda.ici.run_sh module

coseda.ici.run_sh.concat_streams(run_dir, run_str, ev_dirs)[source]
Parameters:

  • run_dir (str)

  • run_str (str)

  • ev_dirs (list[str])

Return type:

int

coseda.ici.run_sh.main()[source]
coseda.ici.run_sh.run_one_event(ev_dir)[source]

Run per-event indexing, then optional mille extraction. Return (event_dir, rc_index, rc_mille)

Parameters:

ev_dir (str)

Return type:

tuple[str, int, int]

coseda.ici.step1_hillmap_wrmsd module

step1_hillmap_wrmsd.py Variant of step1_hillmap that scales Gaussian “hill” amplitudes for successful trials inversely with their normalized wRMSD values. Lower wRMSD → higher hill amplitude → higher local sampling probability.

class coseda.ici.step1_hillmap_wrmsd.Step1Params(radius_mm, rng_seed, n_candidates, A0, hill_amp_frac, drop_amp_frac, explore_floor, min_spacing_mm, first_attempt_center_mm, allow_spacing_relax=True)[source]

Bases: object

Parameters:

  • radius_mm (float)

  • rng_seed (int)

  • n_candidates (int)

  • A0 (float)

  • hill_amp_frac (float)

  • drop_amp_frac (float)

  • explore_floor (float)

  • min_spacing_mm (float)

  • first_attempt_center_mm (Tuple[float, float])

  • allow_spacing_relax (bool)

A0
allow_spacing_relax
drop_amp_frac
explore_floor
first_attempt_center_mm
hill_amp_frac
min_spacing_mm
n_candidates
radius_mm
rng_seed
class coseda.ici.step1_hillmap_wrmsd.Step1Result(done, proposal_xy_mm, reason)[source]

Bases: object

Parameters:

  • done (bool)

  • proposal_xy_mm (Optional[Tuple[float, float]])

  • reason (str)

done
proposal_xy_mm
reason
class coseda.ici.step1_hillmap_wrmsd.Trial(x_mm, y_mm, indexed, wrmsd)[source]

Bases: object

Parameters:

  • x_mm (float)

  • y_mm (float)

  • indexed (int)

  • wrmsd (Optional[float])

indexed
wrmsd
x_mm
y_mm
coseda.ici.step1_hillmap_wrmsd.propose_step1(trials, params, beta=10.0)[source]

Hill-map proposal with wRMSD-weighted hills.

trials: list of Trial(x_mm, y_mm, indexed, wrmsd) params: Step1Params(…) beta: Boltzmann weight for wRMSD (larger = sharper preference for low wRMSD).

Parameters:
Return type:

Step1Result

coseda.ici.step2_dxdy module

step2_dxdy.py — STRICT Step-2 proposer that ONLY uses ‘per_frame_dx_dy.csv’ in the event directory.

Behavior: - Read event_dir/per_frame_dx_dy.csv - Expect columns ‘dx’ and ‘dy’ (case-insensitive) - Use the LAST row (dx, dy) verbatim as the next proposal - If anything is missing/invalid, return (None, None, <reason>) - Absolutely NO fallbacks, NO clamping, NO spacing repair.

Contract:
propose_step2_dxdy(successes_w, failures, tried, R, min_spacing_mm, event_dir, cfg)

-> (dx_mm, dy_mm, reason)

The extra arguments are ignored (kept for API symmetry).

class coseda.ici.step2_dxdy.Step2DxDyConfig(col_dx: 'str' = 'dx', col_dy: 'str' = 'dy')[source]

Bases: object

Parameters:

  • col_dx (str)

  • col_dy (str)

col_dx: str = 'dx'
col_dy: str = 'dy'
coseda.ici.step2_dxdy.propose_step2_dxdy(successes_w, failures, tried, R, min_spacing_mm, event_dir, cfg, **kwargs)[source]

STRICT: Only read per_frame_dx_dy.csv and return its last (dx, dy). No modifications, no fallbacks.

Parameters:
Return type:

Tuple[float | None, float | None, str]

coseda.ici.summarize_image_run_log module

summarize_image_run_log.py

Per-run summary of an iterative indexing/refinement pipeline.

For each run R in image_run_log.csv, prints:

  • Index rate of the FIRST run (using only trials with run_n == first_run)

  • Cumulative index rate up to PREVIOUS run (using best result per event)

  • Cumulative index rate up to CURRENT run (using best result per event)

  • wRMSD mean/median of FIRST run (best result per event at run_n == first_run)

  • Cumulative wRMSD mean/median up to PREVIOUS run (best result per event)

  • Cumulative wRMSD mean/median up to CURRENT run (best result per event)

  • Number of DONE events at this run, fraction of total frames, and their internal wRMSD mean/median (best result per done event)

  • Number of proposals issued at this run due to:

    • never-indexed events, and

    • refinement / Boltzmann search map (events that had indexed before)

class coseda.ici.summarize_image_run_log.Trial(run: 'int', indexed: 'bool', wrmsd: 'Optional[float]', next_dx: 'str', next_dy: 'str')[source]

Bases: object

Parameters:

  • run (int)

  • indexed (bool)

  • wrmsd (float | None)

  • next_dx (str)

  • next_dy (str)

indexed: bool
next_dx: str
next_dy: str
run: int
wrmsd: float | None
coseda.ici.summarize_image_run_log.cumulative_stats(events, upto_run, total_events_global, sidecar=None)[source]

Cumulative stats using all trials with run <= upto_run.

Returns:

(

idx_rate, wr_mean, wr_median, done_count, done_fraction, done_wr_mean, done_wr_median

)

Parameters:

  • events (Dict[Tuple[str, int], List[Trial]])

  • upto_run (int)

  • total_events_global (int)

coseda.ici.summarize_image_run_log.first_run_stats(events, first_run)[source]

Stats using only trials with run == first_run.

Returns:

(index_rate, wrmsd_mean, wrmsd_median)

Parameters:

  • events (Dict[Tuple[str, int], List[Trial]])

  • first_run (int)

coseda.ici.summarize_image_run_log.load_sidecar(run_root)[source]
Parameters:

run_root (str)

coseda.ici.summarize_image_run_log.main(argv=None)[source]
Return type:

int

coseda.ici.summarize_image_run_log.parse_log_rows(path)[source]

Parse grouped image_run_log.csv into an events dict:

events[(h5_path, event_id)] = [Trial(…), …]

Uses section headers of the form:

#/abs/path/to/file.h5 event 123

to identify which image/event the following rows belong to.

Parameters:

path (str)

Return type:

Tuple[Dict[Tuple[str, int], List[Trial]], List[int]]

coseda.ici.summarize_image_run_log.proposal_counts_for_run(events, run, sidecar)[source]

Count proposals for a given run, based on the final proposal recorded per event. Only the last proposal with rn == run is meaningful.

Classification (matching propose_next_shifts.py):
  • reason.startswith(“step1”)
    → Hillmap proposals (Step-1), including Boltzmann-weighted sampling

    when the event has ever been indexed.

  • reason.startswith(“dxdy”)

    → local dx/dy refinement (Step-2) around an indexed solution.

  • reason.startswith(“done”)

    → not a proposal (convergence / finished).

Parameters:

  • events (Dict[Tuple[str, int], List[Trial]])

  • run (int)

coseda.ici.update_image_run_log_grouped module

update_image_run_log_grouped.py

Ingest the latest chunk_metrics_###.csv from runs/run_### and append rows to runs/image_run_log.csv using the existing CSV schema: run_n,det_shift_x_mm,det_shift_y_mm,indexed,wrmsd,next_dx_mm,next_dy_mm

This version groups rows by image-event sections so that new run entries are inserted into the correct section rather than all being appended at the end. If a section (image-event) doesn’t exist yet, it is created at the end.

Duplicates (identical run lines within a given section) are avoided.

Ingest the latest chunk_metrics_###.csv from runs/run_### and append rows to runs/image_run_log.csv using the existing CSV schema: run_n,det_shift_x_mm,det_shift_y_mm,indexed,wrmsd,next_dx_mm,next_dy_mm,next_reason

Here:
  • indexed = ever-indexed sticky flag:

    0 → this event has never had a successful indexing (finite wRMSD) 1 → at least one run for this event has had a finite wRMSD

  • wrmsd = per-run wRMSD value (blank if no wRMSD for that run)

This version groups rows by image-event sections so that new run entries are inserted into the correct section rather than all being appended at the end.

Notes

  • Section headers are lines of the form: “#/abs/path/to/file.h5 event 123”

  • The CSV has a single header line at the very top.

  • “next_*” fields are intentionally left blank; a follow-up script will fill them.

coseda.ici.update_image_run_log_grouped.main(argv=None)[source]
Return type:

int

coseda.ici.update_image_run_log_grouped.resolve_real_source(h5_path)[source]

Return the real HDF5 path if images dataset is an ExternalLink; else the input path.

Parameters:

h5_path (str)

Return type:

str

Module contents

ICI orchestration helpers and scripts.