pycif.plugins.obsoperators.standard — API reference

`pycif.plugins.obsoperators.standard` — API reference#

Configuration reference: standard plugin

pycif.plugins.obsoperators.standard.check.check_inputs(inputs, mode)[source]#

Check the consistency of inputs given to the observation operator.

Validates that mode is one of the accepted values and that inputs carries the attributes required by that mode.

Parameters:

inputs – control or observation vector passed to the operator; must expose at least x for 'fwd' mode, and both x and dx for 'tl' mode.
mode (str) – requested execution mode — one of 'fwd', 'tl', or 'adj'.

Returns:

True if all checks pass.

Return type:

bool

Raises:

Exception – if mode is not one of 'fwd', 'tl', or 'adj'.
Exception – if mode is 'tl' and inputs does not expose both x and dx.

pycif.plugins.obsoperators.standard.flushrun.flushrun(self, workdir, rundir, mode, transform_pipe, full_flush=True)[source]#

Remove intermediate files produced by transforms that are no longer needed.

Iterates over every transform in transform_pipe and calls each transform’s own flushrun method to clean up its output files in rundir. In adjoint mode, when the operator is not running in approximate mode, the forward reference directory of each transform is also flushed — provided it lies inside workdir, to avoid accidentally deleting files outside the managed tree.

Parameters:

self (ObsOperator) – the obs-operator plugin instance.
workdir (str) – root working directory; used to check that transf.adj_refdir is a safe path to flush.
rundir (str) – the run sub-directory whose files should be cleaned.
mode (str) – execution mode — one of 'fwd', 'tl', or 'adj'; controls whether adj_refdir of each transform is also flushed.
transform_pipe – the Transform object holding all transforms for this run.
full_flush (bool, optional) – forwarded to each transform’s own flushrun; if False only a partial cleanup is performed (exact behaviour is transform-specific). Defaults to True.

Raises:

PluginError – caught internally and logged as a warning if a transform’s flushrun raises it; execution continues with the remaining transforms.

pycif.plugins.obsoperators.standard.ndarray_wrapper.forward(self, x: ndarray, reload_results=False) → ndarray[source]#

pycif.plugins.obsoperators.standard.ndarray_wrapper.tangent_linear(self, x: ndarray, reload_results=False) → ndarray[source]#

pycif.plugins.obsoperators.standard.ndarray_wrapper.adjoint(self, _x: ndarray, dy: ndarray, reload_results=False) → ndarray[source]#

pycif.plugins.obsoperators.standard.obsoper.obsoper(self, controlvect, obsvect, mode, run_id=0, datei=datetime.datetime(1979, 1, 1, 0, 0), datef=datetime.datetime(2100, 1, 1, 0, 0), workdir='./', reload_results=False, check_transforms=False, ignore_exceptions=False, force_fetch_results=False, **kwargs)[source]#

Run the standard observation operator in forward, tangent-linear or adjoint mode.

Orchestrates the full observation-operator pipeline:

Creates a per-run sub-directory obsoperator/<mode>_<run_id>/ under workdir.
If reload_results is set, attempts to recover cached outputs from a previous run before computing from scratch.
Dispatches to obsoper_serial() or obsoper_parallel() depending on whether self.parallel is configured.
Dumps the resulting observation or control vector to disk for later use.

Parameters:

self (ObsOperator) – the obs-operator plugin instance.
controlvect (ControlVect) – control-vector object. Must carry x (and dx for 'tl' mode); receives dx in 'adj' mode.
obsvect (ObsVect) – observation-vector object. Receives ysim (and dy for 'tl' mode); provides dy in 'adj' mode.
mode (str) – execution mode — one of 'fwd', 'tl', or 'adj'.
run_id (int | str, optional) – identifier for the current run; used to name the sub-directory. Defaults to 0.
datei (datetime.datetime, optional) – start date of the simulation window. Defaults to datetime.datetime(1979, 1, 1).
datef (datetime.datetime, optional) – end date of the simulation window. Defaults to datetime.datetime(2100, 1, 1).
workdir (str, optional) – parent directory in which the run sub-directory is created. Defaults to "./"
reload_results (bool, optional) – if True, attempt to recover pre-computed outputs from the run sub-directory before running the full pipeline. Defaults to False.
check_transforms (bool, optional) – if True, run each transform in both directions and verify the adjoint / TL identity; disables result reloading. Defaults to False.
ignore_exceptions (bool, optional) – if True, non-fatal transform errors are logged and swallowed rather than re-raised. Defaults to False.
force_fetch_results (bool, optional) – if True and cached outputs cannot be found, raise IOError instead of computing. Defaults to False.
**kwargs – extra keyword arguments (ignored).

Returns:

in 'fwd' and 'tl' modes — the updated obsvect with ysim (and dy) populated.

ControlVect: in 'adj' mode — the updated controlvect with dx populated.

Return type:

ObsVect

Raises:

TypeError – if run_id is neither an int nor a str.
IOError – if force_fetch_results is True and cached outputs cannot be loaded.

pycif.plugins.obsoperators.standard.parallel.run_pycif_in_subprocess(python_path, yaml_path)[source]#

Run a pyCIF configuration file in a blocking subprocess.

Launches python_path -m pycif yaml_path, redirecting stdout to subprocess_stdout.log and stderr to subprocess_stderr.log in the same directory as yaml_path.

Parameters:

python_path (str) – path to the Python interpreter (e.g. self.platform.python).
yaml_path (str) – absolute path to the pyCIF YAML configuration file to execute.

Raises:

RuntimeError – if the subprocess exits with a non-zero return code.

pycif.plugins.obsoperators.standard.parallel.obsoper_parallel(self, controlvect, obsvect, rundir, mode, workdir, check_transforms, ignore_exceptions)[source]#

Run the observation operator in parallel over independent time segments.

Splits the simulation window [self.datei, self.datef] into segments of length self.parallel.segments with optional boundary overlap self.parallel.overlap, then runs each segment independently — either as subprocesses (self.parallel.subprocess = True) or as HPC jobs via the platform plugin.

Each segment is configured via a freshly dumped YAML file that restricts the approx_operator window to its date range, then executed with run_pycif_in_subprocess() or self.platform.submit_job.

After all segments finish, their outputs are reassembled:

'tl' mode — obsvect.ysim and obsvect.dy are set to the element-wise sums over all segment observation vectors.
'adj' mode — controlvect.dx is set to the element-wise sum over all segment adjoint sensitivities; controlvect.x and controlvect.xb are reset to their pre-run values.

Parameters:

self (ObsOperator) – the obs-operator plugin instance. Must have self.parallel (with segments, overlap, subprocess attributes), self.datei, self.datef, self.ref_fwd_dir, and self.platform set.
controlvect (ControlVect) – control-vector object.
obsvect (ObsVect) – observation-vector object.
rundir (str) – the run sub-directory for this operator call.
mode (str) – execution mode — 'tl' or 'adj'. (Forward mode is always dispatched to serial execution.)
workdir (str) – parent working directory.
check_transforms (bool) – if True, validate each segment’s transform adjoint / TL identity.
ignore_exceptions (bool) – if True, non-fatal transform errors inside segments are swallowed.

Raises:

RuntimeError – if a subprocess-based segment exits with a non-zero return code (propagated from run_pycif_in_subprocess()).

pycif.plugins.obsoperators.standard.serial.obsoper_serial(self, controlvect, obsvect, rundir, mode, workdir, check_transforms, ignore_exceptions)[source]#

Run the observation operator sequentially over all transforms and time steps.

Handles bookkeeping common to every serial execution:

'fwd' / 'tl' — zeros obsvect.ysim and obsvect.dy, then dumps the control vector to rundir/controlvect.pickle.
'adj' — initialises controlvect.dx = 0 and enables forward-run chaining for multi-step models.

Dispatches to the Dask execution path (init_dask()) when self.use_dask is set, otherwise runs the standard transform loop via do_transforms().

After the run, calls flushrun() to clean up intermediate files when self.autoflush is set (and the operator is not running in parallel mode).

Stores rundir as self.ref_fwd_dir after a forward run so that the subsequent adjoint can locate the forward outputs.

Parameters:

self (ObsOperator) – the obs-operator plugin instance.
controlvect (ControlVect) – control-vector object.
obsvect (ObsVect) – observation-vector object.
rundir (str) – the run sub-directory for this operator call.
mode (str) – execution mode — one of 'fwd', 'tl', or 'adj'.
workdir (str) – parent working directory; forwarded to flushrun().
check_transforms (bool) – if True, validate each transform’s adjoint / TL identity.
ignore_exceptions (bool) – if True, non-fatal transform errors are swallowed rather than re-raised.

pycif.plugins.obsoperators.standard.transforms.batch_computation.batch_computation(self, all_transforms, mapper, dask=False)[source]#

