Source code for pycif.plugins.obsoperators.standard.ndarray_wrapper

import numpy as np
from ....utils.check.errclass import CifError




def forward(self, x: np.ndarray, reload_results=False) -> np.ndarray:
    self._step += 1
    self.controlvect.x = x
    obsvect = self.obsoper(
        self.controlvect, self.obsvect,
        "fwd",
        datei=self.datei, datef=self.datef, workdir=self.workdir,
        reload_results=reload_results,
        run_id=self._step,
    )
    self.obsvect = obsvect
    return np.asarray(obsvect.ysim, dtype=np.float64)





def tangent_linear(self, x: np.ndarray, reload_results=False) -> np.ndarray:
    self._step += 1
    self.controlvect.x = x
    obsvect = self.obsoper(
        self.controlvect, self.obsvect,
        "tl",
        datei=self.datei, datef=self.datef, workdir=self.workdir,
        reload_results=reload_results,
        run_id=self._step,
    )
    self.obsvect = obsvect
    return np.asarray(obsvect.dy, dtype=np.float64)





def adjoint(self, _x: np.ndarray, dy: np.ndarray, reload_results=False) -> np.ndarray:
    if not self.ref_fwd_dir:
        raise CifError(
            "CIFObsOperator.adjoint() called before forward(). "
            "Run forward() at least once first."
        )
    obsvect = self.obsvect
    obsvect.dy = dy
    result_cv = self.obsoper(
        self.controlvect, obsvect,
        "adj",
        datei=self.datei, datef=self.datef, workdir=self.workdir,
        reload_results=reload_results,
        run_id=self._step,   # same id as the paired forward
    )
    return np.asarray(result_cv.dx, dtype=np.float64)