pycif.plugins.obsoperators.standard — API reference

Configuration reference: standard plugin

pycif.plugins.obsoperators.standard.check.check_inputs(inputs, mode)

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)

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

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

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

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)

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)

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)

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)

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)

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

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)

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)

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)

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)

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)

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)

Arrange all transforms into ordered forward and adjoint execution pipes.

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

1. Propagating sub-simulation periods from each transform to its precursors and successors via default_subsimus().

2. Building a dependency graph and walking it in forward order with fwd_adj_pipe() (mode='forward').

3. 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)

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

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

pycif.plugins.obsoperators.standard.transforms.utils.add_default.generate_internal_pipe(new_id, mapper)

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)

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) – _description_

• tmp_outputs (dict) – _description_

• transform_mode (str) – _description_

• transform_onlyinit (bool) – _description_

• mapper (dict) – _description_

• check_transforms (bool, optional) – _description_. Defaults to False.

Raises:

• Exception – _description_

• TypeError – _description_

Returns:

_description_

Return type:

_type_

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

pycif.plugins.obsoperators.standard.transforms.utils.check_adjtltest.check_adjtltest(self, period_order, mapper, transform_pipe)

pycif.plugins.obsoperators.standard.transforms.utils.check_datavect.check_datavect(self, all_transforms, backup_comps, mapper)

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

pycif.plugins.obsoperators.standard.transforms.utils.connect_pipes.connect_pipes(all_transforms, mapper, transform)

Connect transforms based on their inputs and outputs

pycif.plugins.obsoperators.standard.transforms.utils.connect_pipes.prune_dead_branches(all_transforms, mapper, skip_transform)

pycif.plugins.obsoperators.standard.transforms.utils.default_subsimus.default_subsimus(all_transforms, mapper)

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)

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)

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)

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

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)

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

pycif.plugins.obsoperators.standard.transforms.utils.init_entry.init_entry(self, all_transforms, backup_comps, mapper)

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)

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)

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)

pycif.plugins.obsoperators.standard.transforms.utils.initiate_default_attributes.initiate_default_attributes(all_transforms, mapper, transform_id)

Initiate default values.

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)

Find the precursors/successors in backward and forward mode

Parameters:

• self

• all_transforms

• transforms_ids

• mapper

• pipe_links

• transf

• simu

• mode

• precursors

• fetch_precursors

Returns:

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)

Propagate attributes backward and forward when initializing a new transform.

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)

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)

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)

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

Check compatibility of attributes and determines whether to skip

Parameters:

• list_attributes (_type_) – _description_

• list_initialized (_type_) – _description_

• force_propagate (bool, optional) – _description_. Defaults to False.

Returns:

• List of attributes are not compatible

• List of attributes to propagate

• True if all Nones

Return type:

bool, list, bool

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

pycif.plugins.obsoperators.standard.transforms.utils.propagate_dates.propagate_dates(all_transforms, mapper)

pycif.plugins.obsoperators.standard.transforms.utils.propagate_parameters.propagate_parameters(all_transforms, mapper)

pycif.plugins.obsoperators.standard.transforms.utils.submit_and_kill.submit_and_kill(self)

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

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

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

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)

pycif.plugins.obsoperators.standard.transforms.utils.dask.init_dask.entry_point()

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