Observability

Domain observations as typed events with pluggable logger sinks. The pure core emits events as plain values; a Logger attached at the app boundary interprets them (the null default discards them). Solver diagnostics ride stdlib logging instead — two channels, two mechanisms.

Domain-observation events and the pluggable logger sink (effects-as-data observability).

This is the domain observations channel – distinct from stdlib logging, which carries solver diagnostics. The pure core emits typed events as plain values; a Logger attached at the app/ boundary interprets them. The core installs no sink and runs correctly with the NULL_LOGGER default, so it stays pure, testable, and vendor-free: Weights & Biases / Comet ML / CSV live only in logger backends at the boundary (diffBloch.app.loggers), never in the maths.

The name follows the PyTorch-Lightning convention (WandbLogger / CometLogger / CSVLogger plug into a common Logger); it is the experiment-tracking sink, orthogonal to the stdlib logging.Logger used for diagnostics. Every Event exposes a uniform (channel, measurements) surface – the Phoenix :telemetry “named event + measurements” idea – so a generic logger consumes any event without knowing its concrete type; adding an event never touches a logger. Callers wanting richer handling can still pattern-match the concrete dataclass.

Events fall into two families: a per-unit stream (a RotationScored per rotation, a RefinementStep per optimizer iteration – each carries a step) and a run-level aggregate (InferenceCompleted, RefinementCompleted – step is None).

class diffBloch.observability.CouplingSummary(measurements: Mapping[str, float])

Bases: object

Run-level summary of the plan the refinement/inference consumes (on the coupling channel).

The aggregate companion to the per-rotation RotationCoupling (step None vs a rotation index separates the two on one channel): measurements is diffBloch.preprocess.plan.summarize_plan() – the structure-factor support size/radius plus the coupling aggregates across rotations. Emitted once at the consumer boundary.

channel: ClassVar[str] = 'coupling'

measurements: Mapping[str, float]

property step: int | None

class diffBloch.observability.ConvergenceTrial(control: str, trial_index: int, pass_index: int, previous: float, candidate: float, r_factor: float, n_compared_hkl: int)

Bases: object

One comparison between consecutive numerical settings in a convergence sweep.

control: str

trial_index: int

pass_index: int

previous: float

candidate: float

r_factor: float

n_compared_hkl: int

property channel: str

property step: int | None

property measurements: Mapping[str, float]

class diffBloch.observability.ConvergencePassStarted(pass_index: int, g_max: float, sg_max: float, tilt_steps: int, r_factor_threshold: float, n_orientations: int)

Bases: object

Starting settings for one coordinated convergence pass.

pass_index: int

g_max: float

sg_max: float

tilt_steps: int

r_factor_threshold: float

n_orientations: int

channel: ClassVar[str] = 'convergence pass'

property step: int | None

property measurements: Mapping[str, float]

class diffBloch.observability.ConvergenceSweepStarted(control: str, pass_index: int)

Bases: object

Announcement emitted before one parameter sweep begins.

control: str

pass_index: int

channel: ClassVar[str] = 'convergence sweep'

property step: int | None

property measurements: Mapping[str, float]

class diffBloch.observability.DeviceSelected(requested: str, selected: str, cuda_available: bool)

Bases: object

Execution-device selection for an app run.

Device placement is an execution knob, not scientific provenance. This run-level event makes the selected backend visible to console/CSV/vendor sinks without entering config or checkpoint identity. Presentation wording stays with concrete logger backends; this event carries only stable selection data plus numeric measurements for generic metric sinks.

requested: str

selected: str

cuda_available: bool

channel: ClassVar[str] = 'device'

property step: int | None

property measurements: Mapping[str, float]

class diffBloch.observability.Event(*args, **kwargs)

Bases: Protocol

A named domain observation carrying numeric measurements.

channel is the event’s stable name – usually a class constant, but read as a plain attribute so an event may set it per instance (e.g. PlanStepCompleted uses the pipeline step’s name). measurements maps metric name to value; step is the optional position on the run’s x-axis (a rotation index, later a refinement iteration) or None for a run-level aggregate. Together they let a generic logger record and place any event, no per-type view.

