Source code for blissoda.demo.processors.fluo

import os
import shutil
from typing import Any
from typing import Dict
from typing import Optional
from typing import Tuple

from ...bliss_globals import current_session
from ...fluo.processor import FluoProcessor
from ...processor import BlissScanType
from ...resources import resource_filename
from ...utils import directories




class DemoFluoProcessor(FluoProcessor):

    def __init__(
        self,
        config: Optional[Dict[str, Any]] = None,
        defaults: Optional[Dict[str, Any]] = None,
    ) -> None:
        if defaults is None:
            defaults = dict()
        defaults.setdefault("instrument_name", "demo")
        # TODO: blissdemo worker cannot handle sub-processes so fit one detector for now
        defaults.setdefault("detectors", ["mca1_det2"])
        super().__init__(config=config, defaults=defaults)
        self._set_demo_pymca_configs()
        self._last_future = None

    def _on_new_scan(self, scan: BlissScanType) -> Tuple[Optional[dict], Optional[Any]]:
        metadata, future = super()._on_new_scan(scan)
        self._last_future = future
        return metadata, future

    def _is_xrfmap(self, scan: BlissScanType) -> bool:
        if not scan.scan_info.get("save"):
            return False
        if scan.scan_info["type"] != "amesh":
            return False
        return True

    def _is_fluoxas(self, scan: BlissScanType) -> bool:
        return False

    def _is_mosaic_xrfmap(self, scan: BlissScanType) -> bool:
        return False

    def _set_demo_pymca_configs(self):
        root_dir = self._get_demo_config_dir(current_session.scan_saving.filename)
        dst_cfgfile = os.path.join(root_dir, "pymca.cfg")
        if os.path.exists(dst_cfgfile):
            return

        src_cfgfile = resource_filename("demo", "pymca.cfg")
        os.makedirs(os.path.dirname(dst_cfgfile), exist_ok=True)
        shutil.copyfile(src_cfgfile, dst_cfgfile)

        # Fit each detector separately
        self.pymca_configs = [dst_cfgfile] * len(self.xrf_names)

    def _get_demo_config_dir(self, dataset_filename: str) -> str:
        root_dir = self._get_demo_result_dir(dataset_filename)
        return os.path.join(root_dir, "config")

    def _get_demo_result_dir(self, dataset_filename: str) -> str:
        root_dir = directories.get_processed_dir(dataset_filename)
        return os.path.join(root_dir, "demo", "fluo")