Skip to content

API — nefi.solve

Generated by scripts/gen_api_docs.py — do not edit by hand. Index

nefi.solve

Optimization: curriculum, solver, results, callbacks, post-processing, ensembles.

nefi.solve.balancing

Adaptive loss balancing (GradNorm-style) as a solver callback.

class GradNormBalancing(shares: Mapping[str, float] | None = None, every: int = 50, alpha: float = 0.5, ema: float = 0.5, anchor: str | None = None, bounds: tuple[float, float] = (...

Rebalance loss weights so weighted gradient norms follow target shares.

  • on_stage_start(self, solver, stage_idx, stage)
  • on_step(self, solver, state: StepState)

nefi.solve.batched

Batched multi-measurement solving: N independent problems in one optimization loop.

def batch_invert(problems: Sequence[Any], curriculum: Curriculum | None = None, *, device: str | torch.device = 'auto', dtype: torch.dtype = torch.float32, seeds: int | Sequ...

Solve N independent inverse problems in one batched optimization loop.

def batchable_reason(problem: Any, curriculum: Curriculum | None = None) -> str | None

None if :func:batch_invert can solve problem, else the reason why not.

nefi.solve.callbacks

Solver callbacks: logging, progress bars, checkpoints, plotting.

class Callback()

  • on_run_end(self, solver: Solver, result: Result)
  • on_run_start(self, solver: Solver)
  • on_stage_end(self, solver: Solver, stage_idx: int, stage: Stage, info: dict)
  • on_stage_start(self, solver: Solver, stage_idx: int, stage: Stage)
  • on_step(self, solver: Solver, state: StepState)

class CheckpointCallback(path: str | Path, every: int = 500)

Save field/operator state every every steps and at the end of each stage.

  • on_stage_end(self, solver, stage_idx, stage, info)
  • on_step(self, solver, state)

class FieldSnapshots(every: int = 100, name: str | None = None)

Keep detached CPU snapshots of the primary field every every steps (for animations).

  • on_step(self, solver, state)

class LoggingCallback(every: int = 100)

Log loss components every every steps through the nefi logger.

  • on_stage_end(self, solver, stage_idx, stage, info)
  • on_stage_start(self, solver, stage_idx, stage)
  • on_step(self, solver, state)

class ProgressBar(leave: bool = False)

tqdm progress bar over all curriculum steps.

  • on_run_end(self, solver, result)
  • on_run_start(self, solver)
  • on_step(self, solver, state)

class StepState(stage_idx: int, stage: Stage, step: int, global_step: int, lr: float, progress: float, total: float, components: dict[str, float], data_loss: float, elapsed...

StepState(stage_idx: 'int', stage: 'Stage', step: 'int', global_step: 'int', lr: 'float', progress: 'float', total: 'float', components: 'dict[str, float]', data_loss: 'float', elapsed: 'float', fields: 'dict[str, torch.Tensor] | None' = None, pred: 'torch.Tensor | None' = None, extra: 'dict[str, Any]' = )

nefi.solve.curriculum

Optimization curriculum: multiscale stages, frequency annealing, learning-rate schedules.

class Curriculum(stages: list[Stage] = <factory>, optim: OptimConfig = <factory>, early_stop_patience: int | None = None, early_stop_min_delta: float = 1e-06, discrepancy_ta...

Curriculum(stages: 'list[Stage]' = , optim: 'OptimConfig' = , early_stop_patience: 'int | None' = None, early_stop_min_delta: 'float' = 1e-06, discrepancy_tau: 'float | None' = None, restarts: 'int' = 1, time_budget_s: 'float | None' = None)

  • final_shape (property)
  • multiscale(shape: Sequence[int], n_stages: int = 2, steps: Sequence[int] | int = (3000, 7000), lr: float = 0.001, lr_decay: float = 0.5, min_size: int = 8, **stage_kw)... — Coarse-to-fine stages ending at shape (NeTMY: 32² → 64², lr halved in stage 2).
  • scaled(self, factor: float) -> Curriculum — Copy with every stage's step count multiplied by factor (smoke tests).
  • single(shape: Sequence[int] | None = None, steps: int = 2000, lr: float = 0.001, **kw) -> Curriculum
  • total_steps (property)

class OptimConfig(optimizer: str = 'adamw', weight_decay: float = 0.0001, grad_clip: float | None = 1.0, ema: float | None = None, betas: tuple[float, float] = (0.9, 0.999), ...

OptimConfig(optimizer: 'str' = 'adamw', weight_decay: 'float' = 0.0001, grad_clip: 'float | None' = 1.0, ema: 'float | None' = None, betas: 'tuple[float, float]' = (0.9, 0.999), lbfgs_history: 'int' = 20, lbfgs_max_iter: 'int' = 20, lr_mult: 'dict[str, float]' = )

class Stage(name: str = 'stage', shape: tuple[int, ...] | None = None, steps: int = 1000, lr: float = 0.001, lr_schedule: str = 'cosine', lr_min_ratio: float = 0.01, lr...

One curriculum stage (fixed resolution).

  • lr_at(self, step: int) -> float
  • progress_at(self, step: int) -> float — Annealing progress in [0, 1] (β/K).

nefi.solve.ensemble

Multi-seed ensembles: cheap uncertainty maps for per-measurement inversion.

class EnsembleResult(results: list[Result], mean: dict[str, torch.Tensor], std: dict[str, torch.Tensor], seeds: list[int] = <factory>)

EnsembleResult(results: 'list[Result]', mean: 'dict[str, torch.Tensor]', std: 'dict[str, torch.Tensor]', seeds: 'list[int]' = )

  • coefficient_of_variation(self, name: str | None = None) -> torch.Tensor

def ensemble(problem_factory: Callable[[], object], curriculum: Curriculum | None = None, seeds: Sequence[int] = (0, 1, 2), **solver_kw) -> EnsembleResult

Solve the same problem from several seeds and aggregate the fields.

nefi.solve.postprocess

Post-processing applied once after optimization.

class Clip(field: str | None = None, lo: float | None = 0.0, hi: float | None = None)

class EnergyScaleCorrection(field: str | None = None, homogeneity: float | None = None)

Energy-anchored scale correction (NeTMY Eq. 30, Prop. 1, Eq. 21).

class Postprocess()

Maps (fields, pred, problem, shape) -> (new_fields, info).

class ThresholdMask(threshold: float, field: str | None = None, mode: str = 'below', relative: bool = False, out_name: str | None = None)

Add a binary mask field <field>_mask = 1{x < thr} (mode="below") or > thr.

nefi.solve.refine

Reconstruction-side edge refinement: a short, physics-constrained sharpening stage.

class DoubleWell(field: str, values: Sequence[float], ramp: tuple[float, float] = (0.01, 1.0), name: str | None = None)

Cahn–Hilliard-style multi-well penalty mean Π_j ((x − v_j)/(v_{k−1} − v₀))².

  • energy(self, x: torch.Tensor) -> torch.Tensor
  • forward(self, ctx: Context) -> torch.Tensor

class MultiPhaseHead(values: Sequence[float], offsets: Sequence[float] | None = None, eps_start: float = 0.25, eps_end: float = 0.02, schedule: str = 'geometric', render: str = ...

k-phase head over one level-set function (multilayer level set).

  • eps(self, progress: float = 1.0) -> float — Interface softness ε (φ units) at progress.
  • extra_repr(self) -> str
  • indicators(self, phi: torch.Tensor, progress: float = 1.0) -> list[torch.Tensor] — Smoothed Heavisides H_ε(φ − c_j), j = 1 … k−1 (cell-averaged on grids).
  • phase(self, raw: torch.Tensor) -> torch.Tensor — Integer phase index Σ_j 1{φ > c_j} of every point.
  • transform(self, raw, others, progress: float = 1.0)
  • values(self) -> torch.Tensor — Current scalar phase values (k,) (clamped; free phases: their initial value).

class RefineReport(problem: str, mode: str, field: str, representation: str, render: str, accepted: bool, reason: str, levels_init: list[float], levels_final: list[float], fre...

Everything :func:refine_edges measured and decided (to_markdown(), to_dict()).

  • rows(self) -> list[list[Any]] — [quantity, smooth, refined] rows of the comparison table.
  • summary(self) -> str — One line: verdict, data fit and (with a GT) IoU / Edge-F1 before → after.
  • to_dict(self) -> dict[str, Any]
  • to_markdown(self) -> str

def cell_heaviside(s: torch.Tensor, eps: float, h: torch.Tensor | None = None) -> torch.Tensor

Smooth Heaviside σ(s/ε), or its average over cells spanning h units of s.

def data_fit(pred: torch.Tensor, measurement: Any) -> dict[str, float | None]

Data misfit of a prediction: rmse, rel_rmse and chi over observed entries.

def edge_metrics(pred: Any, gt: Any, *, tau: float | str | None = None, above: bool | str = 'auto', data_range: float | None = None, edge_threshold: float = 0.5, edge_dilate...

Contrast, segmentation and edge metrics of a reconstruction against a ground truth.

def estimate_levels(x: Any, levels: Any = 'auto', threshold: Any = 'auto', quantiles: tuple[float, float] = (0.5, 0.9), value_range: tuple[float, float] = (-inf, inf)) -> tuple...

Phase values v₀ < … < v_{k−1} and value-space thresholds τ₁ < … < τ_{k−1} of a smooth reconstruction.

def mass_matched_fraction(x: Any, lo: float, hi: float) -> float

Volume fraction of the hi phase that conserves the mean of x: f = clip((mean x − lo)/(hi − lo), 0, 1).

def multi_otsu(x: Any, k: int = 2, bins: int = 256) -> list[float]

Globally optimal k-class thresholds of a histogram (multi-level Otsu).

def noise_level(noise_std: Any) -> float | None

A measurement's noise level as one number: the scalar, or the mean of the positive entries of a per-entry noise_std (None when unknown).

def otsu_threshold(x: Any, bins: int = 256) -> float

Otsu's threshold of a field (maximal between-class variance of the value histogram).

def refine_defaults(instance: Any) -> dict[str, Any]

Refinement defaults of an instance: its refine_defaults attribute (dict or fn(cfg)) if set, else :data:REFINE_DEFAULTS by name, else {}.

def refine_edges(problem: Any, result: Result, *, mode: str = 'levelset', levels: Any = 'auto', field: str | None = None, threshold: Any = 'auto', quantiles: tuple[float, fl...

Sharpen the interfaces of a smooth reconstruction under the same operator and data.

def refine_instance(instance: Any, result: Result, measurement: Any, *, gt: Any = None, problem: Any = None, **kw: Any) -> tuple[Result, RefineReport]

:func:refine_edges with the instance's defaults (:func:refine_defaults) and metrics.

def refine_method(mode: str | None = None, name: str | None = None, **refine_kw: Any) -> Any

A benchmark :class:~nefi.bench.protocol.Method: the instance's own method followed by :func:refine_instance (defaults from :data:REFINE_DEFAULTS, refine_kw override).

def refine_run_output(run: Any, *, instance: Any = None, problem: Any = None, **kw: Any) -> Any

Refine a finished run: an :class:~nefi.instances.RunOutput (pass instance), a gallery entry / run (nefi.viz GalleryEntry / GalleryRun: instance and problem are taken from it) or anything with result, measurement, gt attributes.

def transition_band_check(smooth: torch.Tensor, refined: torch.Tensor, values: Sequence[float], band: float = 0.25, refined_above: Sequence[float] | None = None) -> tuple[float, list...

Does every refined interface lie inside the smooth field's transition band?

def volume_matched_threshold(x: Any, fraction: float, above: bool = True) -> float

Threshold τ such that a fraction of the voxels lies above it (below if above=False) — the level whose super-level set has a prescribed volume.

nefi.solve.result

Result container with save / load and a compact summary.

class Result(fields: dict[str, torch.Tensor], raw_fields: dict[str, torch.Tensor], pred: torch.Tensor, history: dict[str, list[float]], stage_results: list[dict[str, Any...

Output of :class:nefi.solve.Solver.

  • final(self, key: str) -> float | None
  • load(path: str | Path) -> Result
  • primary (property)
  • save(self, path: str | Path)
  • summary(self) -> str

nefi.solve.solver

The per-measurement optimizer: runs a :class:Curriculum on an :class:InverseProblem.

class Solver(problem, curriculum: Curriculum | None = None, *, device: str | torch.device = 'auto', dtype: torch.dtype = torch.float32, callbacks: Sequence[Callback] = (...

Run a curriculum of optimization stages on an inverse problem.

  • run(self) -> Result

def normalize_autocast(autocast: str | torch.dtype | None) -> torch.dtype | None

None / "bf16" / "fp16" (or the torch dtypes) → the autocast dtype.

def normalize_compile(compile: bool | str | None) -> str | None

False/None/"none" → None; True/"field" → "field"; "step".