pycif.plugins.modes.response_functions — API reference

Configuration reference: response_functions plugin

pycif.plugins.modes.response_functions.analytical_inversion.compute_inversion(H: ndarray, B: ndarray, R: ndarray, xb: ndarray, dy: ndarray, use_woodbury_identity: Literal[True, False, 'auto']) → Tuple[ndarray, ndarray, ndarray]

Perform the analytical inversion using either straightforward matrix inversion or the Woodbury matrix identity.

Parameters:

• H (2D array (M, N)) – The H matrix.

• B (2D array (N, N)) – The B matrix.

• R (2D array (M, M)) – The R matrix.

• xb (1D array (N,)) – The xb vector.

• dy (1D array (M,)) – The dy vector.

• use_woodbury_identity (bool or "auto") – Whether to use the Woodbury matrix identity for inversion.

Returns:

Pa, xa, and ya.

Return type:

(2D array (N,N), 1D array (N,), 1D array (M,))

pycif.plugins.modes.response_functions.analytical_inversion.analytical_inversion(self: Any, xb: ndarray, h_matrix: ndarray, controlvect: Any, obsvect: Any) → None

Perform an analytical inversions and dumps the matrices an vectors if the ‘analytical_inversion’ input argument is set to ‘true’, otherwise only dumps the H matrix

Parameters:

• self (Mode) – the mode plugin

• xb (1D array) – control vector prior

• h_matrix (2D array) – H matrix

• controlvect (ControlVect) – the control vector plugin

• obsvect (ObsVect) – the observation vector plugin

pycif.plugins.modes.response_functions.execute.execute(self, **kwargs)

Run the response-functions mode: compute H-matrix columns and optionally invert.

Orchestrates the full response-function workflow:

1. Optionally runs a reference forward simulation to populate the obs-vector 'sim' column.

2. Initialises and submits one pyCIF forward run per control-vector dimension (the base functions / response functions).

3. Assembles the resulting H matrix.

4. If analytical_inversion is set, performs a direct Bayesian inversion xa = xb + K(y − H·xb) and dumps the posterior.

Parameters:

• self (Plugin) – mode plugin carrying all configuration attributes (workdir, datei, datef, run_mode, dryrun, analytical_inversion, reload_h_matrix, etc.).

• **kwargs – forwarded verbatim to the observation operator.

Returns:

the observation vector populated with simulated values (prior and/or posterior, depending on analytical_inversion).

Return type:

ObsVect

pycif.plugins.modes.response_functions.h_matrix.get_obsvect_var_name(run_mode: Literal['fwd', 'tl']) → Literal['sim', 'sim_tl']

Map a run mode string to the corresponding obs-vector column name.

Parameters:

run_mode – 'fwd' for a full forward run, 'tl' for the tangent-linear operator.

Returns:

'sim' for run_mode='fwd', 'sim_tl' for 'tl'.

Raises:

ValueError – if run_mode is not 'fwd' or 'tl'.

pycif.plugins.modes.response_functions.h_matrix.init_h(controlvect: Any, obsvect: Any) → ndarray

Allocate a zero-filled H matrix of the correct shape.

Parameters:

• controlvect (ControlVect) – control vector plugin providing dim.

• obsvect (ObsVect) – observation vector plugin providing dim.

Returns:

zero array of shape (obs_dim, control_dim).

Return type:

np.ndarray

pycif.plugins.modes.response_functions.h_matrix.build_h(self: Any, base_function_list: Iterable[BaseFunction]) → ndarray

Iterates over all response functions and observation vector tracers to fill the H matrix

Parameters:

• self (Mode) – the mode plugin

• base_function_list (list of BaseFunction) – base functions

Returns:

H matrix

Return type:

2D array

pycif.plugins.modes.response_functions.h_matrix.read_h_matrix(self: Any, path_list: List[str]) → ndarray

pycif.plugins.modes.response_functions.h_matrix.fill_obsvect(self: Any, h_matrix: ndarray, xb: ndarray) → None

Iterates over all observation vector tracers to fill the observation vector in-place with the response function cntributions in the H matrix

Parameters:

• self (Mode) – the mode plugin

• h_matrix (2D array) – the filled H matrix

• xb (1D array) – control vector prior

pycif.plugins.modes.response_functions.h_matrix.dump_obsvect_decomp(self: Any, h_matrix: ndarray, decompdir: str) → None