property channel: str

property step: int | None

property measurements: Mapping[str, float]

class diffBloch.observability.ExperimentDeclared(name: str, structure: str, experimental_data: str, optimizer: str, seed_thicknesses_by_dataset: tuple[tuple[str, tuple[float, ...]], ...], integration_semiangles: tuple[float, ...], rocking_curve_sampling: int, dsg: float, rsg: float, solve_g_max: float, sg_max: float, absorption: bool, steps: int, learning_rate: float)

Bases: object

The run’s identity and its result-determining knobs, declared once before any compute.

The counterpart to ObjectiveManifest for everything the objective does not cover: which inputs are being refined and under which simulation/optimizer settings. A sink that writes a standalone artifact (the refinement report) needs this to describe the run without being handed the ExperimentConfig directly – which is what keeps such a sink an ordinary Logger rather than a component wired into the app’s orchestration.

Paths, the optimizer name, and the per-dataset seed-thickness declarations ride on the dataclass rather than in measurements, which stays flat-scalar for the generic backends – the same split ThicknessOptimized makes for its candidate grid.

channel: ClassVar[str] = 'experiment'

name: str

structure: str

experimental_data: str

optimizer: str

seed_thicknesses_by_dataset: tuple[tuple[str, tuple[float, ...]], ...]

integration_semiangles: tuple[float, ...]

rocking_curve_sampling: int

dsg: float

rsg: float

solve_g_max: float

sg_max: float

absorption: bool

steps: int

learning_rate: float

property step: int | None

property measurements: Mapping[str, float]

class diffBloch.observability.InferenceCompleted(n_rotations: int, n_evaluated: int, mean_r_obs: float)

Bases: object

The run-level aggregate, emitted once when run_inference finishes.

channel: ClassVar[str] = 'inference'

n_rotations: int

n_evaluated: int

mean_r_obs: float

property step: int | None

property measurements: Mapping[str, float]

class diffBloch.observability.Logger(*args, **kwargs)

Bases: Protocol

A sink for domain-observation events, attached at the app boundary.

A logger performs I/O (print, CSV row, wandb.log); the core only hands it values. The core defaults to NULL_LOGGER so it installs no sink and can run with none attached. Implement a single method to add a backend – see diffBloch.app.loggers.

report(event: Event) → None

class diffBloch.observability.MultiLogger(loggers: tuple[Logger, ...])

Bases: object

Fan each event out to several loggers (e.g. console and wandb at once).

loggers: tuple[Logger, ...]

report(event: Event) → None

class diffBloch.observability.NullLogger

Bases: object

The default sink: discards every event, so the core runs with no logger attached.

report(event: Event) → None

class diffBloch.observability.ObjectiveManifest(penalties: tuple[ObjectiveTerm, ...] = (), constraints: tuple[str, ...] = (), components: tuple[str, ...] = ())

Bases: object

What the refinement objective is composed of, declared once before the first step.

The refinement-side counterpart to the preprocess pipeline’s StepRecord provenance: penalties, constraints, and components are typed Python composition rather than config, so nothing else in a run states which of them are actually in play. This says so up front, before any compute – the “startup summary listing which restraints are active with which weights” that a bare per-epoch loss cannot provide.

Reporting the empty case is the point as much as the populated one: the default CLI path composes no penalties at all, and a run that says penalties: none is making a scientific fact legible rather than leaving it to be inferred from a missing line. measurements carries the three counts plus each penalty’s declared weight; the categorical names ride on the dataclass for a backend that pattern-matches it (as ThicknessOptimized does for its candidate grid).

This is a report, not an identity: it is deliberately not folded into refinement.lock or refinement_config_digest(). Refinement outputs are not checkpoint-reused, so hashing a composed-recipe axis would be identity infrastructure built ahead of the need for it.

channel: ClassVar[str] = 'objective'

penalties: tuple[ObjectiveTerm, ...] = ()

