Source code for pycif.utils.classes.obsoperators

import datetime
from types import MethodType
import numpy as np

from .baseclass import Plugin
from ..check.errclass import PluginError




class ObsOperator(Plugin):
    """Plugin type for observation operators.

    Maps between control/model space and observation space in both forward
    and adjoint modes. Wraps the full chain from model run to simulated
    observation values.

    Concrete implementations live in ``pycif/plugins/obsoperators/``.
    """



    def obsoper(
        self,
        inputs,
        mode,
        run_id=0,
        datei=datetime.datetime(1979, 1, 1),
        datef=datetime.datetime(2100, 1, 1),
        workdir="./",
        **kwargs
    ):
        """The observation operator.
        This function maps information from the control space to the observation
        space and conversely depending on the running mode.

        Args:
            self (ObsOperator): the ObsOperator plugin
            inputs (Controlvect or Obsvect): the inputs of the fwd or adj mode
            mode (str): the running mode
            run_id (int): the ID of the current run (determines the
            sub-directory name
            datei (datetime.datetime): beginning of the simulation window
            datef (datetime.datetime): end of the simulation window
            workdir (str): path to the parent directory
            """
        return




    @classmethod
    def register_plugin(cls, name, version, module, subtype="", **kwargs):
        """Register a module for a plugin and version with possibly options

        Args:
            name (str):  name of the plugin
            version (str):  version of the plugin
            module (types.ModuleType): module defining the interface
                between pyCIF and the plugin
            plugin_type (str): type of plugin
            **kwargs (dictionary): default options for module

        """

        super(ObsOperator, cls).register_plugin(
            name, version, module, plugin_type="obsoperator", subtype=subtype
        )




    def initiate_template(self):
        """Initialise the ObsOperator plugin template.

        Loads the registered observation-operator module and attaches
        ``obsoper`` as a bound method on this instance.
        """
        default_functions = [
            "obsoper", "forward", "adjoint", "tangent_linear"
        ]
        super(ObsOperator, self).initiate_template(
            plg_type="obsoperator",
            default_functions={k: True for k in default_functions}
        )




    def forward(self, x: np.ndarray) -> np.ndarray:
        """Default empty run method"""

        raise PluginError("This is the default empty run method")




    def tangent_linear(self, x: np.ndarray) -> np.ndarray:
        """Default empty run method"""

        raise PluginError("This is the default empty run method")




    def adjoint(self, x: np.ndarray) -> np.ndarray:
        """Default empty run method"""

        raise PluginError("This is the default empty run method")