Iterates over all control vector tracers to get the decomposition of their contribution to each observations. Dumps the results in NetCDF files

Parameters:

• self (Mode) – the mode plugin

• h_matrix (2D array) – the filled H matrix

pycif.plugins.modes.response_functions.init_base_functions.split_by_parameter(self: Any, batch_list: List[BaseFunctionSamplingBatch]) → List[BaseFunctionSamplingBatch]

Splits the base function sampling batches by observation parameter

Parameters:

• self (Mode) – the mode plugin

• batch_list (list of BaseFunctionSamplingBatch) – base function sampling batches

Returns:

splitted base function sampling batches

Return type:

list of BaseFunctionSamplingBatch

pycif.plugins.modes.response_functions.init_base_functions.split_sampling_batches(self: Any, batch_list: List[BaseFunctionSamplingBatch], n: int) → List[BaseFunctionSamplingBatch]

Splits the base function sampling batches

Parameters:

• self (Mode) – the mode plugin

• batch_list (list of BaseFunctionSamplingBatch) – base function sampling batches

• n (int) – maximum batch size

Returns:

splitted base function sampling batches

Return type:

list of BaseFunctionSamplingBatch

pycif.plugins.modes.response_functions.init_base_functions.init_base_functions(self: Any) → List[BaseFunction] | List[BaseFunctionSamplingBatch]

Get the time window of each of the controlvect element and initialize the corresponding base functions or batch sampling of base functions

Parameters:

self (Mode) – the mode plugin

Returns:

simulations to run

Return type:

list of BaseFunction or list of BaseFunctionSamplingBatch

pycif.plugins.modes.response_functions.jobs.submit_job(self: Any, command: str, job_file: str) → str

Submit a job, if dryrun only create the job file

Parameters:

• self (Mode) – the mode plugin

• command (str) – command to execute

• job_file (str) – path to the job file

Returns:

job id

Return type:

str

pycif.plugins.modes.response_functions.jobs.wait_jobs(self: Any, job_id_list: List[str]) → None

Wait for all jobs to finish

Parameters:

• platform (Platform) – the mode plugin

• job_id_list (list of str) – ids of the job to wait for

pycif.plugins.modes.response_functions.jobs.job_batches(base_function_list: Iterable[BaseFunctionType], batch_size: int) → Generator[List[BaseFunctionType], None, None]

Cut the base functions in batches

Parameters:

• base_function_list (iterable of base functions) – base functions

• batch_size (int) – batch size

Yields:

list of base functions – batch of base functions

pycif.plugins.modes.response_functions.jobs.run_jobs_in_batches(self: Any, base_function_list: List[BaseFunctionType]) → None

Iterate over the control vector elements individualy or with batch sample and run the corresponding base function in job batches

Parameters:

• self (Mode) – the mode plugin

• base_function_list (iterable of base functions) – base functions

pycif.plugins.modes.response_functions.periods.apply_spin_down(series: Series, spin_down: str) → DatetimeIndex

Applies spin down in place to the ‘date_end’ column of DataFrame df

Parameters:

• series (pd.Series) – Series of Timestamps, use a view of a DataFrame column, ex: apply_spin_down(df.loc[a:b, colname], spin_down)

• spin_down (str) – valid pandas period alias (1D, 1M, …)

pycif.plugins.modes.response_functions.periods.get_controlvect_periods(self: Any) → DataFrame

Computing the time period covered by each element of the control vector

Parameters:

self (Mode) – the mode plugin

Returns:

start and end times of the period

Return type:

array, array

pycif.plugins.modes.response_functions.periods.groupby_period(date_start: ndarray, date_end: ndarray) → Generator[Tuple[ndarray, datetime64, datetime64], None, None]

Group control vector per time periods

Parameters:

• date_start (array of datetime64) – start date of periods

• date_end (array of datetime64) – end date of periods

Yields:

array of int, datetime64, datetime64 –

control vector indices within the

period, start date of the period, end date of the period

pycif.plugins.modes.response_functions.ref_forward.run_ref_forward(self: Any) → str

Run the reference forward and dumps the observation vector

Parameters:

self (Mode) – the mode plugin

Returns:

path to the reference forward directory

Return type:

str

pycif.plugins.modes.response_functions.ref_forward.get_inicond_from_ref_forward(self: Any, datei: datetime) → Tuple[str, Dict[str, Any]]

