Source code for blissoda.id15b.eiger2crysalis

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

from ..ewoks_utils import submit
from ..processor import BlissScanType
from ..scxrd.eiger2crysalis import DEFAULT_SCAN_PARS
from ..scxrd.eiger2crysalis import DEFAULT_USER_PARS
from ..scxrd.eiger2crysalis import Eiger2CrysalisWithCbfProcessor




class Id15bEiger2Crysalis(Eiger2CrysalisWithCbfProcessor):
    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("trigger_at", "END")
        defaults.setdefault("lima_name", "eiger")
        defaults.setdefault("user_parameters", DEFAULT_USER_PARS.copy())
        defaults.setdefault("scan_parameters", DEFAULT_SCAN_PARS.copy())
        defaults.setdefault("crysalis_ini", "")

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



    def run_conversion(self, scan: BlissScanType) -> None:
        """Executes on given scan"""
        if "fscan" in scan.scan_info["type"] and self.get_lima_name(scan):
            for output_file_format in self.get_output_file_formats():
                kwargs = self.get_submit_arguments(scan, output_file_format)
                kwargs["convert_destination"] = self.workflow_destination(
                    scan, output_file_format
                )
                submit(
                    args=(self.get_workflow(output_file_format),),
                    kwargs=kwargs,
                    queue=self.queue,
                )