Source code for blissoda.demo.processors.bm02

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

from ...bm02.xrpd_processor import Bm02XrpdProcessor
from .utils.pyfai_calib import ensure_pyfai_config




class DemoBm02XrpdProcessor(Bm02XrpdProcessor):
    def __init__(
        self,
        config: Optional[Dict[str, Any]] = None,
        defaults: Optional[Dict[str, Any]] = None,
        **deprecated_defaults: Dict[str, Any],
    ) -> None:
        defaults = self._merge_defaults(deprecated_defaults, defaults)

        defaults.setdefault("queue", "celery")
        defaults.setdefault(
            "integration_options",
            {
                "difflab6": {
                    "method": "no_csr_cython",
                    "nbpt_rad": 4096,
                    "unit": "q_A^-1",
                }
            },
        )
        defaults.setdefault(
            "empty_cell_subtraction_options",
            {
                "difflab6": {
                    "enabled": False,
                    "cell_pattern_url": None,
                    "empty_cell_factor": 1,
                }
            },
        )

        super().__init__(config=config, defaults=defaults)
        self._last_future = None

        if self._HAS_BLISS:
            self.config_filename = {"difflab6": ensure_pyfai_config("difflab6")}

    def _on_new_scan(self, scan):
        metadata, future = super()._on_new_scan(scan)
        self._last_future = future

        return metadata, future