Source code for pycif.plugins.obsoperators.random.obsoper

import datetime
import os
from ....utils import path
from logging import info, debug
import numpy as np




def obsoper(
    self,
    controlvect,
    obsvect,
    mode,
    run_id=0,
    datei=datetime.datetime(1979, 1, 1),
    datef=datetime.datetime(2100, 1, 1),
    workdir="./",
    reload_results=False,
    check_transforms=False,
    ignore_exceptions=False,
    force_fetch_results=False,
    **kwargs
):
    """Apply the random observation operator in forward, tangent-linear or adjoint mode.

    Implements a simple matrix–vector product using the random matrix
    ``self.H_matrix`` built by :func:`~pycif.plugins.obsoperators.random.ini_data`.
    Because no physical model is involved, the operator is instantaneous and
    exact in all three modes.

    **Forward** (``mode='fwd'``):

    .. math::

        \\mathbf{y}_\\text{sim} = \\mathbf{H}\\,\\mathbf{x}

    Sets ``obsvect.ysim`` and returns *obsvect*.

    **Tangent-linear** (``mode='tl'``):

    .. math::

        \\mathbf{y}_\\text{sim} = \\mathbf{H}\\,\\mathbf{x}, \\qquad
        \\delta\\mathbf{y} = \\mathbf{H}\\,\\delta\\mathbf{x}

    Sets both ``obsvect.ysim`` and ``obsvect.dy``; returns *obsvect*.

    **Adjoint** (``mode='adj'``):

    .. math::

        \\delta\\mathbf{x} = \\mathbf{H}^\\top\\,\\delta\\mathbf{y}

    Sets ``controlvect.dx``; returns *controlvect*.

    Args:
        self (ObsOperator): plugin instance; must have ``H_matrix`` set by a
            prior call to :func:`~pycif.plugins.obsoperators.random.ini_data`.
        controlvect (ControlVect): control-vector object.  ``controlvect.x``
            is read in ``fwd`` / ``tl`` modes; ``controlvect.dx`` is read in
            ``tl`` mode and written in ``adj`` mode.
        obsvect (ObsVect): observation-vector object.  ``obsvect.ysim`` is
            written in ``fwd`` / ``tl`` modes; ``obsvect.dy`` is read in
            ``adj`` mode and written in ``tl`` mode.
        mode (str): execution mode — one of ``'fwd'``, ``'tl'``, or ``'adj'``.
        run_id (int, optional): run identifier (unused; present for interface
            compatibility). Defaults to ``0``.
        datei (datetime.datetime, optional): simulation start date (unused;
            present for interface compatibility).
            Defaults to ``datetime.datetime(1979, 1, 1)``.
        datef (datetime.datetime, optional): simulation end date (unused;
            present for interface compatibility).
            Defaults to ``datetime.datetime(2100, 1, 1)``.
        workdir (str, optional): working directory (unused; present for
            interface compatibility). Defaults to ``"./"``
        reload_results (bool, optional): reload pre-computed results (unused).
            Defaults to ``False``.
        check_transforms (bool, optional): validate transform consistency
            (unused). Defaults to ``False``.
        ignore_exceptions (bool, optional): swallow non-fatal errors (unused).
            Defaults to ``False``.
        force_fetch_results (bool, optional): if ``True``, raise
            :exc:`IOError` immediately to signal that results must be fetched
            from a previous run rather than recomputed. Defaults to ``False``.
        **kwargs: additional keyword arguments (ignored).

    Returns:
        ObsVect: in ``'fwd'`` and ``'tl'`` modes — the updated *obsvect*.

        ControlVect: in ``'adj'`` mode — the updated *controlvect*.

    Raises:
        IOError: if ``force_fetch_results`` is ``True``.
    """

    if force_fetch_results:
        raise IOError

    if mode == "fwd":
        obsvect.ysim = np.dot(self.H_matrix, controlvect.x)
        return obsvect

    elif mode == "tl":
        obsvect.ysim = np.dot(self.H_matrix, controlvect.x)
        obsvect.dy = np.dot(self.H_matrix, controlvect.dx)
        return obsvect

    elif mode == "adj":
        controlvect.dx = np.dot(self.H_matrix.T, obsvect.dy)
        return controlvect