# SPDX-FileCopyrightText: 2026 John C. Travers and contributors
#
# SPDX-License-Identifier: GPL-3.0-or-later
"""The outer loop of a run: save planes in, saved planes out.
The inner loop -- many adaptive or replayed steps between two save planes -- lives
inside a backend and is compiled; this is the Python loop around it, which runs a few
hundred times at most (``docs/dev/architecture.md`` section 5). It asks the backend to
land on each save plane in turn, takes the observation and the statistics there, and
hands them to an output handler.
**What this module does not do.** It never conjugates a spectrum and never touches the
state between planes: the backend owns the step loop and the output handler owns the
sign convention on disk (``docs/dev/conventions.md`` section 2, boundary place 2). A
plane is carried across as a :class:`solprop.output.records.SavedPlane` in the internal
FFT convention, with modal amplitudes converted to true-power units.
Progress is reported through :class:`solprop.log.ProgressReporter` -- the run's summary
(:mod:`solprop.driver.summary`) before the first step, one call at the start, one per
attempted step from inside the backend, one per save plane and one at the end. The
protocol and the default :class:`solprop.log.LogReporter` live in :mod:`solprop.log`,
and are re-exported here because this is where a run reaches them. A bar shows the
summary and the closing line on its stream; the line reporter logs them, and
:func:`solprop.log.configure_logging` is how a script sees those records
(``docs/dev/architecture.md`` section 14).
The manual page is ``docs/guide/setup/running.md``.
"""
import logging
import warnings
from collections.abc import Mapping, Sequence
from typing import cast
import numpy as np
from solprop._interrupts import handled
from solprop.backends.populations import axis_refusal
from solprop.backends.protocol import Backend, BackendHistory, Observation
from solprop.backends.registry import statistics_refusal
from solprop.backends.session import RunSession, Start, statistics_unavailable
from solprop.exceptions import (
BackendError,
DriverError,
OutputError,
SolpropWarning,
)
from solprop.log import (
DEFAULT_PROGRESS_PERIOD,
LogReporter,
ProgressChoice,
ProgressReporter,
StopFlag,
resolve_reporter,
)
from solprop.monitor import BoundaryMonitor
from solprop.output.capability import (
capability_of,
declared_capability,
refusal_for,
withheld_accessors,
)
from solprop.output.hdf5 import HDF5Output
from solprop.output.metadata import RunMetadata
from solprop.output.provenance import BackendRecord
from solprop.output.records import (
MarchSnapshot,
RecipeRecord,
SavedPlane,
StepHistory,
StepStats,
)
from solprop.output.snapshots import SnapshotController
from solprop.output.store import (
CheckpointOutput,
FieldRetention,
FieldsChoice,
Output,
RetainingOutput,
field_retention_refusal,
)
from solprop.output.tdse_archive import TDSEArchiveWriter
from solprop.stats import StatsSpec
from solprop.stepper.low_storage import LowStorageRungeKutta
from solprop.stepper.multistep import Multistep
from solprop.translate.problem import PropagationProblem
from solprop.translate.routes import tdse_descriptor
from solprop.translate.run_request import HistoryPolicy, RunRequest
from solprop.translate.tdse import TDSEBinding
__all__ = [
"RESTART_RECORD",
"LogReporter",
"ProgressReporter",
"SessionFacts",
"check_snapshot_request",
"field_retention",
"history_of",
"open_run",
"plane_from_observation",
"retention_at",
"run_problem",
]
logger = logging.getLogger(__name__)
[docs]
def history_of(history: BackendHistory) -> StepHistory:
"""Return a backend's step history in the form an output handler stores.
The two records carry the same five columns, and two of them mean different things
on the two sides of the seam: the backend's ``z`` is the position at the *end* of
each attempt and its ``wall_time`` is the clock reading since the previous
observation, which is what a stepper measures cheaply, while
:class:`solprop.output.records.StepHistory` stores the position at the *start* of
each attempt and the time that one attempt took, which is what a reader wants per
row. The conversion is made here: the start is the end less the length, to
round-off, and the per-attempt times are the differences of the clock, the first
from zero.
Parameters
----------
history : solprop.backends.protocol.BackendHistory
The attempts a backend reports, each ending at its ``z``.
Returns
-------
solprop.output.records.StepHistory
The same attempts, each starting at its ``z``, with per-attempt times.
"""
return StepHistory(
z=history.z - history.dz,
h=history.dz,
norm=history.error_norm,
accepted=history.accepted,
wall_time=np.diff(history.wall_time, prepend=0.0),
)
[docs]
def plane_from_observation(
observation: Observation,
stats: Mapping[str, np.ndarray] | None = None,
*,
field: bool = True,
) -> SavedPlane:
"""Turn one observation and its statistics into the plane an output handler stores.
The seam's arrays are one trajectory's, as a stored plane is, so no shape changes
on the way (``docs/dev/conventions.md`` section 6). Nothing is conjugated --
the field stays in the internal FFT convention, and the handler converts it on the
way to storage (``docs/dev/conventions.md`` section 2). Nothing is rescaled either:
the field is whatever amplitude the observation carries, the backend's
vacuum-normalised one unless the caller has applied the interface conversion to
true-power amplitudes (a session's export with
``ExportRecipe(power=True, ...)``; see ``docs/dev/decisions.md``).
Parameters
----------
observation : solprop.backends.protocol.Observation
The state at a save plane, from
:meth:`solprop.backends.protocol.HostStepping.observe`.
stats : collections.abc.Mapping, optional
What :meth:`solprop.backends.protocol.HostStepping.stats` returned there, keyed
by statistic name. Empty when omitted.
field : bool, optional
Whether the plane keeps the field and its noise phase; ``False``, or an
observation that carries no field, builds a field-free plane (SEP 22), which
copies neither.
Returns
-------
solprop.output.records.SavedPlane
The position, the field, the statistics and the step history.
Examples
--------
>>> import numpy as np
>>> from solprop.backends.protocol import BackendHistory, Observation
>>> from solprop.driver.run import plane_from_observation
>>> history = BackendHistory(np.array([0.5]), np.array([0.5]), np.array([0.3]),
... np.array([True]), np.array([0.02]))
>>> observation = Observation(0.5, np.ones((1, 1, 4), complex), None, history, 1, 0)
>>> plane = plane_from_observation(observation, {"energy": np.ones((1, 1))})
>>> plane.z, plane.shape, plane.stats["energy"].shape
(0.5, (1, 1, 4), (1, 1))
"""
values = {} if stats is None else stats
rows = observation.step_stats
return SavedPlane(
z=observation.z,
field=observation.field if field else None,
stats={name: np.asarray(value) for name, value in values.items()},
step_history=history_of(observation.step_history),
step_stats=StepStats(z=rows.z, step=rows.step, values=dict(rows.values)),
noise_phase=observation.noise_phase if field else None,
)
[docs]
class SessionFacts:
"""A session's execution facts, as the service loop's records read a backend's.
The transitional output and record helpers (:mod:`solprop.driver.run`) read a
backend: its name, metadata, method history and in-flight state. A session that
owns its backend answers them through its protocol: its execution record, its
method-history size, and the in-place step a failure left in flight, which a
session that steps in place reports as ``in_flight`` (the native session's
low-storage replay, SEP 47 section 5.5); any other leaves none.
Parameters
----------
session : solprop.backends.session.RunSession
The session.
"""
def __init__(self, session: RunSession) -> None:
"""Read the session's facts through its protocol."""
self._session = session
record = session.execution_record()
self.name = str(record.get("backend", type(session).__name__))
self.reads_transverse_edge = bool(
getattr(session, "reads_transverse_edge", False)
)
[docs]
@property
def in_flight(self) -> tuple[float, float] | None: # numpydoc ignore=RT01
"""In-place step a failure left in flight, or ``None``.
Where the step started and its length, m, as the session reports it.
"""
return getattr(self._session, "in_flight", None)
[docs]
def march_snapshot(self) -> MarchSnapshot | None:
"""Return the session's method history at its pending committed state.
Returns
-------
solprop.output.records.MarchSnapshot or None
The session's snapshot
(:meth:`~solprop.backends.session.RunSession.march_snapshot`), or
``None`` without one or with no batch pending, for which the session's
:meth:`~solprop.backends.session.RunSession.pending_ticket` gives none.
"""
session = self._session
ticket = session.pending_ticket()
return None if ticket is None else session.march_snapshot(ticket)
[docs]
def march_snapshot_nbytes(self, z: float) -> int | None:
"""Return the size of the session's method-history snapshot, or ``None``.
Parameters
----------
z : float
The position, m.
Returns
-------
int or None
Bytes, or ``None`` without one.
"""
del z
return self._session.march_snapshot_nbytes()
[docs]
def march_export_refusal(self) -> str | None:
"""Return why the session's method history is not exported here, or ``None``.
Returns
-------
str or None
The session's own reason
(:meth:`~solprop.backends.session.RunSession.march_export_refusal`).
"""
return self._session.march_export_refusal()
[docs]
@handled() # SIGTERM and Ctrl-C stop a run between its commits
def run_problem(
problem: PropagationProblem,
backend: Backend,
output: Output,
*,
start: Start,
metadata: RunMetadata,
stats_spec: StatsSpec | None = None,
progress: ProgressReporter | ProgressChoice | None = None,
progress_period: float = DEFAULT_PROGRESS_PERIOD,
tdse_binding: TDSEBinding | None = None,
resume_history: bool | None = None,
fields: FieldsChoice = "all",
field_planes: Sequence[int] = (),
snapshots: SnapshotController | None = None,
tdse_archive: TDSEArchiveWriter | None = None,
boundary_monitor: BoundaryMonitor | None = None,
stop: StopFlag | None = None,
recipe: RecipeRecord | None = None,
restoration: Mapping[str, object] | None = None,
) -> None:
"""Run one translated problem to its save planes, writing each one out.
The loop of ``docs/dev/architecture.md`` section 5, usable without a
:class:`solprop.driver.simulation.Simulation`. The run starts a session on the
constructed backend from `start` (SEP 47); the handler is opened here, saved to
plane by plane, and closed in a ``finally`` so that an interrupted run leaves a
readable file. The backend stays open for the caller, who may start another run
on it.
The plane at ``problem.initial.z`` is saved first, with an empty step history, and
then each save plane in turn: propagate, observe, collect statistics, save. Steps
land on the planes exactly, so a saved field is an exact step endpoint and there
is no interpolation (``docs/dev/stepper.md`` section 3), unless the request asks
for dense output. With ``stepper.step_planes`` the march also stops at the end of
every ``step_planes``-th accepted step, which is saved too before the same plane
is asked for again. A plane whose propagation did not meet its tolerance is logged
as a warning naming that plane (``docs/dev/stepper.md`` section 8).
Parameters
----------
problem : solprop.translate.problem.PropagationProblem
The translated problem.
backend : solprop.backends.protocol.Backend
The backend stepping it, constructed for `problem` and its run's request
(:attr:`~solprop.backends.protocol.Backend.request`), whose save planes are
the itinerary; the run's own request is that one with this run's start,
retention, statistics, monitor and history
(:meth:`~solprop.translate.run_request.RunRequest.for_run`). The run loads
`start` into it and leaves it open.
output : solprop.output.store.Output
Where the planes go. Closed when the loop ends, whether or not it succeeded.
start : solprop.backends.session.Start
Where the run starts: :meth:`~solprop.backends.session.Start.entrance` for a
fresh run, or a resumed state with its ``plane_index``, the first save plane
strictly beyond it. A resume loads either the last regular plane or a separate
terminal checkpoint at its actual absolute position; the launch plane is
written when the handler holds no regular observations, so a resumed run never
stores it twice. The session takes the start's field.
metadata : solprop.output.metadata.RunMetadata
The record of the run, written when the handler opens.
:func:`solprop.driver.metadata.metadata_for` builds one.
stats_spec : solprop.stats.StatsSpec, optional
Which statistics to collect at each plane. Defaults to the backend's request's
own (section 8 of ``docs/dev/kernel-contract.md``).
progress : ProgressReporter, {'auto', 'bar', 'log', 'none'} or None, optional
Where progress goes. ``None`` takes the process default
(:func:`solprop.log.default_progress`, ``"auto"`` unless
:func:`solprop.log.configure_logging` was told otherwise): a
:class:`solprop.log.BarReporter` drawing one redrawing line on standard error
when that is a terminal or the process is a Jupyter kernel, and a
:class:`solprop.log.LogReporter` -- a periodic line, one line per plane and one
at the end, all at ``INFO`` -- otherwise, so redirected, piped and captured
output gets lines and never a bar. The four names force a choice; a reporter
is used as given.
progress_period : float, optional
Seconds between the line reporter's periodic lines, s; ``0`` leaves only the
plane and closing lines. A bar has its own redraw period and a given reporter
its own business, so it is ignored for both.
tdse_binding : solprop.translate.tdse.TDSEBinding or None, optional
Explicit live atomic runtime owned by the caller. Its ledgers and execution
facts are captured before the caller closes it. TDSE requires an output
supporting the separate checkpoint capability.
resume_history : bool or None, optional
Whether an output that can keep it (``save_march``) keeps the Adams march's
state at every save plane and terminal checkpoint, so that a resume continues
the history instead of starting it afresh. ``None`` keeps it while it is at
most :data:`~solprop.translate.run_request.MARCH_SNAPSHOT_AUTO_BYTES`.
fields : {'all', 'planes', 'none'}, optional
Which saved planes keep their field (SEP 22): every one (the default), those
of `field_planes`, or none. A plane that keeps none stores its position, its
statistics and its step history, and is not copied, converted or stored;
the march is the same either way. A terminal checkpoint, written when the
run stops, carries the field whatever this says, so a stopped run resumes.
field_planes : sequence of int, optional
Saved-plane indices for ``fields='planes'``, the entrance as 0 and negative
from the end.
snapshots : solprop.output.SnapshotController or None, optional
Requests immutable disk copies at committed accepted steps, each serviced at
the next accepted endpoint, after every interpolated plane its step crossed.
Requires HDF5 and all fields.
tdse_archive : solprop.output.tdse_archive.TDSEArchiveWriter or None, optional
Live archive to suspend and copy beside the optical writer.
boundary_monitor : solprop.monitor.BoundaryMonitor or None, optional
Watches the time window's edges (SEP 21): attached to the backend once its
launch is recorded and fed every committed step. Its record is written into
the execution record at every save plane without the per-plane rows, which
the closing record (complete, failed or out of budget) carries, and its removed
fraction is reported at the end. ``None`` watches nothing.
stop : solprop.log.StopFlag or None, optional
The run's own stop flag, any object with an ``is_set()`` method (SEP 41,
S17). It is polled before the first step, before every trial step (on a native
``Multistep()`` march, before every native call) and between save planes; once
it is set the run writes its state as for SIGTERM, tells the reporter through
``interrupted``, and raises
:class:`~solprop.exceptions.StopRequested`, which
:meth:`~solprop.driver.simulation.Simulation.run` turns into a return.
recipe : solprop.output.records.RecipeRecord or None, optional
The recipe the run was built from (SEP 41, S2), recorded in a plane store at
the first run start and never rewritten; ``None`` records none.
restoration : collections.abc.Mapping or None, optional
On a resume, what the caller restored of the method history, written into
every execution record under ``outcome.execution["restart"]`` beside what this
execution keeps; ``None`` records no restoration. When the start carries a
method history (``start.march``), this is the record of its restoration, and
a history the session cannot install is warned about and recorded as reset
instead.
Raises
------
DriverError
If the statistics given here disagree with the request's own about the
between-planes channel; if the record and the problem disagree about
whether this is a general modal run; if the start's ``plane_index`` is not a
plane of this problem; or if the field retention is not valid, names a plane
outside the run, asks for ``fields='planes'`` beside ``step_planes`` (whose
plane count is not known in advance), or is not ``'all'`` for an output that
cannot store field-free planes.
Examples
--------
>>> from solprop.backends.registry import select
>>> from solprop.backends.session import Start
>>> from solprop.driver.run import run_problem
>>> from solprop.driver.metadata import metadata_for
>>> from solprop.grid import EnvelopeGrid
>>> from solprop.nonlinear.gnlse import problem_from
>>> from solprop.output import MemoryOutput
>>> from solprop.pulses import SechPulse
>>> from solprop.stepper import PIController
>>> from solprop.translate.problem import assemble
>>> grid = EnvelopeGrid.from_limits(0.01, 835e-9, (500e-9, 2e-6), 1e-12)
>>> gnlse = problem_from([0.0, 0.0, -1.183e-26], omega_ref=grid.omega0,
... gamma=0.11, shock=False)
>>> start = SechPulse(wavelength=835e-9, width=100e-15, peak_power=9.8).build(grid)
>>> problem, request = assemble(grid, gnlse.mode, [gnlse.kerr], start,
... controller=PIController(rtol=1e-6, atol=1e-16),
... save_planes=[0.005, 0.01], dz_max=0.01, dz_init=1e-3,
... density=1.0, shock=False, prefactor_index=1.0)
>>> backend = select("numpy_ref", problem, request, grid=grid, modes=[gnlse.mode])
>>> handler = MemoryOutput()
>>> run_problem(problem, backend, handler, start=Start.entrance(problem),
... metadata=metadata_for(problem, request, grid, modes=[gnlse.mode]))
>>> backend.close()
>>> handler.n_saved, handler["z"].tolist()
(3, [0.0, 0.005, 0.01])
"""
prepared = backend.request
spec = prepared.observation.stats if stats_spec is None else stats_spec
planes = np.asarray(prepared.observation.save_planes, dtype=np.float64)
start_plane = start.plane_index
_refuse_unrunnable(
problem, prepared, spec, start_plane, planes.size, metadata=metadata
)
# The backend's door for a spec handed here rather than the problem's own, which
# `select` already checked (SEP 23: a backend without a statistic refuses it).
refusal = statistics_refusal(backend.name, spec.names)
if refusal is not None:
raise DriverError(refusal)
retention = field_retention(prepared, fields, field_planes)
_check_retainable(output, retention)
check_snapshot_request(prepared, output, snapshots, resume_history, fields)
if tdse_descriptor(problem.route) is not None and not isinstance(
output, CheckpointOutput
):
raise DriverError(
"TDSE output must support terminal checkpoints and call ledgers"
)
reporter = resolve_reporter(
progress,
period=progress_period,
n_steps=_replayed_steps(prepared, start_plane),
# The low-storage march records 0.0, "not estimated" (SEP 24 section 6.3).
estimated=not isinstance(prepared.integration.method, LowStorageRungeKutta),
)
# The session and its service loop (SEP 47): imported here, since the service
# loop builds on this module's output transactions.
from solprop.backends.sessions import start_session # noqa: PLC0415
from solprop.driver.session_loop import ServiceContext, serve # noqa: PLC0415
request = prepared.for_run(
stats=spec,
retention=retention,
start_plane=start_plane,
monitor=boundary_monitor,
history=HistoryPolicy.from_choice(resume_history),
snapshots=snapshots is not None,
)
logger.debug(
"running %d save planes from index %d, collecting %d statistics",
planes.size,
start_plane,
len(spec.names),
)
march = start.march
session = start_session(
backend,
problem,
request,
start,
started=reporter.started,
stop=stop,
owns=False,
)
try:
if march is not None and restoration is not None:
restoration = _restored(
march, restoration, getattr(session, "march_refusal", None)
)
serve(
session,
output,
ServiceContext(
problem=problem,
request=request,
metadata=metadata,
reporter=reporter,
retention=retention,
resume_history=resume_history,
snapshots=snapshots,
tdse_binding=tdse_binding,
tdse_archive=tdse_archive,
recipe=recipe,
start_plane=start_plane,
stop=stop,
restoration=restoration,
),
)
finally:
session.close()
def _restored(
march: MarchSnapshot,
restoration: Mapping[str, object],
refusal: str | None,
) -> Mapping[str, object]:
"""Return a resume's restoration record once its session tried the history.
The caller hands the session the method history it decided to continue, with the
record of a successful restoration; a history the session could not install
(another engine's rows, another build's shapes) starts afresh, which is warned
about once and recorded as a reset (SEP 47 section 5.5).
- ``refusal``: why the session could not install it, or ``None`` when it did.
"""
if refusal is None:
logger.info(
"continuing the Adams history at order %d, %d samples",
march.order,
len(march.identities),
)
return restoration
# Asked to continue a history this backend cannot install: the run is still
# correct to the integrator's tolerance, but it does less than was asked, so say so.
warnings.warn(
f"the Adams history starts afresh: {refusal}", SolpropWarning, stacklevel=4
)
return {**restoration, "state": "reset", "reason": "incompatible_representation"}
[docs]
def check_snapshot_request(
request: RunRequest,
output: Output,
snapshots: SnapshotController | None,
resume_history: bool | None,
fields: FieldsChoice,
) -> None:
"""Refuse unsupported snapshot policies before allocating native resources.
No backend is refused by name: every backend's fused and exponential Adams
engines, the native backend's resident one included, export their history
(SEP 29 section 6.6).
Parameters
----------
request : solprop.translate.run_request.RunRequest
The run's request, whose method decides what a snapshot needs.
output : Output
Requested optical writer.
snapshots : SnapshotController or None
Optional snapshot request controller.
resume_history : bool or None
Whether the caller explicitly disabled Adams history retention.
fields : FieldsChoice
Saved-field retention policy.
Raises
------
DriverError
If snapshots are requested outside their supported scope.
"""
if snapshots is None:
return
if not isinstance(snapshots, SnapshotController):
raise DriverError("snapshots must be a SnapshotController or None")
if not isinstance(output, HDF5Output):
raise DriverError("snapshots require HDF5Output")
output._refuse_readonly()
# A snapshot is taken at the next accepted endpoint, persisted as a running
# checkpoint when no plane is saved there, so it needs no step planes (SEP 47
# section 5.5).
# Interpolated output is admitted: every session takes a snapshot at an accepted
# endpoint only, once the planes its step crossed are observed, so a checkpoint
# never needs the step's extension to reconstruct a plane (SEP 47 section 5.5).
if fields != "all":
raise DriverError("snapshots require fields='all'")
if isinstance(request.integration.method, Multistep) and resume_history is False:
raise DriverError(
"snapshots require Adams history; resume_history=False is incompatible"
)
[docs]
def field_retention(
request: RunRequest, fields: object, field_planes: Sequence[int]
) -> FieldRetention:
"""Return a run's field retention, resolved against its saved planes.
The driver's door to :func:`solprop.output.store.field_retention_refusal`, the
validator :class:`~solprop.driver.settings.SimulationSettings` and
:func:`~solprop.driver.scan.run_scan` share.
Parameters
----------
request : solprop.translate.run_request.RunRequest
The run's request: its save planes, and the entrance, are the planes.
fields : object
``"all"``, ``"planes"`` or ``"none"``.
field_planes : collections.abc.Sequence of int
Saved-plane indices for ``"planes"``, negative from the end.
Returns
-------
solprop.output.store.FieldRetention
The rule, its indices counted from the entrance.
Raises
------
DriverError
If the request is not valid, names a plane outside the run, asks for
``fields='planes'`` while ``step_planes`` saves an unknown number of planes.
Examples
--------
>>> from solprop.driver import Simulation
>>> from solprop.driver.run import field_retention
>>> simulation = Simulation.from_gnlse(
... 0.11, 0.01, [0.0, 0.0, -1.183e-26], wavelength=835e-9,
... wavelength_limits=(500e-9, 2e-6), trange=1e-12,
... width=100e-15, peak_power=9.8, shape="sech", n_save=3,
... )
>>> field_retention(simulation._run_request(), "planes", [-1]).field_planes
(2,)
"""
refusal = field_retention_refusal(fields, field_planes)
if refusal is not None:
raise DriverError(refusal)
observation = request.observation
if fields == "planes" and observation.step_planes:
msg = (
"fields='planes' names saved planes by index, and step_planes saves the "
"march's own steps as well, so the plane count is not known in advance; "
"keep fields='all' or 'none' beside step_planes"
)
raise DriverError(msg)
try:
return FieldRetention(
cast("FieldsChoice", fields), tuple(int(index) for index in field_planes)
).resolved(observation.save_planes.size + 1)
except OutputError as error:
raise DriverError(str(error)) from error
def _check_retainable(output: Output, retention: FieldRetention) -> None:
"""Refuse a field-free rule for an output that cannot hold one.
Checked before the output is opened. An output can hold one when it has ``retain``
(:class:`~solprop.output.store.RetainingOutput`).
"""
if retention.fields != "all" and not isinstance(output, RetainingOutput):
msg = (
f"this run asks for {retention.describe()} and its output "
f"{type(output).__name__} cannot store field-free planes: give it a "
"retain method (RetainingOutput), or use a MemoryOutput or an HDF5Output, "
"or fields='all'"
)
raise DriverError(msg)
[docs]
def open_run(
output: Output,
metadata: RunMetadata,
backend: Backend | SessionFacts,
binding: TDSEBinding | None,
) -> CheckpointOutput | None:
"""Open a run's output with the runtime facts outside its scientific identity.
A run refused after its resources were prepared and before its first step (a
failed haCC consistency check, SEP 20 section 5.4) is recorded by this alone: the
record, with the binding's provenance, and no plane or terminal checkpoint, so
it cannot be mistaken for a propagation state.
Parameters
----------
output : solprop.output.store.Output
The handler, not yet open.
metadata : solprop.output.metadata.RunMetadata
The run's record, without its backend.
backend : solprop.backends.protocol.Backend or SessionFacts
What executes the run, whose metadata the record's backend section takes.
binding : solprop.translate.tdse.TDSEBinding or None
A TDSE response's binding, whose metadata the record carries too.
Returns
-------
solprop.output.store.CheckpointOutput or None
The handler's checkpoint interface, or ``None`` for a handler without one.
"""
# A session that holds its monitor from the start (the native session) reports
# the launch's zeros here. The execution record carries the monitor instead
# (`_record_execution`), as a group's points do (`_scan_batch`); a scan writes the
# opening record into every point's metadata.
executed = {
name: value
for name, value in backend.metadata().items()
if name != "boundary_monitor"
}
if binding is not None:
executed = {**executed, "tdse": dict(binding.metadata())}
output.open(metadata.with_backend(BackendRecord.from_metadata(executed)))
return output if isinstance(output, CheckpointOutput) else None
RESTART_RECORD = "restart"
"""The execution record's entry for what a restart keeps and restores (SEP 47 5.5).
Optional, under the output compatibility rules: ``"retention"``, what the run kept of
its method history at the record's position (``_MarchKeeper``), and, in a resumed
execution, ``"restoration"`` (what it restored,
``solprop.driver.resume._continue_march``) and ``"monitor"`` (how
the boundary monitor was restored). Each decision is a mapping with a ``"state"`` and
a ``"reason"``. A file written before this record has none, which a reader takes as
unknown, never as a reason (``retention_at``).
"""
[docs]
def retention_at(
execution: Mapping[str, object] | None, z: float
) -> Mapping[str, object] | None:
"""Return the method-history retention an execution record states at `z`.
Parameters
----------
execution : collections.abc.Mapping or None
An execution record (:attr:`solprop.output.records.RunOutcome.execution`).
z : float
The position a resume starts from, m.
Returns
-------
collections.abc.Mapping or None
The retention decision recorded there, or ``None`` when the record states
none at `z`: a file written before the restart record, or a position where no
method history was persisted.
"""
if execution is None:
return None
restart = execution.get(RESTART_RECORD)
if not isinstance(restart, Mapping):
return None
retention = cast("Mapping[str, object]", restart).get("retention")
if not isinstance(retention, Mapping):
return None
retention = cast("Mapping[str, object]", retention)
return retention if retention.get("z") == z else None
def _replayed_steps(request: RunRequest, start_plane: int) -> int | None:
"""Return the step count a replayed schedule will take, when it is knowable.
A fixed schedule knows how many steps are left, which is a better ETA than any rate
of z: the steps of a schedule are not uniform in z, so a run replaying one covers
the last tenth of a metre at a different price from the first. A **resumed** replay
is deliberately excluded -- the schedule's cursor belongs to the backend, not to
this loop, and a total that is wrong by the steps already taken is worse than
falling back to the rate of z.
- ``request``: the run's request, whose ``replay`` is the schedule if there is one.
- Returns: the schedule's whole step count, or ``None`` for an adaptive or resumed
run.
"""
if request.replay is None or start_plane > 0:
return None
return request.replay.schedule.n_steps
def _refuse_a_record_that_misnames_the_basis(
problem: PropagationProblem, metadata: RunMetadata
) -> None:
"""Refuse a run whose record and problem disagree about the basis, by name.
The lowest of the whitelist's doors, and the one a caller reaches by building the
pieces itself: ``run_problem(general_problem, backend, handler, metadata=...)``
with a record that carries no ``modal`` identity would write a file that
:func:`solprop.output.capability.capability_of` reads as an **ordinary modal**
run, and every legacy accessor would then be admitted on a three-component basis
-- the one thing the whitelist exists to prevent. The two are required to agree
in both directions: a record claiming a general modal guide for a problem that
carries no tables would promise a reconstruction no guide can be rebuilt for.
What is compared is the record's **basis**, not whether it carries an identity at
all, and it is read exactly as :func:`solprop.output.capability.capability_of` reads
it. A capillary carrying a coherent wall block writes an identity too, but its basis
is ``"capillary HE_1m"``: the block is a coupled flow through the same orthogonal
``HE_1m`` set the run always had, it builds no ``transform.modal_tables``, and the
ordinary modal accessors stay admitted on it. Keying on the identity's presence
alone would refuse that run here, for a basis disagreement that does not exist.
"""
general = problem.transform.modal_tables is not None
if general == (capability_of(metadata) == "general_modal"):
return
if general:
msg = (
"this problem carries transform.modal_tables and its run record carries "
"no modal identity: the file would be read back as an ordinary modal "
"run, on which energy, peak_power and peak_intensity are admitted, and "
"they are not defined for a three-component basis. Build the record "
"with metadata_for(problem, grid, modes=..., modal=modal_identity(...))"
)
else:
msg = (
"this run record carries a general modal identity and the problem "
"carries no transform.modal_tables: the file would promise a "
"reconstruction from a guide nothing in this run was built from. Pass "
"the record of the problem being run"
)
raise DriverError(msg)
def _refuse_unrunnable(
problem: PropagationProblem,
request: RunRequest,
spec: StatsSpec,
start_plane: int,
n_planes: int,
*,
metadata: RunMetadata,
) -> None:
"""Refuse, by name, every run this loop cannot execute.
- ``spec``: the statistics asked for. A ``periodic`` subset is accepted -- the seam
carries it, and a name no backend can evaluate between planes was refused by
:class:`~solprop.stats.StatsSpec` when the spec was built -- but it must agree
with the request's own, because the channel is armed when the backend is built and
reads the request's spec, not this one.
- ``metadata``: the record about to be written, which must say the same thing about
the basis as the problem does.
"""
_refuse_a_record_that_misnames_the_basis(problem, metadata)
channel = request.observation.stats
if (spec.periodic, spec.period) != (channel.periodic, channel.period):
msg = (
f"the statistics given to this loop ask for {list(spec.periodic)} every "
f"{spec.period} accepted steps and the translated problem asks for "
f"{list(channel.periodic)} every {channel.period}; the between-planes "
"channel is armed when the backend is built, from the problem's own spec, "
"so the two would disagree -- pass the problem's spec, or translate again"
)
raise DriverError(msg)
if not 0 <= start_plane <= n_planes:
msg = (
f"start_plane must be a save plane of this problem, between 0 and "
f"{n_planes}, got {start_plane}"
)
raise DriverError(msg)
if problem.transform.modal_tables is not None and "quadrature_error" in spec.names:
# A silent `nan` at every plane would read as "no alarm raised" rather than
# "no alarm exists": a provider's node rule carries no embedded coarse pair
# to difference, so there is nothing to score.
msg = (
"quadrature_error is unavailable for this general modal guide; its "
"node rule carries no embedded coarse rule to difference, so the "
"transverse-resolution alarm has no error estimate and the statistic "
"would be nan at every plane. The rule's certified tail bound and its "
"sampled Gram defect are measured once, when the guide is built"
)
raise DriverError(msg)
# The accessor whitelist, at the first of its two doors: never redefine
# an old name on a basis it is not defined for, and never answer a general name
# from a file that carries neither the certified metric nor the E/H reference it
# promises. The second door is the processing accessors, and the reference
# backend checks again for a caller who builds it and asks it directly.
refusal = refusal_for(
declared_capability(
general_modal=problem.transform.modal_tables is not None,
spatial=problem.transform.geometry is not None,
),
[*spec.names, *spec.periodic],
withheld=withheld_accessors(
general_modal=problem.transform.modal_tables is not None,
coupled_flow=problem.linear.coupled_flow is not None,
),
)
if refusal is not None:
raise DriverError(refusal)
refusal = axis_refusal(problem, [*spec.names, *spec.periodic])
if refusal is not None:
raise DriverError(refusal)
# What the problem does not carry, refused here for every executor rather than at
# the first plane (SEP 47 section 6.3), as a preparation and a group refuse it.
refusal = statistics_unavailable(problem, [*spec.names, *spec.periodic])
if refusal is not None:
raise BackendError(refusal)