Source code for blissoda.demo.processors.xrpd_id31

from __future__ import annotations

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

from ...bliss_globals import setup_globals
from ...id31.xrpd_processor import Id31XrpdProcessor
from .utils.id31 import ensure_difflab6_id31_flats
from .utils.pyfai_calib import ensure_pyfai_config

_logger = logging.getLogger(__name__)




class DemoId31XrpdProcessor(Id31XrpdProcessor):
    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)

        if self._HAS_BLISS:
            newflat, oldflat = ensure_difflab6_id31_flats()
            defaults.setdefault("pyfai_config", ensure_pyfai_config("difflab6"))

            defaults.setdefault(
                "second_detector_pyfai_config", ensure_pyfai_config("diffcam")
            )

        defaults.setdefault("queue", "celery")
        defaults.setdefault("lima_names", ["difflab6"])

        super().__init__(config=config, defaults=defaults)

        if self._HAS_BLISS:
            self.newflat = newflat
            self.oldflat = oldflat



    def pct(self, *args, **kw):
        s = setup_globals.ct(*args, **kw)
        self.on_new_scan(s)
        return s


    def _expected_data_shapes_1d(self, lima_name: str, npoints: int):
        return {
            f"{lima_name}:q": (4096,),
            f"{lima_name}:intensity": (npoints, 4096),
            f"{lima_name}:points": (npoints,),
        }