Adapt the transform pipeline for batch computation of Monte-Carlo samples.

Traverses the forward period order in reverse and calls Transform.mapper2batch() on each forward-direction transform to extend its input/output mapper so that a batch of nsamples perturbations can be computed simultaneously.

Parameters:

self (ObsOperator) – the obs-operator plugin instance. Must have self.batch_computation.nsamples, self.batch_computation.dir_samples, and self.batch_computation.file_samples set.
all_transforms – the Transform object holding all transforms.
mapper (dict) – the pipeline mapper dictionary mapping transform IDs to their input/output/precursor/successor metadata; updated in-place.
dask (bool) – whether to use dask for parallel computation.

pycif.plugins.obsoperators.standard.transforms.do_transforms.input_output_msg(trid_list: List[Tuple[str, str]]) → str[source]#

Format a list of tracer IDs as a human-readable multi-line string.

Groups parameters by component and produces output of the form:

- component1: param1, param2
- component2: param1

Parameters:: trid_list (list[tuple[str, str]]) – list of (component, parameter) tracer-ID pairs to format.
Returns:: multi-line string, one component per line, suitable for logging.
Return type:: str

pycif.plugins.obsoperators.standard.transforms.do_transforms.do_transforms(self, transform_pipe, mapper, controlvect, obsvect, mode, rundir, workdir, do_simu=True, onlyinit=False, check_transforms=False, adj_test_threshold=10, save_debug=False, ignore_exceptions=False, ref_fwd_dir='', dump_metadata_only=False, **kwargs)[source]#

Execute the full ordered transform pipeline for a single operator call.

Iterates over all (date, transform, direction) entries in self.period_order_fwd (or self.period_order_adj in adjoint mode) and applies each transform in sequence. Key responsibilities:

Datastore initialisation — builds a nested transform_pipe.datastore[transform][date] dictionary on the first call to track intermediate inputs/outputs shared between transforms.
Restart support — reads rundir/finished_transforms.txt to skip transforms already completed in a previous interrupted run when self.autorestart is enabled.
Input/output routing — uses fetch_inputs_outputs() and aggregate_inout() to gather inputs from precursor datastores, and deaggregate_inout() to redistribute outputs to successor datastores.
Approximate operator — when self.approx_operator is set (parallel mode), transforms outside the segment window execute in dry-run (onlyinit) mode only.
Memory monitoring — tracks peak memory with tracemalloc when self.monitor_memory is enabled.
Memory cleaning — releases unused datastore entries after each transform when self.clean_memory is enabled.
Autokill / restart — kills the job and resubmits if the elapsed wall-clock time exceeds self.autokill_time.
Adjoint / TL test — when check_transforms is True, saves copies of each transform’s in/outputs and calls check_adjtltest() at the end of the adjoint pass.

Parameters:

self (ObsOperator) – the obs-operator plugin instance.
transform_pipe – the Transform pipeline object populated by init_transform().
mapper (dict) – the pipeline mapper dictionary.
controlvect (ControlVect) – control-vector object.
obsvect (ObsVect) – observation-vector object.
mode (str) – execution mode — one of 'fwd', 'tl', or 'adj'.
rundir (str) – the run sub-directory for this operator call.
workdir (str) – parent working directory.
do_simu (bool, optional) – if False, skip actual transform execution (used internally for dry runs). Defaults to True.
onlyinit (bool, optional) – if True, run all transforms in initialisation / dry-run mode only. Defaults to False.
check_transforms (bool, optional) – if True, validate each transform’s adjoint / TL identity. Defaults to False.
adj_test_threshold (float, optional) – relative tolerance used by the adjoint test. Defaults to 10.
save_debug (bool, optional) – if True, dump the inputs and outputs of each transform to rundir for post-run inspection. Defaults to False.
ignore_exceptions (bool, optional) – if True, non-fatal errors inside individual transforms are swallowed and execution continues. Defaults to False.
ref_fwd_dir (str, optional) – path to the reference forward run directory, forwarded to each transform for adjoint input location. Defaults to "".
dump_metadata_only (bool, optional) – passed through to dump_debug() when save_debug is True. If True, writes lightweight plain-text metadata summaries (dimensions, min/max, NaN presence) instead of full NetCDF/datastore files. Greatly reduces wall-time and disk overhead while still allowing data-flow inspection. Defaults to False.
**kwargs – additional keyword arguments (ignored).

pycif.plugins.obsoperators.standard.transforms.dump_read_inout.dump_read_inout(self, all_transforms, backup_comps, mapper)[source]#

Insert dump and load transforms for explicit input/output file I/O.

Scans all transforms for inputs flagged with force_dump and outputs flagged with force_loadout, then inserts extra transforms into the pipeline accordingly:

force_dump on an input: inserts a dump2inputs transform immediately before the flagged transform to write that input to disk so it can be inspected or reused.
force_loadout on an output: inserts a loadfromoutputs transform immediately after the flagged transform to reload that output from disk (useful for transforms whose results must survive across restarts).

Before scanning, propagate_parameters() is called to ensure force_load flags have been propagated through the pipeline.

Parameters:

self (ObsOperator) – the obs-operator plugin instance.
all_transforms – the Transform object holding all transforms; modified in-place.
backup_comps (dict) – backed-up component definitions forwarded to add_default().
mapper (dict) – the pipeline mapper dictionary; updated in-place to include entries for the newly inserted transforms.

pycif.plugins.obsoperators.standard.transforms.init_control_transformations.init_control_transformations(self, all_transforms, controlvect, backup_comps, mapper)[source]#

Initialize transforms on the control-vector side.

Reads controlvect.transform_pipe and inserts each of its transforms before the first element of self.mainpipe in all_transforms, preserving the user-defined order.

Also loops over all components/tracers of the datavect and, for those that specify the unit_conversion argument, automatically inserts a unit_conversion transform.

Parameters:

self (ObsOperator) – the obs-operator plugin instance; uses self.mainpipe to determine the insertion point.
all_transforms – the Transform object holding all transforms; modified in-place.
controlvect (ControlVect) – control-vector object; its transform_pipe is read to determine which transforms to insert.
backup_comps (dict) – backed-up component definitions forwarded to add_default().
mapper (dict) – the pipeline mapper dictionary; updated in-place.

pycif.plugins.obsoperators.standard.transforms.init_mainpipe.init_mainpipe(self, all_transforms, backup_comps, mapper)[source]#

Initialize the core of the transform pipeline.

Reads self.transform_pipe (transforms defined directly on the observation operator in the YAML) and inserts each of its transforms before the first element already present in self.mainpipe. If no transform_pipe is defined and self.ignore_model is False, a default run_model transform is added automatically.

Warning

If transform_pipe is specified in the observation operator, only the explicitly listed transforms are used — the CTM model is not added automatically. To run the model on top of custom transforms, include run_model explicitly in the list. For most applications it is preferable to define extra transforms in the controlvect or obsvect transform_pipe instead.

Parameters:

self (ObsOperator) – the obs-operator plugin instance. On return, self.mainpipe is updated with the IDs of the newly inserted transforms.
all_transforms – the Transform object holding all transforms; modified in-place.
backup_comps (dict) – backed-up component definitions forwarded to add_default().
mapper (dict) – the pipeline mapper dictionary; updated in-place.

pycif.plugins.obsoperators.standard.transforms.init_obsvect_transformations.init_obsvect_transformations(self, all_transforms, obsvect, backup_comps, mapper)[source]#

Initialize transforms on the observation-vector side.

Appends a toobsvect transform to the pipeline for every observed species (component/tracer pair where param.isobs is True). For satellites components, a satellites transform is inserted immediately before the corresponding toobsvect step.

Then reads obsvect.transform_pipe and prepends each of its transforms before all other transforms in all_transforms, preserving the user-defined order.

Parameters:

self (ObsOperator) – the obs-operator plugin instance. On return, self.mainpipe is populated with the IDs of the newly inserted toobsvect (and satellites) transforms.
all_transforms – the Transform object holding all transforms; modified in-place.
obsvect (ObsVect) – observation-vector object; its transform_pipe and datavect are read to determine which transforms to insert.
backup_comps (dict) – backed-up component definitions forwarded to add_default().
mapper (dict) – the pipeline mapper dictionary; updated in-place.

pycif.plugins.obsoperators.standard.transforms.period_pipe.period_pipe(self, all_transforms, mapper)[source]#

Arrange all transforms into ordered forward and adjoint execution pipes.

Determines the chronologically correct execution order for every (transform, sub-simulation date) pair by:

Propagating sub-simulation periods from each transform to its precursors and successors via default_subsimus().
Building a dependency graph and walking it in forward order with fwd_adj_pipe() (mode='forward').
Walking the same graph in reverse order (mode='adjoint').