constraints: tuple[str, ...] = ()

components: tuple[str, ...] = ()

property step: int | None

property measurements: Mapping[str, float]

class diffBloch.observability.ObjectiveTerm(name: str, weight: float)

Bases: object

One declared soft-penalty term: the objective name it reports under and its weight.

name: str

weight: float

class diffBloch.observability.OrientationOptimized(rotation_index: int, score: float, residual: str, n_matched_hkl: int, n_trials: int, n_passes: int, pass_cap: int)

Bases: object

One rotation’s finished orientation search, emitted per rotation by optimize_orientation.

The fit is the long phase of a run (a coupled search solves ~100+ trials per rotation), so this is the progress stream that makes it observable: rotation_index is the original zero-based PETS rotation index, score the final orientation’s value under residual – the LossMetricsConfig name ("wr2"/"robs") that produced it, carried alongside so a consumer can label the number correctly (measurements keys on it directly, e.g. {"wr2": ...} or {"robs": ...}) rather than a generic, misleading wr2 field under a different residual. n_trials the number of trial orientations the search scored, n_passes scipy’s reported iteration count (the quantity NelderMeadSearch.max_iterations caps), and pass_cap that cap itself – carried per event so a plot can show each rotation’s headroom (n_passes vs pass_cap) and flag any rotation that ran to the cap. With workers > 1 events arrive in completion order (the plan itself stays ordered). The channel is shared with the step’s PlanStepCompleted summary line, like the refinement stream’s events.

channel: ClassVar[str] = 'orientation'

rotation_index: int

score: float

residual: str

n_matched_hkl: int

n_trials: int

n_passes: int

pass_cap: int

property step: int | None

property measurements: Mapping[str, float]

class diffBloch.observability.OrientationOptimizationStarted(total_rotations: int)

Bases: object

The rotation count optimize_orientation is about to search, emitted once before any of it.

Exists so a progress display can show a countdown (n_seen / total_rotations) against OrientationOptimized without needing to know the plan size in advance – the plan is only assembled deep inside the step itself. Deliberately a distinct channel from OrientationOptimized (not merely a different type) – a consumer such as EarlyAbortLogger that filters by event.channel alone must not mistake this for a per-rotation result.

channel: ClassVar[str] = 'orientation_started'

total_rotations: int

property step: int | None

property measurements: Mapping[str, float]

class diffBloch.observability.OrientationOptimizationSummary(n_orientations: int, mean_score: float, residual: str, unique_matched_hkl: int, unique_strong_hkl: int, unique_observed_hkl: int, total_trials: int, max_passes: int)

Bases: object

Aggregate statistics after every rotation’s orientation fit has completed.

unique_* counts are deduplicated distinct (h, k, l) counts across every rotation’s own set (unique_hkl_count()), not a sum of each rotation’s own count – a reflection re-observed (or matched) in more than one rotation is counted once, not once per rotation. unique_strong_hkl is “matched and I > 3*sigma in at least one rotation” – the same reflection can be strong in one rotation and weak in another, so this is a lower bound on “genuinely always weak,” not a claim every occurrence was strong.

n_orientations: int

mean_score: float

residual: str

unique_matched_hkl: int

unique_strong_hkl: int

unique_observed_hkl: int

total_trials: int

max_passes: int

channel: ClassVar[str] = 'orientation summary'

property step: int | None

property measurements: Mapping[str, float]

class diffBloch.observability.PlanSeeded(measurements: Mapping[str, float])

Bases: object

The Plan a preprocess pipeline is about to run on, summarised before the first step.

Exists so every PlanStepCompleted has a predecessor to be read against: a step’s counts are only a survival count if the incoming counts were reported too, and the seed is produced by from_experiment (or loaded from a checkpoint on resume) rather than by any step, so no PlanStepCompleted covers it. measurements is diffBloch.preprocess.plan.summarize_plan() of that incoming plan.

Deliberately a distinct channel from the per-step stream, and step is None: a consumer filtering on channel alone must not mistake the baseline for a stage result.

