Source code for blissoda.tests.test_scxrd_eiger2crysalis

from types import SimpleNamespace
from unittest.mock import patch

from .. import resources
from ..scxrd.eiger2crysalis import Eiger2CrysalisLima2Processor
from ..scxrd.eiger2crysalis import Eiger2CrysalisProcessor
from ..scxrd.eiger2crysalis import Eiger2CrysalisWithCbfProcessor


def _scan(tmp_path):
    filename = tmp_path / "RAW_DATA" / "sample" / "sample_0001" / "sample_0001.h5"
    scan_info = {
        "filename": str(filename),
        "scan_nb": 10,
        "npoints": 5,
        "type": "fscan",
        "channels": {"eiger:image": {}},
        "dataset_metadata_snapshot": {"Sample_name": "sample"},
        "instrument": {
            "fscan_parameters": {
                "start_pos": -5.0,
                "step_size": 0.5,
                "acq_time": 0.1,
            }
        },
    }
    scan_saving = SimpleNamespace(
        filename=scan_info["filename"],
        images_path=str(tmp_path / "images" / "{scan_number}_{img_acq_device}_"),
        sample_name="sample",
    )
    return SimpleNamespace(
        scan_number=scan_info["scan_nb"],
        scan_info=scan_info,
        scan_saving=scan_saving,
    )


def _inputs_by_name(inputs):
    return {item["name"]: item["value"] for item in inputs}




def test_default_workflow_resources(mock_persistent):
    processor = Eiger2CrysalisProcessor()
    processor_with_cbf = Eiger2CrysalisWithCbfProcessor()
    lima2_processor = Eiger2CrysalisLima2Processor()

    assert processor.workflow == resources.resource_filename(
        "scxrd", "eiger2crysalis.json"
    )
    assert processor_with_cbf._workflow == resources.resource_filename(
        "scxrd", "eiger2crysalis.json"
    )
    assert processor_with_cbf._cbf_workflow == resources.resource_filename(
        "scxrd", "eiger2cbf_xds.json"
    )
    assert lima2_processor.workflow == resources.resource_filename(
        "scxrd", "eiger2crysalis_lima2.json"
    )





def test_get_inputs_for_cbf_filters_crysalis_only_parameters(mock_persistent, tmp_path):
    processor = Eiger2CrysalisWithCbfProcessor(defaults={"output_file_format": "cbf"})

    inputs = processor.get_inputs(_scan(tmp_path), output_file_format="cbf")
    cbf_inputs = [item for item in inputs if item["task_identifier"] == "Eiger2CBF"]
    xds_inputs = [item for item in inputs if item["task_identifier"] == "CreateXDSInp"]
    names = _inputs_by_name(cbf_inputs)
    xds_names = _inputs_by_name(xds_inputs)

    assert {item["task_identifier"] for item in inputs} == {
        "CreateXDSInp",
        "Eiger2CBF",
    }
    assert names["output"].endswith(".cbf")
    assert names["omega"] == "-5.0+index*0.5"
    assert "processed_output" not in names
    assert "polarization" not in names
    assert "theta" not in names
    assert "calc_mask" not in names
    assert "exposure_time" not in names
    assert xds_names["output"].endswith(".cbf")
    assert xds_names["detector"] == "EIGER"
    assert xds_names["nx"] == 3108
    assert xds_names["ny"] == 3262
    assert xds_names["qx"] == 0.075
    assert xds_names["qy"] == 0.075
    assert xds_names["beam"] == [1000, 1100]
    assert xds_names["distance"] == 100
    assert xds_names["wavelength"] == 0.1
    assert xds_names["data_range"] == [1, 5]
    assert xds_names["spot_range"] == [1, 5]
    assert xds_names["background_range"] == [1, 5]
    assert xds_names["starting_frame"] == 1
    assert xds_names["starting_angle"] == -5.0
    assert xds_names["oscillation_range"] == 0.5
    assert xds_names["fraction_of_polarization"] == 0.99
    assert xds_names["rotation_axis"] == [0.0, -1.0, 0.0]
    assert xds_names["incident_beam_direction"] == [0.0, 0.0, 1.0]
    assert xds_names["trusted_region"] == [0.0, 1.41]
    assert xds_names["filename"] == "XDS.INP"
    assert xds_names["job"] == "XYCORR INIT COLSPOT IDXREF DEFPIX INTEGRATE CORRECT"
    assert xds_names["sensor_thickness"] == 0.75
    assert xds_names["silicon"] == 18.023
    assert xds_names["untrusted_rectangles"]
    assert "xds_extra" not in xds_names





def test_run_conversion_dispatches_selected_workflows(mock_persistent, tmp_path):
    processor = Eiger2CrysalisWithCbfProcessor(
        defaults={
            "_workflow": "esperanto_workflow.json",
            "_cbf_workflow": "cbf_workflow.json",
            "output_file_format": ["cbf", "esperanto"],
        }
    )

    assert processor.get_output_file_formats() == ["cbf", "esperanto"]

    with patch("blissoda.scxrd.eiger2crysalis.submit") as submit_mock:
        processor.run_conversion(_scan(tmp_path))

    assert [call.kwargs["args"][0] for call in submit_mock.call_args_list] == [
        "cbf_workflow.json",
        "esperanto_workflow.json",
    ]

    cbf_call = submit_mock.call_args_list[0].kwargs
    esperanto_call = submit_mock.call_args_list[1].kwargs

    assert cbf_call["kwargs"]["convert_destination"].endswith("_cbf.json")
    assert cbf_call["kwargs"]["inputs"][0]["task_identifier"] == "Eiger2CBF"
    assert any(
        item["task_identifier"] == "CreateXDSInp"
        for item in cbf_call["kwargs"]["inputs"]
    )

    assert esperanto_call["kwargs"]["convert_destination"].endswith(".json")
    assert any(
        item["task_identifier"] == "Eiger2Crysalis"
        for item in esperanto_call["kwargs"]["inputs"]
    )