Each returned pipe is a list of (date, transform_id, direction) tuples, where direction is either 'forward' or 'adjoint' and controls whether a transform runs in its normal or dry-run mode.

Parameters:

self (ObsOperator) – the obs-operator plugin instance.
all_transforms – the Transform object holding all initialized transforms.
mapper (dict) – the pipeline mapper dictionary mapping transform IDs to their sub-simulation, input/output and precursor/successor metadata.

Returns:

(pipe_fwd, pipe_adj) where each element is a list of (datetime.datetime, str, str) tuples giving the execution order for forward and adjoint runs respectively.

Return type:

tuple[list, list]

pycif.plugins.obsoperators.standard.transforms.utils.add_default.add_default(self, transforms, yml_dict, position='last', index=0, init=False, mapper={}, transform_type='state', backup_comps={}, ref_transform='', precursor=None, successor=None, transform_id=None, do_pipe_entry=False)[source]#

Instantiate a new transform from a YAML-like config and wire it in.

Creates and registers a transform described by yml_dict (via Setup), gives it a unique id (either transform_id, or a name derived from its plugin name/version, or a generic default_{index}), and inserts it into transforms.attributes at the requested position. If init is True, also initializes its mapper entry (Transform.ini_mapper()), forces the precursors/successors given in precursor/successor (update_successors_precursors()), and builds its internal inputs/outputs paths (generate_internal_pipe()).

After creation, connects the new transform to the rest of the pipe (connect_pipes.connect_pipes()), propagates attributes backward/forward (propagate_attributes.propagate_attributes()), optionally initializes its pipe entry when it has no precursor (init_pipe_entry()), and finally lets init_default_transformations.init_default_transformations() insert any further default transforms needed to reconcile input/output formats.

Parameters:

self – The parent object (obs operator), exposing monitor_memory.
transforms – Namespace holding all registered transform instances, mutated in place with the new transform.
yml_dict – Dictionary describing the transform to create, in the same format as a transform entry in the YAML configuration (must include a "plugin" block).
position – Where to insert the new transform id in transforms.attributes: "last"/"end", "start"/"first", or "index" (uses index).
index – Insertion index used when position == "index".
init – Whether to initialize the transform’s mapper entry.
mapper – Dictionary mapping each transform id to its inputs/outputs/ precursors/successors metadata, updated with the new entry.
transform_type – Type passed through to the transform setup, currently unused directly in this function.
backup_comps – Backup of components used to restore/compare state.
ref_transform – Id of a reference transform used by Transform.ini_mapper() to resolve relative attributes.
precursor – Optional {trid: precursor_id(s)} mapping forcing the new transform’s precursors for the given input trids.
successor – Optional {trid: successor_id(s)} mapping forcing the new transform’s successors for the given output trids.
transform_id – Optional explicit id to use instead of an auto-generated one.
do_pipe_entry – Whether to initialize the pipe entry (default “fromcontrol”/”unit_conversion” transforms) for the new transform.

Returns:

(new_transf, new_id), the newly created transform instance and its id in transforms.

Return type:

tuple

pycif.plugins.obsoperators.standard.transforms.utils.add_default.init_pipe_entry(self, all_transforms, backup_comps, mapper, transform)[source]#

Add default entry transforms for the inputs of transform with no precursor.

For every input trid of transform that has no precursor yet, resolves the corresponding component/parameter in self.datavect.components and:

If the component/parameter cannot be resolved, records it in self.missing (and, if it is required for initialization per self.init_inputs, in self.init_missing too) so it can be reported later, and skips inserting a transform for it.
If the component is an observation (isobs), skips it (it is not data to be initialized from control).
Otherwise, inserts a fromcontrol transform as new precursor via add_default(), plus a unit_conversion transform right after it if the parameter defines a unit_conversion block.

Along the way, self.required_inputs is populated for every encountered component/parameter for later debug dumping.

Parameters:

self – The obs operator, exposing/receiving required_inputs, missing, init_missing, datavect and optionally init_inputs.
all_transforms – Namespace holding all registered transform instances.
backup_comps – Backup of components, used to resolve component names that were renamed/replaced.
mapper – Dictionary mapping each transform id to its inputs/outputs/ precursors metadata, updated with any inserted transform.
transform – Id of the transform whose entry (precursor-less) inputs are being initialized.

pycif.plugins.obsoperators.standard.transforms.utils.add_default.update_successors_precursors(new_id, precursors2add, successor2add, transf_mapper_loc, mapper)[source]#

Force explicit precursors/successors on a newly created transform.

Ensures every input/output trid of the new transform has a "precursors"/"successors" list (creating empty ones if absent), then appends the ids given in precursors2add/ successor2add (a string or list of strings per trid) to the matching lists.

When both precursors2add and successor2add are provided and share a trid whose precursor/successor pair was already directly connected (i.e. the precursor already lists the successor, and vice versa), that direct link is rewired through the new transform: the new transform id replaces the old successor in the precursor’s successors list, and replaces the old precursor in the successor’s precursors list.

Parameters:

new_id – Id of the newly created transform being spliced in.
precursors2add – Optional {trid: precursor_id(s)} mapping of precursors to force for the new transform’s inputs.
successor2add – Optional {trid: successor_id(s)} mapping of successors to force for the new transform’s outputs.
transf_mapper_loc – Mapper entry of the new transform, mutated in place with the forced precursors/successors.
mapper – Dictionary mapping each transform id to its precursors/ successors metadata, mutated in place when rewiring direct links through the new transform.

Raises:

CifTypeError – If a value in precursors2add or successor2add
is neither a string nor a list. –

pycif.plugins.obsoperators.standard.transforms.utils.add_default.generate_internal_pipe(new_id, mapper)[source]#

Build the internal input/output linkage of a transform’s mapper entry.

If not already present, sets outputs2inputs so that every output trid is linked to all input trids (the default assumption that any output may depend on any input). Then derives the reverse mapping inputs2outputs from it via Transform.generate_inputs2outputs().

Parameters:

new_id – Id of the transform whose internal pipe is being generated.
mapper – Dictionary mapping each transform id to its mapper entry; mapper[new_id] is mutated in place with outputs2inputs and inputs2outputs.

pycif.plugins.obsoperators.standard.transforms.utils.aggreg_deaggreg_inout.aggregate_inout(transform, ddi, tmp_inputs, tmp_outputs, transform_mode, transform_onlyinit, mapper, check_transforms=False)[source]#

Aggregate inputs and outputs from precursors and successors respectively.

There can be only one precursor per trid, with the possibility of having several sub-dates. By construction, it is not possible to have several precursors for a given trid to avoid ambiguity. This could become possible in the future, but it would require further complexity in identifying datastores (i.e., including precursor and successor in dictionaries, at the cost of reduced readability).

Outputs can have several successors for a given trid. This implies that adjoint sensitivities must be properly propagated backwards.

Parameters:

transform (str) – The name of the transform currently being processed
ddi (datetime.datetime) – The period being processed
tmp_inputs (dict) – Per-trid, per-date datastore of raw inputs as fetched from precursors (nested under the single precursor id and its own sub-simulation date), as produced by fetch_inoutputs.fetch_inputs_outputs(). Collapsed in place to {trid: {date: <data>}}.
tmp_outputs (dict) – Per-trid, per-date datastore of raw outputs as fetched from successors (nested under each successor id and its own sub-simulation date). Collapsed in place to {trid: {date: <data>}}, summing (or concatenating, for sparse data) contributions from multiple successors.
transform_mode (str) – Either "fwd" or "adj", the direction currently being executed.
transform_onlyinit (bool) – Whether the transform is only being initialized (no actual data processing), used to decide whether to compute the delta_adj_out debug diagnostic.
mapper (dict) – Dictionary mapping each transform id to its precursors/successors/outputs metadata.
check_transforms (bool, optional) – Whether to keep track of the original adjoint output (adj_out_original) so that check_adjtltest.check_adjtltest() can later verify consistency. Defaults to False.

Raises:

CifError – If more than one precursor is found for a given input trid (should never happen by construction).
CifTypeError – If a transform mixes sparse and non-sparse outputs for the same trid.

Returns:

(tmp_inputs, tmp_outputs), the same dictionaries passed in, mutated and collapsed to one entry per (trid, date).

Return type:

tuple

pycif.plugins.obsoperators.standard.transforms.utils.aggreg_deaggreg_inout.deaggregate_inout(transform, transform_mode, transform_onlyinit, ddi, tmp_datastore, mapper, check_transforms=False)[source]#

Spread a transform’s collapsed inputs/outputs back out to precursors/successors.

This is the inverse of aggregate_inout(): it takes the already-computed tmp_datastore["inputs"]/["outputs"] for one sub-simulation (ddi) of transform, and re-nests each {trid: {date: <data>}} entry back under the relevant precursor (for inputs) or successor (for outputs) and their own sub-simulation date, so that each neighbour transform finds its expected data shape in the shared datastore.