channel: ClassVar[str] = 'plan_seeded'

measurements: Mapping[str, float]

property step: int | None

class diffBloch.observability.PlanStepCompleted(channel: str, index: int, measurements: Mapping[str, float])

Bases: object

The Plan produced by one preprocess pipeline step, summarised as the recipe runs.

Unlike the other events its channel is the step name (select_beams, optimize_orientation, …), set per instance rather than a class constant – so the console reads optimize_orientation[4] n_orientations=55 ..., carrying the categorical step identity a fixed channel cannot. index is the step’s ordinal in the recipe (its step on the run’s x-axis); measurements is diffBloch.preprocess.plan.summarize_plan() of the resulting plan. Emitted only on a fresh preprocess run – a reused checkpoint runs no steps.

channel: str

index: int

measurements: Mapping[str, float]

property step: int | None

class diffBloch.observability.RecordingLogger(events: list[Event] = <factory>)

Bases: object

An in-memory logger that keeps every event (the doc’s “in-memory history” sink).

A shippable backend – useful for post-hoc inspection of a run and as the natural test double (assert on events instead of scraping a console). Unlike the vendor backends it performs no external I/O, so it stays vendor-free here beside NullLogger / MultiLogger.

events: list[Event]

report(event: Event) → None

class diffBloch.observability.RefinedRotationMetrics(rotation_index: int, wr2: float, r_obs: float, n_matched: int, is_validation: bool)

Bases: object

One rotation’s wR2/R_obs under the final refined model, emitted after the loop.

Distinct from RefinementOrientationStep, which is a per-epoch training diagnostic: this is the settled result, scored once on the best model by the reporting engine, so it covers every rotation including the held-out ones (is_validation marks those). The refinement loop cannot emit it – the loop only ever sees the training engine – so the app boundary emits it once the run has finished.

channel: ClassVar[str] = 'refined rotation'

rotation_index: int

wr2: float

r_obs: float

n_matched: int

is_validation: bool

property step: int | None

property measurements: Mapping[str, float]

class diffBloch.observability.RefinementCompleted(n_steps: int, best_step: int, best_loss: float, selection: Literal['training', 'validation']='training', reflection_counts: Mapping[str, int]=<factory>)

Bases: object

The refinement-run aggregate, emitted once when run_refinement finishes.

Shares the "refinement" channel with RefinementStep, separated from the stream by step (the iteration index vs None). The two are not the same quantity at different granularities: RefinementStep always reports the training objective, whereas best_loss is whichever objective actually selected the epoch. selection names that objective – "training" by default, or "validation" when run_refinement_model was given a held-out selection engine.

Because those two populations are not comparable, measurements emits best_loss under a different key per mode (best_training_loss / best_validation_loss) rather than one shared key plus a flag. A generic backend cannot then plot a train-selected and a val-selected run as one series: the key is absent instead of silently wrong.

channel: ClassVar[str] = 'refinement'

n_steps: int

best_step: int

best_loss: float

selection: Literal['training', 'validation'] = 'training'

reflection_counts: Mapping[str, int]

property step: int | None

property measurements: Mapping[str, float]

class diffBloch.observability.RefinementOrientationStep(iteration: int, rotation_index: int, wr2: float | None = None, r_obs: float | None = None, diff_loss: float | None = None)

Bases: object

One rotation’s wR2/R_obs/diffraction-loss diagnostics within a refinement epoch.

The per-orientation companion to RefinementStep’s epoch mean: run_refinement_model emits one of these per rotation per step only when its verbose flag is set (the “verbose refinement” reporting mode) – the per-rotation stream is n_orientations``x louder than the epoch summary, so it is a diagnosis tool, not the default reporting shape. ``iteration places it on the same x-axis as RefinementStep; rotation_index (the original zero-based PETS rotation index) is this event’s step, matching the per-rotation convention of RotationScored / OrientationOptimized.

channel: ClassVar[str] = 'refinement orientation'

iteration: int

