Source code for blissoda.automation

import logging
from typing import Any
from typing import Dict
from typing import Optional

from pydantic import BaseModel
from pydantic import Field

from .persistent.parameters import WithPersistentParameters
from .utils.classes import NoMethodAssignment

logger = logging.getLogger(__name__)




class BlissAutomationConfig(BaseModel, extra="forbid"):
    """Pydantic model for parsing Beacon configuration of a Bliss object for automation.

    For example:

    .. code-block:: yaml

        - name: my_processor
          plugin: generic
          class: MyProcessor
          package: blissoda.mycategory.my_processor
          default_parameters:
            param1: value1
            param2: value2
          shared: false
          singleton: true

    In this example the Redis key is `blissoda:demo_session:MyProcessor`.
    """

    name: Optional[str] = Field(
        None, description="Name of the instance (used if singleton is True)."
    )
    plugin: Optional[str] = Field(None, description="Plugin type, typically 'generic'.")
    class_: Optional[str] = Field(
        None, alias="class", description="Class name of the object to instantiate."
    )
    package: Optional[str] = Field(
        None, description="Python package where the class is located."
    )
    default_parameters: Dict[str, Any] = Field(
        default_factory=dict,
        description="Initial values for parameters stored in Redis.",
    )
    shared: bool = Field(
        False, description="If True, parameters are shared between Bliss sessions."
    )
    singleton: bool = Field(
        True,
        description="If True, parameters are a singleton for the Bliss session (``shared=False``)"
        "or globally (``shared=True``).",
    )





class BlissAutomationObject(WithPersistentParameters, NoMethodAssignment):
    """A Bliss object for automation (data processing and/or acquisition).

    It has persistent parameters in Redis and configuration in Beacon.

    The Beacon configuration is expected to use the
    `generic plugin <https://bliss.gitlab-pages.esrf.fr/bliss/master/config_plugins.html#plugin-generic>`_.
    """

    CONFIG_MODEL_CLASS = BlissAutomationConfig

    def __init__(
        self,
        config: Optional[Dict[str, Any]] = None,
        defaults: Optional[Dict[str, Any]] = None,
    ) -> None:
        """
        :param config: Beacon configuration.
        :param defaults: Initial values for parameters stored in Redis;
                         merged with priority over ``config["default_parameters"]``.
        """
        if not issubclass(self.CONFIG_MODEL_CLASS, BlissAutomationConfig):
            raise TypeError(
                "CONFIG_MODEL_CLASS must be a subtype of BlissAutomationConfig"
            )

        config_dict = config or {}
        config_model = self.CONFIG_MODEL_CLASS(**config_dict)

        shared = config_model.shared
        name = None if config_model.singleton else config_model.name

        defaults = defaults or {}
        defaults = {**defaults, **config_model.default_parameters}

        super().__init__(shared=shared, name=name, defaults=defaults)

        self._config = config_model