For outputs backed by a pandas.DataFrame (sparse data), the per-successor/per-date split uses the aggregation_index recorded by aggregate_inout() (falling back to the current successor/date index if unavailable). For dict-based (non-sparse) outputs, the original_outputs saved during aggregation is used to restore each successor’s individual adj_out contribution.

Parameters:

transform – The name of the transform currently being processed.
transform_mode – Either "fwd" or "adj", the direction currently being executed.
transform_onlyinit – Whether the transform is only being initialized (no actual data processing).
ddi – The sub-simulation index being processed.
tmp_datastore – This transform’s {"inputs": ..., "outputs": ...} datastore for sub-simulation ddi, holding the collapsed {trid: {date: <data>}} entries to spread back out.
mapper – Dictionary mapping each transform id to its precursors/ successors/subsimus metadata.
check_transforms – Whether to compute the delta_adj_out debug diagnostic (difference between the adjoint output before and after aggregation) for precursor inputs.

Returns:

(tmp_inputs, tmp_outputs), the input/output datastores re-nested per precursor/successor and their own sub-simulation date.

Return type:

tuple

pycif.plugins.obsoperators.standard.transforms.utils.check_adjtltest.check_adjtltest(self, period_order, mapper, transform_pipe)[source]#

Run the adjoint / tangent-linear dot-product test on each transform.

For every (date, transform) pair processed in the forward direction, computes the two sides of the dot-product identity used to validate the adjoint of a transform:

<dx | H^T(H(dx))>  vs  <H(dx) | H(dx)>

dx_in (left-hand side) is accumulated from the transform’s inputs, by summing the element-wise product of the tangent-linear increment (incr) and the adjoint sensitivity (delta_adj_out if present, else adj_out) of each precursor. dx_out (right-hand side) is accumulated the same way from the transform’s outputs and successors. The two quantities should be equal, up to floating-point round-off, if the transform’s adjoint is correctly implemented; the relative difference is reported in units of machine epsilon.

Results for all checked transforms are logged and written to {self.workdir}/check_transforms.log.

Parameters:

self – obsoperator instance exposing data_tl and data_adj (datastores keyed by transform name, then by sub-simulation date/id ddi, holding the tangent-linear increments and adjoint outputs for each input/output tracer trid) as well as workdir.
period_order (list[tuple]) – ordered (ddi, transform, direction) triplets describing the transforms scheduled to run, as built by the transform pipeline. Only direction == "forward" entries are checked here (the tangent-linear/adjoint pair for a transform is compared once, not once per direction).
mapper (dict) – per-transform metadata (inputs, outputs, precursors, successors, subsimus) describing how sub-simulations and tracers (trid) are linked across transforms.
transform_pipe – container exposing each transform instance as an attribute (getattr(transform_pipe, transform)); used here only to retrieve the plugin name of transforms and their precursors/successors.

Returns:

None. Results are emitted via the info logger and written to check_transforms.log in self.workdir.

pycif.plugins.obsoperators.standard.transforms.utils.check_datavect.check_datavect(self, all_transforms, backup_comps, mapper)[source]#

Check that the data vector has no missing required inputs.

Dumps self.required_inputs to pipe_inputs.txt for debugging. If self.missing is empty, returns immediately. Otherwise, dumps the full list of missing inputs to missing_inputs.txt and the subset needed for initialization (self.init_missing) to missing_inputs_init.txt, both under {self.workdir}/obsoperator/. Raises a CifError unless self has an init_inputs attribute and no inputs are missing for initialization.

Parameters:

self – The obs operator, exposing workdir, required_inputs, missing and init_missing.
all_transforms – Namespace holding all registered transform instances, used to resolve each transform’s plugin name/version.
backup_comps – Unused backup of components, kept for interface consistency with related functions.
mapper – Unused transform mapper, kept for interface consistency with related functions.

Raises:

CifError – If required inputs are missing from the data vector.

pycif.plugins.obsoperators.standard.transforms.utils.clean_memory.clean_memory(self, transform_pipe, mapper, period_order, ddi, transform, direction, mode, only_init)[source]#

Free datastore entries of a transform that are no longer needed.

After a transform has run, clears (sets to an empty dict) the parts of transform_pipe.datastore that will not be read again:

Per-date entries of the transform’s own inputs/outputs (depending on direction) that are not consumed by any of its successors in self.pipe_links_fwd/self.pipe_links_adj.
The transform’s “incoming” side (inputs in forward mode, outputs in adjoint mode), unless flagged as still needed via need_{inputs,outputs} and only_init is set.
Neighbours earlier in period_order whose only remaining successor is the current transform.

Entries flagged with keep_data_after_init in mapper are preserved when only_init is True.

Parameters:

self – The parent object exposing pipe_links_fwd/pipe_links_adj and (optionally) monitor_memory.
transform_pipe – Object holding the datastore to clean, keyed by transform, sub-simulation index (ddi) and inputs/outputs.
mapper – Dictionary mapping each transform id to its inputs/outputs metadata (including keep_data_after_init and need_* flags) and sub-simulation structure.
period_order – Ordered list of (ddi, transform, direction) tuples describing the execution order of the pipe.
ddi – Index of the sub-simulation currently being cleaned.
transform – Id of the transform whose datastore is being cleaned.
direction – Either "forward" or "adjoint".
mode – Execution mode, unused here but kept for interface consistency.
only_init – Whether cleaning happens right after initialization, in which case data flagged as still needed is preserved.

Returns:

The (mutated in place) transform_pipe.

pycif.plugins.obsoperators.standard.transforms.utils.connect_pipes.connect_pipes(all_transforms, mapper, transform)[source]#

Wire a newly added transform into the pipe based on shared trids.

For each of transform’s input trids, scans every transform registered before it for a matching output trid and records it as a precursor (symmetrically, successors are found by scanning transforms registered after it). Pre-existing precursors/successors (e.g. forced via add_default.update_successors_precursors()) are left untouched.

Then prunes redundant indirect links: if a precursor of transform for a trid is itself reachable through another of that trid’s precursors, it is removed from the direct list (and symmetrically for successors), so that only the most immediate producer/consumer chain is kept. If several precursors remain for the same trid on a downstream transform, only the closest one (last in pipe order) is kept, and the corresponding transforms’ successors lists are updated to match.

Parameters:

all_transforms – Namespace holding all registered transform instances, in pipe order, used to scan for candidate precursors/successors.
mapper – Dictionary mapping each transform id to its inputs/outputs/ precursors/successors metadata, mutated in place.
transform – Id of the transform to connect to the rest of the pipe.

Raises:

CifAttributeError – If transform is not yet registered in
all_transforms.attributes` –

pycif.plugins.obsoperators.standard.transforms.utils.connect_pipes.prune_dead_branches(all_transforms, mapper, skip_transform)[source]#

Remove transforms that have no successors left in the pipe.

Iterates over a snapshot of the registered transforms and deletes any transform (other than toobsvect transforms and skip_transform) whose successors mapping is empty for all its outputs: it is dropped from its precursors’ successors lists, removed from mapper, and removed from all_transforms.attributes (with the corresponding attribute set to None).

Parameters:

all_transforms – Namespace holding all registered transform instances, mutated in place to drop dead transforms.
mapper – Dictionary mapping each transform id to its precursors/ successors metadata, mutated in place.
skip_transform – Id of the transform currently being initialized, which must never be pruned.

pycif.plugins.obsoperators.standard.transforms.utils.default_subsimus.default_subsimus(all_transforms, mapper)[source]#: Initialize sub-simulations for each transform for which sub-simulations are not already defined. By default, the sub-simulations of a given transform are deduced from the shape of the “input_dates” of each component/tracer in the outputs.

pycif.plugins.obsoperators.standard.transforms.utils.dump_debug.dump_debug(transform, transf_mapper, tmp_datastore, runsubdir, ddi, entry='outputs', transform_onlyinit=False, dump_metadata_only=False)[source]#

Dump inputs or outputs of a transform step for post-run inspection.

For each tracer ID (trid) in tmp_datastore[entry], writes debug files under:

<runsubdir>/../transform_debug/<transform>/<ddi>/<component>/<parameter>/

Two modes are supported:

Full dump (dump_metadata_only=False)

Writes the complete datastore to disk as NetCDF files (xr.Dataset) or pyCIF datastore files (pd.DataFrame). One file is produced per tracer ID and date. This mode is accurate but slow and disk-intensive.

Metadata-only dump (dump_metadata_only=True)

Writes lightweight plain-text files instead of full data files. Dramatically reduces wall-time overhead and disk usage while still capturing enough information to trace data flow and detect anomalies.

For xr.Dataset values — records the dimension names and sizes, min/max values, and NaN presence for the spec variable (and incr when it exists in the dataset).
For pd.DataFrame values — records the row count and, for each of the maindata, spec, and incr columns that are present, the min/max values and NaN presence.

Parameters:

transform (str) – name of the transform being debugged.
transf_mapper (dict) – mapper entry for the transform (not directly used here; kept for API consistency).
tmp_datastore (dict) – the datastore for the current transform step, keyed by "inputs" and "outputs".
runsubdir (str) – the per-period run sub-directory. Debug files are written to <runsubdir>/../transform_debug/….
ddi (datetime.datetime) – the current simulation date (used both for directory naming and for per-date filename formatting).
entry (str, optional) – which side of the datastore to dump — "inputs" or "outputs". Defaults to "outputs".
transform_onlyinit (bool, optional) – when True, the transform ran in dry-run / init-only mode. Currently unused inside this function but kept for API consistency with the call sites in do_transforms(). Defaults to False.
dump_metadata_only (bool, optional) – when True, write lightweight metadata text files instead of full NetCDF/datastore files. Defaults to False.

Raises:

TypeError – if a datastore entry is neither an xr.Dataset nor a dict of xr.Dataset / pd.DataFrame.

pycif.plugins.obsoperators.standard.transforms.utils.dump_transform_description.dump_transform_description(self, all_transforms, mapper)[source]#

Dump a human-readable description of the transform pipe to disk.

Writes {self.workdir}/obsoperator/transform_description.txt, listing for each transform its plugin name/version and its inputs, outputs, precursors and successors. Unless self.use_dask is set, also writes transform_pipe_forward.txt and transform_pipe_adjoint.txt, listing each transform in its self.period_order_fwd/self.period_order_adj execution order together with its sub-simulation index, direction and input ids.

Parameters:

self – The obs operator, exposing workdir, use_dask, period_order_fwd and period_order_adj.
all_transforms – Namespace holding all registered transform instances.
mapper – Dictionary mapping each transform id to its inputs/outputs/ precursors/successors metadata.

pycif.plugins.obsoperators.standard.transforms.utils.fetch_inoutputs.fetch_inputs_outputs(transform, ddi, transform_pipe, tmp_datastore, subsimus, successors, precursors, transf_mapper, mapper, return_links=False)[source]#

Collect, for one sub-simulation, the data available from neighbours.

For each output trid/date of transform, looks up the successors’ datastores to find matching sub-periods already computed for them, and nests that data under tmp_outputs[trid][date][successor]. Symmetrically, for each input trid/date, looks up the precursors’ output datastores and nests the available data under tmp_inputs[trid][date][precursor].

Parameters:

transform – Id of the transform for which inputs/outputs are fetched.
ddi – Index of the sub-simulation being processed.
transform_pipe – Object holding the full datastore of every transform, indexed by transform, sub-simulation index and inputs/outputs.
tmp_datastore – This transform’s own {"inputs": ..., "outputs": ...} datastore, used as the base to extend.
subsimus – Dictionary describing, for "inputs" and "outputs", which trids and dates are relevant to this sub-simulation.
successors – Mapping from output trid to the list of successor transform ids consuming it.
precursors – Mapping from input trid to the list of precursor transform ids producing it.
transf_mapper – Mapper entry of transform (used to check the sampled flag on inputs).
mapper – Dictionary mapping each transform id to its subsimus metadata, used to align precursor/successor sub-periods.
return_links – If True, return only the mapping of which neighbour sub-simulations were actually used, instead of the merged data.

Returns:

If return_links is False, a tuple (tmp_inputs, tmp_outputs) holding this transform’s inputs/outputs datastores augmented with the data fetched from precursors/successors. If return_links is True, a tuple (used_inputs, used_outputs) describing which precursor/successor sub-simulations (indexed by ddi, date and trid) were actually consulted.

pycif.plugins.obsoperators.standard.transforms.utils.fwd_pipe.fwd_adj_pipe(self, all_transforms, mapper, mode='forward')[source]#

Compute the optimal execution order for one direction of the pipe.

Builds, for every (sub-simulation, transform) pair, the graph of immediate precursors/successors (pipe_links, saved on self as pipe_links_fwd/pipe_links_adj for later memory cleaning), then appends a virtual entry transform (final_toobsvect in forward mode, final_fromcontrol in adjoint mode) that gathers every pipe-end node so the whole graph can be walked from a single root.

If self.use_dask is set, the raw links are stored (pipe_links_fwd_dask/pipe_links_adj_dask) and, for the adjoint call, the function returns early since Dask handles ordering itself.

Otherwise, it iteratively expands the graph from the virtual root (vectorized with pandas for speed) to produce, for each node, the chronologically correct sequence of precursors, marking as “dead branches” those adjoint/forward paths that do not eventually reach a control-vector-relevant fromcontrol transform (unless self.force_full_operator is set). Dead-end nodes (with no successors in the current direction) are then pruned iteratively.

The final ordered list of (ddi, transform, direction) tuples is stored on self.period_order_fwd (mode "forward") or self.period_order_adj (mode "adjoint").

Parameters:

self – The obs operator, exposing use_dask, force_full_operator and datef, and receiving the computed pipe_links_* and period_order_* attributes.
all_transforms – Namespace holding all registered transform instances, in pipe order.
mapper – Dictionary mapping each transform id to its subsimus, inputs and outputs metadata.
mode – Either "forward" or "adjoint", selecting which direction of the pipe to order.

pycif.plugins.obsoperators.standard.transforms.utils.init_default_transformations.init_default_transformations(self, all_transforms, backup_comps, mapper, transform, do_pipe_entry=False, trid_to_check=None)[source]#

Initialize default transformations based on compatibility of input/output formats of successive transforms.

For each output trid of transform (or only trid_to_check if given) and each of its successors tr, compares the output attributes (trid_dict) against the successor’s matching input attributes (tmp_dict) and inserts, as needed, one of:

init_sparse.init_sparse(), if either side is sparse/sampled data.
init_reindex.init_reindex(), for temporal re-indexing — inserted before the horizontal/vertical reprojection when the tracer’s time_interpolation.first flag is set, after it otherwise.
init_regrid.init_regrid(), for horizontal/vertical reprojection when the domains differ.

Each helper may itself splice a new transform into the pipe; this function then relies on the recursive nature of add_default.add_default() (which calls back into this function) to resolve any remaining mismatches for other dimensions, so at most one bridging transform is inserted per successor per call.

Parameters:

self – The parent object (obs operator), exposing datavect.
all_transforms – Namespace holding all registered transform instances.
backup_comps – Backup of components used to resolve component names that were renamed/replaced.
mapper – Dictionary mapping each transform id to its inputs/outputs/ precursors/successors metadata, updated when transforms are inserted.
transform – Id of the transform whose outputs (successors’ inputs) are being reconciled.
do_pipe_entry – Whether to run the pipe-entry initialization step for any newly inserted transform.
trid_to_check – Optional list restricting the loop to specific output trids instead of all of transform’s successors.

pycif.plugins.obsoperators.standard.transforms.utils.init_entry.init_entry(self, all_transforms, backup_comps, mapper)[source]#

Initialize the pipe entry (default attributes) for every transform.

Iterates over a snapshot of the currently registered transforms and calls add_default.init_pipe_entry() on each of them, so that transforms added while iterating are not themselves re-processed.

Parameters:

self – The parent object (obs operator) holding the transform pipe.
all_transforms – Namespace holding all registered transform instances.
backup_comps – Backup of components used to restore/compare state.
mapper – Dictionary mapping each transform id to its inputs/outputs/ precursors/successors metadata.

pycif.plugins.obsoperators.standard.transforms.utils.init_regrid.init_regrid(self, trid, tmp_dict, trid_dict, precursor_id, transform, param, all_transforms, mapper, backup_comps, precursors, return_last=True, do_pipe_entry=False)[source]#

Insert a regrid or vertical-interpolation transform if domains differ.

Compares the horizontal domain of tmp_dict (typically a precursor’s output) against trid_dict (the target input) and, if they differ (grid or LBC status), inserts a regrid transform between precursor_id and transform using add_default.add_default(). Otherwise, if the vertical domains differ (grid or top-of-atmosphere status) and vdomain_from_previous is not forced, inserts a vertical_interpolation transform instead. At most one of the two transforms is inserted per call.

Parameters:

self – The parent object (obs operator) holding the transform pipe.
trid – (component, parameter) tuple identifying the tracer.
tmp_dict – Attribute dictionary of the source side (precursor’s output), holding at least "domain" and optionally "is_lbc"/"is_top".
trid_dict – Attribute dictionary of the target side (this transform’s input), holding the same keys to compare against.
precursor_id – Id of the precursor transform providing the data.
transform – Id of the transform whose input is being checked.
param – Object holding optional regrid/vertical_interpolation configuration blocks used to parametrize the inserted transform.
all_transforms – Namespace holding all registered transform instances.
mapper – Dictionary mapping each transform id to its inputs/outputs metadata, updated when a new transform is inserted.
backup_comps – Backup of components used to restore/compare state.
precursors – Precursors metadata, kept for interface consistency with related init functions.
return_last – Unused flag, kept for interface consistency with related init functions.
do_pipe_entry – Whether to run the pipe-entry initialization step for the newly inserted transform.

Returns:

(local_pipe, did_nothing) where local_pipe is [precursor_id, transform] with the newly inserted transform id spliced in when applicable, and did_nothing is False if a transform was inserted, True otherwise.

Return type:

tuple

pycif.plugins.obsoperators.standard.transforms.utils.init_reindex.init_reindex(self, trid, tmp_dict, trid_dict, precursor_id, transform, param, all_transforms, mapper, backup_comps, precursors, do_pipe_entry=False)[source]#

Insert a time-interpolation transform if the date indices differ.

Compares the input_dates of tmp_dict (a precursor’s output) against trid_dict (the target input), per sub-simulation. If they already match exactly for every sub-simulation, nothing is inserted. Otherwise, a time_interpolation transform is created via add_default.add_default() and spliced between precursor_id and transform.

Parameters:

self – The parent object (obs operator) holding the transform pipe.
trid – (component, parameter) tuple identifying the tracer.
tmp_dict – Attribute dictionary of the source side (precursor’s output), holding "input_dates" and "sparse_data".
trid_dict – Attribute dictionary of the target side (this transform’s input), holding "input_dates" to compare against.
precursor_id – Id of the precursor transform providing the data.
transform – Id of the transform whose input is being checked.
param – Object holding an optional time_interpolation configuration block used to parametrize the inserted transform.
all_transforms – Namespace holding all registered transform instances.
mapper – Dictionary mapping each transform id to its inputs/outputs metadata, updated when a new transform is inserted.
backup_comps – Backup of components used to restore/compare state.
precursors – Precursors metadata, kept for interface consistency with related init functions.
do_pipe_entry – Whether to run the pipe-entry initialization step for the newly inserted transform.

Returns:

The id of the transform now directly feeding transform for this trid — either the newly inserted time_interpolation transform, or the original precursor_id if indices already match.

Return type:

str

pycif.plugins.obsoperators.standard.transforms.utils.init_sparse.init_sparse(self, trid, precursor_dict, ref_dict, tr, transform, param, all_transforms, mapper, backup_comps, precursors, do_pipe_entry=False)[source]#

Insert the transform needed to reconcile sparse/sampled data formats.

Compares the data representation of a precursor’s output (precursor_dict) against the target input (ref_dict) for a given tracer, and inserts, at most, a single transform to bridge the first mismatch found, in this priority order:

sparse2sample if the precursor is sparse but the target expects gridded data.
array2sampled if the precursor is gridded but the target expects sampled/sparse data (raises CifError if the target is flagged sampled, which should never occur here).
time_interpolation if the precursor’s input_dates do not match the target’s.
vertical_interpolation if neither side has a continuous vertical domain and the precursor is gridded.
regrid if neither side has a continuous horizontal domain and the precursor is gridded.

Each inserted transform is created via add_default.add_default() and spliced between the precursor (tr) and transform. As soon as one transform is inserted, the function returns immediately: the caller re-invokes the recursive default-initialization logic to handle any remaining mismatches for other dimensions.

Parameters:

self – The parent object (obs operator) holding the transform pipe.
trid – (component, parameter) tuple identifying the tracer.
precursor_dict – Attribute dictionary of the precursor’s output (sparse_data, sampled, domain, input_dates, continuous_hdomain, continuous_vdomain, etc.).
ref_dict – Attribute dictionary of the target input to reconcile precursor_dict against.
tr – Id of the precursor transform providing the data.
transform – Id of the transform whose input is being checked.
param – Object holding optional time_interpolation, vertical_interpolation and regrid configuration blocks used to parametrize the inserted transform.
all_transforms – Namespace holding all registered transform instances.
mapper – Dictionary mapping each transform id to its inputs/outputs metadata, updated when a new transform is inserted.
backup_comps – Backup of components used to restore/compare state.
precursors – Precursors metadata, kept for interface consistency with related init functions.
do_pipe_entry – Whether to run the pipe-entry initialization step for the newly inserted transform.

Returns:

The id of the transform now directly feeding transform for this trid when a bridging transform was inserted. Returns None if none of the mismatch conditions apply (no bridging transform is needed).

Return type:

str or None

Raises:

CifError – If ref_dict is flagged as sampled while the
precursor is neither sampled nor sparse (should never happen). –

pycif.plugins.obsoperators.standard.transforms.utils.initiate_default_attributes.initiate_default_attributes(all_transforms, mapper, transform_id)[source]#

Fill in default values for attributes not explicitly set on a transform.

Runs once per transform (subsequent calls are no-ops, tracked via the default_attributes_initialized flag). For every input trid, sets recombine_periods (and, if a domain is already set, is_lbc, is_top and force_loadin) to their default values when absent, each paired with an {attribute}_default marker flag. For every output trid, similarly defaults recombine_periods, is_lbc and is_top, and additionally marks all_successors_initialized as True when the trid has no successor (or only itself as successor).

Parameters:

all_transforms – Namespace holding all registered transform instances, unused directly here but kept for interface consistency with related propagation functions.
mapper – Dictionary mapping each transform id to its inputs/outputs/ successors metadata; mapper[transform_id] is mutated in place.
transform_id – Id of the transform whose default attributes are being initialized.

pycif.plugins.obsoperators.standard.transforms.utils.precursors_successors.add_precursors_successors(self, all_transforms, transforms_ids, mapper, pipe_links, pipe_subend, transf, simu, mode='adjoint', fetch_precursors=True)[source]#

Register a transform’s sub-simulation and link it to its immediate neighbours.

Called once per (transform, sub-simulation) pair, twice (with fetch_precursors True then False) from fwd_pipe.fwd_adj_pipe(). Appends (simu, transf, "adjoint" if fetch_precursors else "forward") to transforms_ids and, unless already present in pipe_links, initializes its (empty) list of links — skipping dead-end transforms that have no successors/precursors and are not a pipe end/start (unless self.force_full_operator is set).

Then, for each input trid (if fetch_precursors) or output trid (otherwise) of this sub-simulation, walks its precursors/successors and, for each of their matching sub-simulations, appends the corresponding node id to pipe_links[transf_id]. If the trid is flagged with break_{fwd,adj}_onlyinit_pipe on either side, the link is instead redirected to a same-transform self-loop and the neighbour is recorded in pipe_subend (used later to seed the virtual end-of-pipe node), so that only-init propagation stops there instead of walking further into the neighbour.

If the transform has no precursors/successors at all in the relevant direction, appends a self-loop link so the transform still appears in its own pipe.

Parameters:

self – The parent object (obs operator), exposing force_full_operator.
all_transforms – Namespace holding all registered transform instances, used to check end_pipe/start_pipe flags.
transforms_ids – List of every registered (ddi, transform, direction) node id, mutated in place.
mapper – Dictionary mapping each transform id to its inputs/outputs/ precursors/successors/subsimus metadata.
pipe_links – Mapping of each node id to the list of its immediate neighbour node ids, mutated in place.
pipe_subend – List of neighbour node ids reached through a break_*_onlyinit_pipe boundary, mutated in place; used to seed the virtual end-of-pipe node.
transf – Id of the transform being processed.
simu – Sub-simulation date/index being processed.
mode – Either "forward" or "adjoint", the direction of the pipe currently being built (distinct from fetch_precursors, which selects which side of the transform to walk).
fetch_precursors – If True, walk the transform’s inputs/precursors (building the adjoint-direction node); if False, walk its outputs/successors (building the forward-direction node).

Returns:

None. transforms_ids, pipe_links and pipe_subend are mutated in place.

pycif.plugins.obsoperators.standard.transforms.utils.propagate_attributes.propagate_attributes(self, all_transforms, mapper, transform_id, backup_comps=None, next_did_nothing=False, previous_did_nothing=False, only_backwards=False, only_forwards=False, parent_transform=None, parent_trids=None)[source]#

Propagate attributes backward and forward through the pipe from a transform.

Recursively spreads a fixed set of transform attributes (domain, tracer, sampling/sparsity flags, dates, files, control/obsvect ancestry flags, etc.) between transform_id and its precursors/successors in mapper, via repeated calls to propagate_attribute(), until nothing changes anymore.

First initializes default values for attributes not explicitly set (initiate_default_attributes.initiate_default_attributes()). Then, for a first “priority” pass, propagates {attribute}_from_previous flags backward only, so downstream code knows which attributes must not be overwritten. It then propagates the main attribute batches, some restricted to one direction only (e.g. has_control_ancestor forwards, has_obsvect_successor backwards) and some in both directions.

If anything changed backward (or the caller’s forward pass changed something), recurses into precursors with only_backwards=True. Symmetrically, recurses into successors with only_forwards=True if anything changed forward. Finally, if anything changed in either direction, triggers init_default_transformations.init_default_transformations() to insert any reprojection/reindex/etc. transforms now made necessary by the newly propagated attributes.

Parameters:

self – The parent object (obs operator) holding the transform pipe.
all_transforms – Namespace holding all registered transform instances.
mapper – Dictionary mapping each transform id to its inputs/outputs/ precursors/successors metadata, mutated in place.
transform_id – Id of the transform to start propagating from.
backup_comps – Backup of components used to restore/compare state.
next_did_nothing – Whether the successor-side propagation that led to this call changed nothing (controls whether to keep recursing backward).
previous_did_nothing – Whether the precursor-side propagation that led to this call changed nothing (controls whether to keep recursing forward).
only_backwards – Restrict this call to backward propagation only.
only_forwards – Restrict this call to forward propagation only.
parent_transform – Id of the transform that triggered this call (used to scope which trids/neighbours are walked); None at the top-level entry point.
parent_trids – Optional list of trids to restrict propagation to; [] (as opposed to None) short-circuits the call entirely.

pycif.plugins.obsoperators.standard.transforms.utils.propagate_attributes.propagate_attribute(self, all_transforms, mapper, attributes, transform_id, only_backwards=False, only_forwards=False, next_did_nothing=False, previous_did_nothing=False, backup_comps=None, parent_transform=None, parent_trids=None, force_propagate=False, deal_boolean=None)[source]#

Propagate one group of attributes backward and/or forward for a transform.

Thin wrapper dispatching to propagate_backwards() (unless only_forwards) and propagate_forwards() (unless only_backwards) for the same set of attributes.

Parameters:

self – The parent object (obs operator) holding the transform pipe, exposing force_propagate_attributes.
all_transforms – Namespace holding all registered transform instances.
mapper – Dictionary mapping each transform id to its inputs/outputs/ precursors/successors metadata.
attributes – List of attribute names to propagate together (e.g. ["domain", "is_top", "is_lbc"]).
transform_id – Id of the transform being processed.
only_backwards – If True, only propagate to precursors.
only_forwards – If True, only propagate to successors.
next_did_nothing – Whether the previous backward-propagation step (from the successor’s perspective) changed nothing.
previous_did_nothing – Whether the previous forward-propagation step (from the precursor’s perspective) changed nothing.
backup_comps – Backup of components used to restore/compare state.
parent_transform – Id of the transform that triggered this propagation, used to scope the trids being walked; None when called from the top-level entry point.
parent_trids – Optional list of trids to restrict propagation to.
force_propagate – Whether to force propagation even when several neighbours disagree, as long as they are compatible once None values are ignored.
deal_boolean – Optional list (one entry per attribute) of "or"/ "and"/None describing how to merge boolean attributes across multiple neighbours.

Returns:

(did_nothing_forwards, list_updated_output_trids, did_nothing_backwards, list_updated_input_trids).

Return type:

tuple

pycif.plugins.obsoperators.standard.transforms.utils.propagate_attributes.propagate_backwards(self, all_transforms, mapper, attributes, transform_id, next_did_nothing=False, backup_comps=None, parent_transform=None, parent_trids=None, force_propagate=False, deal_boolean=None)[source]#

Propagate a group of attributes from successors down to this transform’s inputs.

First pulls attributes from each output trid’s successors into the corresponding output entry of transform_id (skipping trids that already carry a value from a different successor, in which case a conflict is logged and the attribute set to None, unless deal_boolean is used to merge boolean values instead). Then propagates those (possibly just-updated) output values down to the matching input trids, using skip_attributes() to decide whether enough neighbours agree, and skipping when {attribute}_from_previous forbids backward propagation.

Parameters:

self – The parent object (obs operator), exposing force_propagate_attributes.
all_transforms – Namespace holding all registered transform instances, used to check whether a recorded _from_successor reference is still a live transform.
mapper – Dictionary mapping each transform id to its inputs/outputs/ successors/inputs2outputs/outputs2inputs metadata.
attributes – List of attribute names to propagate together.
transform_id – Id of the transform being processed.
next_did_nothing – Whether the previous backward-propagation step triggered this call, used to decide whether debug prints fire.
backup_comps – Unused here, kept for interface consistency with propagate_attribute().
parent_transform – If given, restrict propagation to the output trids/successors linked to this specific parent transform (used for a targeted, single-hop propagation).
parent_trids – Optional list of trids to restrict propagation to.
force_propagate – Whether to force propagation to inputs even when not all outputs strictly agree, as long as non-None values are mutually compatible (see skip_attributes()).
deal_boolean – Optional list (one entry per attribute) of "or"/ "and"/None describing how to merge boolean attributes.

Returns:

(did_nothing, list_updated_input_trids) when parent_transform is None, describing whether anything changed (outputs or inputs) and which input trids were updated. When parent_transform is given, did_nothing instead reflects only whether inputs were updated (did_nothing_inputs).

Return type:

tuple

pycif.plugins.obsoperators.standard.transforms.utils.propagate_attributes.propagate_forwards(self, all_transforms, mapper, attributes, transform_id, previous_did_nothing=False, backup_comps=None, parent_transform=None, parent_trids=None, force_propagate=False, deal_boolean=None)[source]#

Propagate a group of attributes from precursors up to this transform’s outputs.

First pulls attributes from each input trid’s precursors into the corresponding input entry of transform_id (skipping trids that already carry a non-None value from a different precursor, in which case a conflict is logged and the attribute set to None). Then propagates those (possibly just-updated) input values up to the matching output trids, using skip_attributes() to decide whether enough neighbours agree.

Parameters:

self – The parent object (obs operator), exposing force_propagate_attributes.
all_transforms – Namespace holding all registered transform instances, used to check whether a recorded _from_precursor reference is still a live transform.
mapper – Dictionary mapping each transform id to its inputs/outputs/ precursors/inputs2outputs/outputs2inputs metadata.
attributes – List of attribute names to propagate together.
transform_id – Id of the transform being processed.
previous_did_nothing – Whether the previous forward-propagation step triggered this call, used to decide whether debug prints fire.
backup_comps – Unused here, kept for interface consistency with propagate_attribute().
parent_transform – If given, restrict propagation to the input trids/precursors linked to this specific parent transform (used for a targeted, single-hop propagation).
parent_trids – Unused here, kept for interface consistency with propagate_backwards().
force_propagate – Whether to force propagation to outputs even when not all inputs strictly agree, as long as non-None values are mutually compatible (see skip_attributes()).
deal_boolean – Unused here (boolean merging is only implemented for propagate_backwards()), kept for interface consistency.

Returns:

(did_nothing, list_updated_output_trids) when parent_transform is None, describing whether anything changed (inputs or outputs) and which output trids were updated. When parent_transform is given, did_nothing instead reflects only whether outputs were updated (did_nothing_inputs).

Return type:

tuple

pycif.plugins.obsoperators.standard.transforms.utils.propagate_attributes.skip_attributes(list_attributes, list_initialized=None, force_propagate=False, deal_boolean=None)[source]#

Check compatibility of attributes across neighbours and determine whether to skip.

Each element of list_attributes holds the attribute values from one neighbour (precursor or successor), in the same order as the attributes list passed by the caller. The elements are compared pairwise (via compare_attribute()) to decide whether they are consistent enough to propagate:

If empty, or all neighbours’ values are None, skip.
If list_initialized is given and not all neighbours are fully initialized, skip.
If there is a single neighbour, propagate its (non-None) values.
If all neighbours fully agree, propagate that common value.
Otherwise, unless force_propagate (or list_initialized is given) is set, skip. When forcing, propagate only if all non-None values across neighbours agree (using the first fully non-None neighbour as reference); otherwise skip due to ambiguity.

Parameters:

list_attributes – List of per-neighbour lists of attribute values, one inner list per neighbour, aligned with the caller’s attributes order.
list_initialized – Optional list (aligned with list_attributes) of booleans indicating whether each neighbour is fully initialized (e.g. all_successors_initialized).
force_propagate – Whether to force propagation despite partial disagreement, as long as non-None values are mutually compatible.
deal_boolean – Optional list (one entry per attribute) of "or"/ "and"/None describing how to compare boolean attributes.

Returns:

(skip, to_propagate, ind_to_propagate, any_none) where skip is True if attributes should not be propagated, to_propagate is the list of values to propagate (empty when skipping), ind_to_propagate is the index (within list_attributes) the values were taken from (None when skipping), and any_none is True if any neighbour had a None value.

Return type:

tuple

pycif.plugins.obsoperators.standard.transforms.utils.propagate_attributes.compare_attribute(attr1, attr2, deal_boolean=None, return_boolean_value=False)[source]#

Compare two attribute values, or merge them if they are booleans/dicts of booleans.

When deal_boolean is set ("or" or "and"):

If return_boolean_value is False, always returns True (the values are always considered “comparable” under boolean merging).
If return_boolean_value is True, returns the merged value: for plain booleans, attr1 or attr2/attr1 and attr2; for dictionaries of booleans (used e.g. for per-species flags), merges key-wise with or (missing keys default to False); None values are treated as absorbing (the other value is returned as-is).

Otherwise (deal_boolean is None), performs a regular equality check: type mismatch is False; dicts are compared key-by-key (recursively); lists/tuples/arrays are compared element-wise with numpy.all(); pandas.DataFrame uses equals(); anything else uses ==.

Parameters:

attr1 – First value to compare or merge.
attr2 – Second value to compare or merge.
deal_boolean – "or", "and" or None, selecting boolean-merge mode versus regular equality comparison.
return_boolean_value – In boolean-merge mode, whether to return the merged value instead of a comparability flag.

Returns:

bool, True if the values are equal. In boolean-merge mode with return_boolean_value=False: True. In boolean-merge mode with return_boolean_value=True: the merged boolean (or dict of booleans) value.

Return type:

In regular comparison mode

pycif.plugins.obsoperators.standard.transforms.utils.propagate_parameters.propagate_parameters(all_transforms, mapper)[source]#

Propagate a fixed set of scalar parameters backward through the pipe.

For force_loadin, loadin_perturb_full_vertical and surface_level, sweeps the pipe backwards (from the last to the first transform): whenever an output value differs from its default, it is copied to the matching input, and then to the corresponding precursors’ outputs.

Parameters:

all_transforms – Namespace holding all registered transform instances, ordered as they appear in the pipe.
mapper – Dictionary mapping each transform id to its inputs/outputs/ precursors metadata, mutated in place.

pycif.plugins.obsoperators.standard.transforms.utils.submit_and_kill.submit_and_kill(self)[source]#

Resubmit the current run as a new job and terminate this process.

Used when a run is about to exceed its authorized wall-clock time (self.autokill_time): writes a new YAML configuration file next to the original one, named {name}_resubmit_{NNN}.yml, with obsoperator.autorestart forced to True (and, if self.rename_resubmit_logfile is set, a renamed logfile entry), submits it via self.platform.submit_job, then exits the current process.

Parameters:

self – The obs operator, exposing reference_instances, max_resubmissions, autokill_time, rename_resubmit_logfile, from_yaml and platform.

Raises:

CifRuntimeError – If the number of previous re-submissions (detected
from existing _resubmit_NNN files) has reached –
self.max_resubmissions` –