rotation_index: int

wr2: float | None = None

r_obs: float | None = None

diff_loss: float | None = None

property step: int | None

property measurements: Mapping[str, float]

class diffBloch.observability.RefinementOutputsWritten(structure: str, artifacts: Mapping[str, str]=<factory>)

Bases: object

The refined artifacts are on disk – the run’s terminal event.

This is what lets a report be a plain Logger despite having to be written exactly once, after everything else, without adding a close/finalize method to the protocol: a sink that must finish at the end simply acts on this event. Putting the lifecycle in the stream keeps it observable (a RecordingLogger shows it) instead of implicit in a call order.

structure is the path to the written refined_structure.cif. A sink reads it back rather than being handed parsed values, so anything it reports about the structure is byte-consistent with the committed file by construction.

channel: ClassVar[str] = 'outputs'

structure: str

artifacts: Mapping[str, str]

property step: int | None

property measurements: Mapping[str, float]

class diffBloch.observability.RefinementStarted(total_steps: int)

Bases: object

The epoch budget run_refinement_model is about to run, emitted once before the loop.

Exists so a progress display can show a countdown (iteration / total_steps) against RefinementStep without needing the config’s refinement.steps passed in separately. Deliberately a distinct channel from RefinementStep (not merely a different type) – a consumer that filters by event.channel alone must not mistake this for a per-epoch result.

channel: ClassVar[str] = 'refinement_started'

total_steps: int

property step: int | None

property measurements: Mapping[str, float]

class diffBloch.observability.RefinementStep(iteration: int, loss: float, wr2: float | None = None, r_obs: float | None = None, diff_loss: float | None = None, objective_total: float | None = None, components: Mapping[str, ~collections.abc.Mapping[str, float]]=<factory>, n_rotations: int | None = None, n_wr2_evaluated: int | None = None, n_r_obs_evaluated: int | None = None)

Bases: object

One refinement epoch.

wr2/r_obs are always-computed reporting diagnostics (mean weighted-R2 / R_obs across orientations), free regardless of ExperimentConfig.loss_metrics (which decides what loss actually minimises, not what gets reported here) – so both are always shown. Contrast the preprocessing search’s events (OrientationOptimized / ThicknessOptimized), which report only the configured residual, since computing the other would cost an extra solve.

components carries each named objective term’s raw scientific diagnostic, its weight, and the contribution that weight produces, and measurements flattens every one of them to a "{term}/{field}" key so the generic backends (console, CSV, W&B, Comet) report a restraint’s state without knowing any term by name. A term that was never composed into the objective has no entry, so it cannot surface as a satisfied 0.0; that absence is the reportable fact, and it is why the flattening is unconditional rather than keyed on a fixed term list.

wr2/r_obs are means over the rotations that produced a finite score, so each carries its own denominator: n_rotations is how many the objective covered (the training set when a validation split is on) and n_wr2_evaluated/n_r_obs_evaluated how many actually entered each mean. They are separate counts because the two metrics are NaN-filtered independently – a rotation can contribute to one and not the other – and a mean whose denominator is implicit can improve simply by evaluating fewer rotations. Compare InferenceCompleted, which has always reported n_evaluated beside its mean.

channel: ClassVar[str] = 'refinement'

iteration: int

loss: float

wr2: float | None = None

r_obs: float | None = None

diff_loss: float | None = None

objective_total: float | None = None

components: Mapping[str, Mapping[str, float]]

n_rotations: int | None = None

n_wr2_evaluated: int | None = None

n_r_obs_evaluated: int | None = None

property step: int | None

property measurements: Mapping[str, float]

class diffBloch.observability.RotationCoupling(index: int, n_coupling_segments: int, n_tilts: int, max_tilts_per_segment: int, n_union_beams: int, max_beams_per_segment: int)

Bases: object

One rotation’s coupled solve geometry, emitted per rotation at the consumer boundary.

