otf.optim.gradient

Modules:

Name	Description
adjoint
gradient_computer
sensitivity

Classes:

Name	Description
AdjointGradient	Compute parameter gradients using adjoint-based methods.
GradientComputer	Base class for objects that compute gradients of the assimilation error.
SensitivityGradient	Compute gradients using sensitivity (forward) equations.

AdjointGradient

AdjointGradient(
    system: BaseSystem,
    update_option: UpdateOption = UpdateOption.asymptotic,
    solver: tuple[type[SinglestepSolver | MultistepSolver]]
    | type[SinglestepSolver | MultistepSolver]
    | None = None,
    dt: float | None = None,
    interval_fraction: float = 1,
)

Bases: GradientComputer

Compute parameter gradients using adjoint-based methods.

Supports different update strategies controlled by UpdateOption.

Initialize an AdjointGradient.

Parameters:

Name	Type	Description	Default
system	BaseSystem	BaseSystem instance to analyze.	required
update_option	UpdateOption	Which adjoint/update method to use (UpdateOption).	asymptotic
solver	tuple[type[SinglestepSolver | MultistepSolver]] | type[SinglestepSolver | MultistepSolver] | None	Solver class or tuple of solver classes used when simulation-based adjoint computation is selected (complete or unobserved).	None
dt	float | None	Time-step used with the solver (required when solver is used).	None
interval_fraction	float	Fraction of the input time series to use for gradient computation (value in (0, 1]).	1

Source code in src/otf/optim/gradient/adjoint.py

def __init__(
    self,
    system: BaseSystem,
    update_option: UpdateOption = UpdateOption.asymptotic,
    solver: tuple[type[SinglestepSolver | MultistepSolver]]
    | type[SinglestepSolver | MultistepSolver]
    | None = None,
    dt: float | None = None,
    interval_fraction: float = 1,
):
    """Initialize an `AdjointGradient`.

    Parameters
    ----------
    system
        `BaseSystem` instance to analyze.
    update_option
        Which adjoint/update method to use (`UpdateOption`).
    solver
        Solver class or tuple of solver classes used when simulation-based
        adjoint computation is selected (`complete` or `unobserved`).
    dt
        Time-step used with the solver (required when `solver` is used).
    interval_fraction
        Fraction of the input time series to use for gradient computation
        (value in (0, 1]).
    """
    super().__init__(system)

    if not (0 < interval_fraction <= 1):
        raise ValueError(
            "`interval_fraction` should be in (0, 1]"
            f" (was {interval_fraction})"
        )
    self._interval_fraction = interval_fraction

    self._compute_adjoint = self._set_up_adjoint_method(update_option)

    if update_option is not UpdateOption.asymptotic:
        if dt is None:
            raise ValueError("`dt` must not be None for this update option")
        if solver is None:
            raise ValueError(
                "`solver` must not be None for the given update option"
            )

        adjoint_system = self._set_up_adjoint_system(system, update_option)

        self._dt = dt
        self._solver = self._set_up_solver(adjoint_system, solver)

GradientComputer

GradientComputer(system: BaseSystem)

Base class for objects that compute gradients of the assimilation error.

Subclasses implement compute_gradient(observed_true, assimilated).

Create a GradientComputer bound to system.

Parameters:

Name	Type	Description	Default
system	BaseSystem	BaseSystem instance for which gradients are computed.	required

Methods:

Name	Description
set_weight	Weight the error using a (positive definite) matrix.

Source code in src/otf/optim/gradient/gradient_computer.py

def __init__(self, system: BaseSystem):
    """Create a `GradientComputer` bound to `system`.

    Parameters
    ----------
    system
        `BaseSystem` instance for which gradients are computed.
    """
    self._system = system
    self._weight = None

set_weight

set_weight(weight: jndarray | None)

Weight the error using a (positive definite) matrix.

Use this matrix to weight the error (defined as the two-norm of the difference between the data assimilated system state and the observed portion of true system state). This accordingly weights the derivative of the error with respect to the unknown parameters of the data assimilated system.

If assimilating a system with noisy measurements, it is common to use the inverse of the covariance matrix as the weight.

Parameters:

Name	Type	Description	Default
weight	jndarray | None	Square array. Each dimension should be equal to the number of observed variables. Pass None to unset the weight (equivalently, set the weight to the identity).	required

Source code in src/otf/optim/gradient/gradient_computer.py

def set_weight(self, weight: jndarray | None):
    """Weight the error using a (positive definite) matrix.

    Use this matrix to weight the error (defined as the two-norm of the
    difference between the data assimilated system state and the observed
    portion of true system state). This accordingly weights the derivative
    of the error with respect to the unknown parameters of the data
    assimilated system.

    If assimilating a system with noisy measurements, it is common to use
    the *inverse* of the covariance matrix as the weight.

    Parameters
    ----------
    weight
        Square array. Each dimension should be equal to the number of
        observed variables. Pass `None` to unset the weight (equivalently,
        set the weight to the identity).
    """
    self._weight = weight

SensitivityGradient

SensitivityGradient(
    system: BaseSystem,
    update_option: UpdateOption = UpdateOption.last_state,
    solver: tuple[type[SinglestepSolver | MultistepSolver]]
    | type[SinglestepSolver | MultistepSolver]
    | None = None,
    dt: float | None = None,
    use_unobserved_asymptotics: bool = False,
)

Bases: GradientComputer

Compute gradients using sensitivity (forward) equations.

Different UpdateOptions select how the sensitivity information is assembled (last state, mean state, mean gradient, or complete simulation).

Initialize a SensitivityGradient.

Parameters:

Name	Type	Description	Default
system	BaseSystem	BaseSystem instance to analyze.	required
update_option	UpdateOption	Strategy for forming the gradient (UpdateOption).	last_state
solver	tuple[type[SinglestepSolver | MultistepSolver]] | type[SinglestepSolver | MultistepSolver] | None	Solver class or tuple of solver classes used when the complete update option is selected.	None
dt	float | None	Time-step used with the solver (required when solver is used).	None
use_unobserved_asymptotics	bool	When True attempt to use asymptotic information from unobserved state components (experimental).	False

Source code in src/otf/optim/gradient/sensitivity.py

def __init__(
    self,
    system: BaseSystem,
    update_option: UpdateOption = UpdateOption.last_state,
    solver: tuple[type[SinglestepSolver | MultistepSolver]]
    | type[SinglestepSolver | MultistepSolver]
    | None = None,
    dt: float | None = None,
    use_unobserved_asymptotics: bool = False,
):
    """Initialize a `SensitivityGradient`.

    Parameters
    ----------
    system
        `BaseSystem` instance to analyze.
    update_option
        Strategy for forming the gradient (`UpdateOption`).
    solver
        Solver class or tuple of solver classes used when the `complete`
        update option is selected.
    dt
        Time-step used with the solver (required when `solver` is used).
    use_unobserved_asymptotics
        When True attempt to use asymptotic information from unobserved
        state components (experimental).
    """
    super().__init__(system)

    self._update_option = update_option
    self.compute_gradient = self._set_up_gradient(update_option)

    if update_option is UpdateOption.complete:
        if dt is None:
            raise ValueError("`dt` must not be None for this update option")
        if solver is None:
            raise ValueError(
                "`solver` must not be None for the given update option"
            )

        sensitivity_system = _SensitivitySystem(system)

        self._dt = dt
        self._solver = self._set_up_solver(sensitivity_system, solver)

    self._use_unobserved_asymptotics = use_unobserved_asymptotics