pycif.plugins.obsoperators.standard.transforms.utils.dask.individual_transform.do_individual_transform(*args, **kwargs)[source]#

Runs a single transform, batching its log output.

Thin wrapper around _do_individual_transform: dask worker threads run many transforms concurrently, and without batching their debug/info calls interleave line-by-line with other threads’ output. See pycif.utils.check.batched_logging.

pycif.plugins.obsoperators.standard.transforms.utils.dask.individual_transform.input_output_msg(trid_list: List[Tuple[str, str]]) → str[source]#

Format a list of tracer IDs as a human-readable multi-line string.

Groups (component, parameter) pairs by component and produces one line per component, listing its parameters (or nothing after the colon when the only parameter is the empty string).

Parameters:: trid_list – List of (component, parameter) tuples.
Returns:: The formatted, newline-joined description.
Return type:: str

pycif.plugins.obsoperators.standard.transforms.utils.dask.init_dask.plot_task_graph(tasks, dependencies, rundir)[source]#

Render the resolved transform DAG as an interactive HTML graph.

dask’s own visualize() rasterizes through Graphviz, which becomes unreadable past a few dozen nodes. pyvis (vis.js) instead produces a pannable/zoomable/draggable HTML page with hover tooltips, which stays usable with the hundreds of transform tasks a typical CIF run builds.

