Source code for blissoda.demo.processors.id09

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

from ...bliss_globals import setup_globals
from ...id09.txs_processor import TxsProcessor




class DemoTxsProcessor(TxsProcessor):
    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("detector", "difflab6")
        defaults.setdefault("pixel", (10e-3, 10e-3))
        defaults.setdefault("energy", 10e3)

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



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


    @property
    def lima_names(self) -> List[str]:
        return [self.detector]

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