pycif.plugins.modes.response_functions — API reference

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][source]#

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[source]#

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)[source]#

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'][source]#

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[source]#

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[source]#

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[source]#
pycif.plugins.modes.response_functions.h_matrix.fill_obsvect(self: Any, h_matrix: ndarray, xb: ndarray) → None[source]#

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[source]#

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][source]#

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][source]#

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][source]#

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[source]#

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[source]#

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][source]#

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[source]#

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[source]#

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[source]#

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][source]#

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[source]#

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]][source]#

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