networkx and pyvis are optional (pip install networkx pyvis or the graph extra): imported lazily here so their absence never breaks a run that doesn’t request plotting.

Parameters:

tasks (dict) – mapping of task name -> delayed object, as built by init_dask’s main loop.
dependencies (dict) – mapping of task name -> list of precursor task names, as built by init_dask before the main loop.
rundir (str) – directory to write dask_graph.html into.

pycif.plugins.obsoperators.standard.transforms.utils.dask.init_dask.init_dask(self, pipe_links, mode='fwd', do_simu=True, onlyinit=False, check_transforms=False, adj_test_threshold=10, save_debug=False, ignore_exceptions=False, ref_fwd_dir='', run_id=0)[source]#

Build and execute the Dask task graph for one direction of the pipe.

Inverts pipe_links (which maps each node to its successors) into a dependencies mapping (node -> precursors), topologically sorts it, augments it with extra serialization dependencies coming from force_dump/force_loadout I/O constraints (update_dependencies_with_io()), then walks the sorted nodes to build one Dask delayed task per transform (add_transform()), skipping dead branches and nodes with no resolved dependency. Also attaches a “dry run” metadata dependency on the opposite-direction counterpart of each node so that side-effect-only initialization runs before the real computation.

If self.plot_dask_graph is set, writes an interactive HTML graph of the resulting task DAG (plot_task_graph()).