Get the path to the reference forward run initial conditions (forward) file for running a response functions starting on datetime ‘datei, and returns the model initial condition component in the YAML configuration file and the data to use the initial conditions file

Used for response functions in tangent mode on a period that is simulated a sub-period of the full simulation period

The model used needs to be implemented here, be carefull to keep all control vector arguments when implementing a new model here.

Parameters:

• self (Mode)

• datei (datetime.datetime) – start date of the response function simulation period

Returns:

component of the model initial condition paragraph in the YAML configuration,

and data containing the path to the reference forward run initial conditions file

Return type:

str, dict

pycif.plugins.modes.response_functions.base_function.base_function.to_datetime(dt: datetime | Timestamp | datetime64) → datetime

A function to convert various datetime-like objects to standard Python datetime objects.

Parameters:

dt (DatetimeLike) – The input datetime-like object to be converted.

Returns:

The standard Python datetime object after conversion.

Return type:

datetime.datetime

Raises:

• ValueError – If the input datetime-like object cannot be converted to a standard datetime object.

• TypeError – If the input type is unexpected and cannot be converted to a datetime object.

class pycif.plugins.modes.response_functions.base_function.base_function.BaseFunction(mode: Any, index: int, component: str, parameter: str, date_start: datetime | Timestamp | datetime64, date_end: datetime | Timestamp | datetime64)

Bases: object

property rundir: str

property outdir: str

property obsdir: str

property controlvect_path: str

property yaml_config_path: str

is_ignored() → bool

True if the base fonction (component, parameter) couple is in the ‘ignore_tracer’ input argument

has_run_succesfully() → bool

True is the base function has run successfully (output directory is present)

update_inicond_date() → None

Updates the initial conditions tracers date if those are in the control vector and the response function start date is different from the simulation window initial date

dump_controlvect() → None

Updates the controlvector and dumps it

fetch_obsvect() → None

Fetch the response function outputs (observation vector) and move it to its ‘obsvect’ directory

flush() → None

Delete some temporary files to free some disk space

pycif.plugins.modes.response_functions.base_function.base_function_batch.chain_outputs(in_out_dict: Dict[Tuple[str, str], Set[Tuple[str, str]]], list_output: Set[Tuple[str, str]]) → Set[Tuple[str, str]]

For each tracer id in ‘list_output’, if it is an input tracer (a key in ‘in_out_dict’) then replace it with the corresponding set of output tracers. This is done recursively.

Parameters:

• in_out_dict (dict((str, str) -> set of (str, str))) – dictionary with input tracers as keys and a set of output tracers as values

• list_output (set of (str, str)) – set of output tracers

Returns:

chained outputs

Return type:

set of (str, str)

pycif.plugins.modes.response_functions.base_function.base_function_batch.get_transforms_in_out_mapping(controlvect: Any) → Dict[Tuple[str, str], Set[Tuple[str, str]]]

For each input tracer of each transform in the controlvector ‘transform_pipe’, gets the associated output tracer (ignoring the output tracer component)

Parameters:

controlvect (ControlVect) – the control vector

Returns:

dictionary with input tracers as

keys and a set of output tracers as values

Return type:

dict((str, str) -> set of (str, str))

class pycif.plugins.modes.response_functions.base_function.base_function_batch.BaseFunctionSamplingBatch(mode: Any, batch_index: int, indices: Sequence[int] | ndarray, components: Sequence[str], parameters: Sequence[str], date_start: datetime | Timestamp | datetime64, date_end: datetime | Timestamp | datetime64)

Bases: BaseFunction, Iterable

iter_all() → Iterator[BaseFunction]

All base functions within the batch (ignored or not)

property obsdir: str

is_ignored() → bool

True if the base fonction (component, parameter) couple is in the ‘ignore_tracer’ input argument

dump_controlvect() → None

Updates the controlvector and dumps it

fetch_obsvect() → None

Fetch the response function outputs (observation vector) and move it to its ‘obsvect’ directory

pycif.plugins.modes.response_functions.base_function.yaml_config.dump_yaml_config(self: Any, base_function: BaseFunction | BaseFunctionSamplingBatch) → None

Generates and dump a YAML configuration file for a base function

Parameters:

• self (Mode) – the mode plugin

• base_function (BaseFunction or BaseFunctionSamplingBatch) – the base function or sampling batch of base functions