The shape the refinement loop repeats every step: n_coupling_segments coupled unions over n_tilts rocking-curve tilts, the widest union spanning max_tilts_per_segment tilts, the deduped union carrying n_union_beams beams, and the largest single segment max_beams_per_segment beams – the N of the dominant per-segment eigensolve. Fires on every run (fresh or checkpoint-reuse), so the coupling a long refine is about to chew on is legible before the first step.

channel: ClassVar[str] = 'coupling'

index: int

n_coupling_segments: int

n_tilts: int

max_tilts_per_segment: int

n_union_beams: int

max_beams_per_segment: int

property step: int | None

property measurements: Mapping[str, float]

class diffBloch.observability.RotationScored(index: int, r_obs: float, n_observed: int, n_beams: int)

Bases: object

One rotation’s forward-inference score, emitted per rotation by run_inference.

channel: ClassVar[str] = 'rotation'

index: int

r_obs: float

n_observed: int

n_beams: int

property step: int | None

property measurements: Mapping[str, float]

class diffBloch.observability.ThicknessOptimized(rotation_index: int, score: float, residual: str, thickness: float, candidate_thicknesses: tuple[float, ...], candidate_score: tuple[float, ...])

Bases: object

One rotation’s finished thickness grid search, emitted per rotation by optimize_thickness.

The thickness fit is the memory-heavy tail phase (each rotation scores the whole ThicknessGrid in one segmented solve), so like OrientationOptimized this makes it a progress stream rather than a silent block: rotation_index is the original zero-based PETS rotation index, score the baked thickness’s value under residual – the LossMetricsConfig name ("wr2"/"robs") that produced it, carried alongside so a consumer can label the number correctly (measurements keys on it directly) rather than a generic, misleading wr2 field under a different residual, and thickness that winning candidate (Angstrom). candidate_thicknesses/candidate_score carry the whole scored grid (same order, one entry per ThicknessGrid step) – deliberately excluded from measurements (which stays flat-scalar for the generic console/CSV/wandb/comet backends); a plotting backend such as ThicknessPlotLogger pattern-matches the concrete dataclass to read them. Emitted in plan order (the fit is sequential).

channel: ClassVar[str] = 'optimize_thickness'

rotation_index: int

score: float

residual: str

thickness: float

candidate_thicknesses: tuple[float, ...]

candidate_score: tuple[float, ...]

property step: int | None

property measurements: Mapping[str, float]

class diffBloch.observability.ThicknessOptimizationStarted(total_rotations: int)

Bases: object

The rotation count optimize_thickness is about to grid-search, emitted once up front.

Exists so a progress display can show a countdown (n_seen / total_rotations) against ThicknessOptimized without needing to know the plan size in advance – mirrors OrientationOptimizationStarted. Deliberately a distinct channel from ThicknessOptimized (not merely a different type) – a consumer such as EarlyAbortLogger that filters by event.channel alone must not mistake this for a per-rotation result.

channel: ClassVar[str] = 'thickness_started'

total_rotations: int

property step: int | None

property measurements: Mapping[str, float]

class diffBloch.observability.ThicknessProfile(form: str, min_thickness: float, max_thickness: float, rotation_indices: tuple[int, ...], alphas: tuple[float, ...], thicknesses: tuple[float, ...], label: str)

Bases: object

One dataset’s trained apparent-thickness curve, sampled at its rotations’ tilt angles.

Emitted once per composed thickness network after refinement – one event per dataset, each labeled by its inputs.exp_data ref. The whole curve rides on the dataclass as parallel tuples (one entry per rotation, in plan order) rather than as ~100 separate events or ~300 flat measurement keys – the shape ThicknessOptimized already uses for its candidate grid. measurements carries only the scalar summary.

channel embeds the label (the per-instance form the Event protocol anticipates) so metric sinks that key series on channel/name keep pooled datasets’ curves apart.

form: str

min_thickness: float

max_thickness: float

rotation_indices: tuple[int, ...]

alphas: tuple[float, ...]

thicknesses: tuple[float, ...]

label: str

property channel: str

property step: int | None

property measurements: Mapping[str, float]