Source code for blissoda.id11.eiger2crysalis

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

from ..ewoks_utils import convert_graph
from ..ewoks_utils import submit
from ..persistent.parameters import ParameterInfo
from ..processor import BlissScanType
from ..scxrd.eiger2crysalis import DEFAULT_SCAN_PARS
from ..scxrd.eiger2crysalis import DEFAULT_USER_PARS
from ..scxrd.eiger2crysalis import PAR_METADATA_KEYS
from ..scxrd.eiger2crysalis import Eiger2CrysalisProcessor

_DEFAULT_USER_PARS = {
    **DEFAULT_USER_PARS,
    "flip_lr": True,
}




class Id11Eiger2Crysalis(
    Eiger2CrysalisProcessor,
    parameters=[
        ParameterInfo(
            "mode",
            category="Mode",
            doc="Choose 'Execute' to run the conversion or 'Json' just to generate the .json file without executing",
        ),
        ParameterInfo(
            "overwrite_par_file_metadata",
            category="Eiger2Crysalis",
            doc="If True, include individual user and scan parameters (distance, beam, etc.) will be overwritten in the par file",
        ),
    ],
):
    default_user_parameters = _DEFAULT_USER_PARS

    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("user_parameters", _DEFAULT_USER_PARS.copy())
        defaults.setdefault("scan_parameters", DEFAULT_SCAN_PARS.copy())
        defaults.setdefault("crysalis_ini", "")
        defaults.setdefault("overwrite_par_file_metadata", False)

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



    def get_omega(self, scan: BlissScanType) -> str:
        """
        Provide the omega parameter as a formatted string based on scan info.

        The scan is assumed to be centered around zero.
        The omega string is computed after ensuring:
        - The step_size is negative (as required by CrysAlis).
        """
        params = scan.scan_info["instrument"]["fscan_parameters"]
        omega_start = params["start_pos"]
        step_size = params["step_size"]
        nsteps = params["npoints"]
        if step_size >= 0:
            print(
                "WARNING! CrysAlis cannot handle positive steps!! Changing to negative"
            )
            step_size = -step_size

        end = nsteps * step_size + omega_start

        npts = (end - omega_start) / step_size
        center = step_size * npts / 2 + omega_start
        omega_start_symetric = omega_start - center
        omega_end_symetric = omega_start + center
        self._omega_start_symetric = omega_start_symetric
        self._omega_end_symetric = -omega_end_symetric
        self._step_size = step_size
        self._omega_start = omega_start
        self._omega_end = end
        return f"-{omega_start}-index*{step_size}"




    def get_run_parameters(self, scan: BlissScanType) -> Dict[str, Any]:
        parameters = super().get_run_parameters(scan)
        parameters.update(
            {
                "omega_start": self._omega_start,
                "omega_end": self._omega_end,
                "domega": self._step_size,
            }
        )
        return parameters




    def collect_par_file_metadata_overwrites(
        self, parameters: Dict[str, Any]
    ) -> Dict[str, Any]:
        """
        Collect par-file metadata keys if overwrite_par_file_metadata is enabled.
        """
        overwrite = {}
        for key in PAR_METADATA_KEYS:
            if key in parameters:
                overwrite[key] = parameters[key]
        return overwrite




    def get_extra_inputs(self, scan: BlissScanType, parameters: Dict[str, Any]) -> list:
        inputs = []
        if self.overwrite_par_file_metadata:
            par_overwrites = self.collect_par_file_metadata_overwrites(parameters)
            for key, val in par_overwrites.items():
                inputs.append(
                    {
                        "id": "5",
                        "task_identifier": "CreateParFiles",
                        "name": key,
                        "value": val,
                    }
                )
        inputs.extend(super().get_extra_inputs(scan, parameters))
        return inputs




    def run_conversion(self, scan: BlissScanType) -> None:
        """Executes on given scan"""
        if "fscan" in scan.scan_info["type"] and self.get_lima_name(scan):
            if self.mode == "Execute":
                kwargs = self.get_submit_arguments(scan)
                kwargs["convert_destination"] = self.workflow_destination(scan)
                submit(args=(self.workflow,), kwargs=kwargs, queue=self.queue)
            elif self.mode == "Json":
                kwargs = {"inputs": self.get_inputs(scan)}
                kwargs["destination"] = self.workflow_destination(scan)
                convert_graph(args=(self.workflow,), kwargs=kwargs)
            else:
                raise TypeError(f"{self.mode} is not Execute or Json")