Finally triggers the computation of the graph’s final task (final_toobsvect in forward/tangent-linear mode, final_fromcontrol in adjoint mode), using the scheduler resolved from self.dask_mode (_resolve_dask_scheduler), restoring the root logger’s level afterwards in case a worker left it altered.

Parameters:

self – The obs operator, exposing workdir, transform_pipe, datei, controlvect, obsvect, plot_dask_graph and dask_mode.
pipe_links – Mapping of each (ddi, transform, direction) node to the list of its successor nodes, as built by fwd_pipe.fwd_adj_pipe().
mode – Execution mode: "fwd", "tl" or "adj".
do_simu – Whether to actually run the underlying model simulations.
onlyinit – Whether to only initialize (not execute) every transform.
check_transforms – Whether to keep debug information needed by the adjoint/tangent-linear consistency test.
adj_test_threshold – Threshold passed through to transforms for the adjoint/tangent-linear consistency test.
save_debug – Whether to dump debug datasets for each transform.
ignore_exceptions – Whether to swallow exceptions raised by transforms unless required outputs are missing.
ref_fwd_dir – Path to the reference forward run directory, used by adjoint/tangent-linear transforms that need forward outputs.
run_id – Numeric id used to name this run’s sub-directory under {self.workdir}/obsoperator/{mode}_{run_id}.

Raises:

CifImportError – If Dask is not installed.

Returns:

None. The task graph is executed for its side effects (writing to the shared datastore via each transform’s forward/ adjoint method).

pycif.plugins.obsoperators.standard.transforms.utils.dask.init_dask.entry_point()[source]#

Return an empty result shaped like a real transform task’s output.

Used as the result of the virtual final_toobsvect/ final_fromcontrol graph nodes, which have no actual transform to run but must still produce a value in the shape downstream/upstream tasks expect.

Returns:: {"main": {"inputs": {}, "outputs": {}}, "meta": []}.
Return type:: dict

pycif.plugins.obsoperators.standard.transforms.utils.dask.init_dask.update_dependencies_with_io(self, static_order, end_point, start_point, dependencies)[source]#

Add extra Dask dependencies to serialize transforms sharing dump/load files.

Some transforms write/read shared files on disk (via dump2inputs in forward mode when a trid is flagged force_dump, or loadfromoutputs in adjoint mode when a trid is flagged force_loadout). Because Dask has no visibility into that filesystem side effect, this walks every node’s mapper for such flags and, for each declared dependency (dumpin_dependencies/loadout_dependencies), adds an explicit edge in dependencies between the corresponding dump2inputs/loadfromoutputs transform ids (resolved via dump2inputs_ids/loadfromoutputs_ids in the mapper) so that Dask executes them in the required order.

Parameters:

self – The obs operator, exposing transform_pipe.mapper.
static_order – Topologically sorted list of (ddi, transform, direction) nodes to scan.
end_point – The virtual end-of-pipe node, skipped during the scan.
start_point – The virtual start-of-pipe node, skipped during the scan.
dependencies – Mapping of each node to the list of its precursor nodes, mutated in place with the extra I/O-derived edges.

Raises:

CifError – If a transform declares a dump/load dependency on
another trid for which no corresponding –
dump2inputs`/loadfromoutputs transform was created –

Returns:

The same dependencies mapping, augmented in place.

Return type:

dict

pycif.plugins.obsoperators.standard — API reference

Contents

pycif.plugins.obsoperators.standard — API reference#

`pycif.plugins.obsoperators.standard` — API reference#