import getpass
import os
import pathlib
import shutil
import tempfile
from unittest.mock import patch
import pytest
from ewoksmx.tasks.Grenades_fastproc_pipeline import Grenades_fastproc_pipeline
from ..data import resource_path_context
[docs]
@pytest.fixture()
def xds_inp_path():
with resource_path_context("XDS.INP") as xds_inp_path:
yield xds_inp_path
[docs]
@pytest.fixture
def ensure_get_spg_and_cell_params():
with patch.object(
Grenades_fastproc_pipeline, "_get_spg_and_cell_params"
) as mock_method:
mock_method.return_value = (1, [77.5, 155.8, 155.9, 60.1, 89.9, 89.8])
yield
[docs]
def test_grenades_fastproc_prepare(
tmp_path: pathlib.Path, xds_inp_path: pathlib.Path, ensure_get_spg_and_cell_params
):
metadata = {
"beamline": "id23eh2",
"proposal": "20231212",
"MX_dataCollectionId": 123456,
"reprocess_path": str(tmp_path),
"xds_inp_path": str(xds_inp_path),
"MX_directory": str(tmp_path / "mx_directory"),
"forced_spacegroup": "P1",
"exclude_range": [[10, 20], [30, 40]],
"no_pipelines": 1,
}
pipeline = Grenades_fastproc_pipeline(
inputs={"metadata": metadata, "grenades_fastproc": True}
)
pipeline.execute()
assert pipeline.xds_inp_path.exists()
_validate_xds_inp(pipeline.xds_inp_path, metadata)
for dcolid_path in pipeline.dcloid_file_paths:
assert dcolid_path.exists()
_validate_dcolid(dcolid_path, metadata)
assert pipeline.script_file_path.exists()
expected = {
"slurm_params": {
"script_file_path": str(
tmp_path / "grenades_fastproc" / "nobackup" / "grenades_fastproc.sh"
),
"queue": "mx",
"mem": 16000,
"nodes": 1,
"core": 20,
"time": "2:00:00",
"icat_dir": str(tmp_path / "grenades_fastproc"),
"icat_callback_url": None,
"no_pipelines": 1,
"pipeline_name": "grenades_fastproc",
"error_message": None,
},
"pipeline_name": "grenades_fastproc",
}
assert pipeline.get_output_values() == expected
def _validate_xds_inp(xds_inp_path: pathlib.Path, metadata: dict):
with open(xds_inp_path) as f:
xds_inp = f.read()
for begin, end in metadata["exclude_range"]:
assert f"EXCLUDE_DATA_RANGE= {begin} {end}" in xds_inp
assert "SPACE_GROUP_NUMBER= 1" in xds_inp
assert "UNIT_CELL_CONSTANTS= 77.5 155.8 155.9 60.1 89.9 89.8" in xds_inp
def _validate_dcolid(dcolid_path: pathlib.Path, metadata: dict):
with open(dcolid_path) as f:
dcolid = f.read()
assert dcolid == f"datacollectionID:{metadata['MX_dataCollectionId']}\n"
XDS_INP = """
! ispyb web service request time: 0.16s
! generated Mon, 10 Feb 2025 19:55:56
! params: raw_data=True, basedir=../links, request_basedir=../links
JOB= ALL !XYCORR INIT COLSPOT IDXREF DEFPIX XPLAN INTEGRATE CORRECT
!JOB= DEFPIX XPLAN INTEGRATE CORRECT
DATA_RANGE= 1 900
SPOT_RANGE= 1 50
SPOT_RANGE= 426 475
SPOT_RANGE= 851 900
BACKGROUND_RANGE= 1 10
!EXCLUSION OF HORIZONTAL DEAD AREAS OF THE EIGER 9M DETECTOR
UNTRUSTED_RECTANGLE= 0 3109 512 551
UNTRUSTED_RECTANGLE= 0 3109 1062 1101
UNTRUSTED_RECTANGLE= 0 3109 1612 1651
UNTRUSTED_RECTANGLE= 0 3109 2162 2201
UNTRUSTED_RECTANGLE= 0 3109 2712 2751
UNTRUSTED_RECTANGLE= 1452 1756 1579 1666
UNTRUSTED_RECTANGLE= 1753 2686 1590 1672
UNTRUSTED_RECTANGLE= 2684 3101 1595 1670
UNTRUSTED_RECTANGLE= 3101 3104 1672 1671
!EXCLUSION OF VERTICAL DEAD PIXELS OF THE EIGER 9M DETECTOR
UNTRUSTED_RECTANGLE= 1028 1041 0 3262
UNTRUSTED_RECTANGLE= 2068 2081 0 3262
!Exclusion of beamstop
! UNTRUSTED_RECTANGLE= 1550 1590 1665 3262
!EXCLUSION OF DEAD AREAS OF THE EIGER 9M DETECTOR
! UNTRUSTED_RECTANGLE= 0 1035 1100 1355
TRUSTED_REGION=0.0 1.2 !Relative radii limiting trusted detector region
SECONDS=300
MINIMUM_NUMBER_OF_PIXELS_IN_A_SPOT= 3
!STRONG_PIXEL= 3.0
OSCILLATION_RANGE= 0.2000
STARTING_ANGLE= 89.999
STARTING_FRAME= 1
X-RAY_WAVELENGTH= 0.87313
LIB= dectris-neggia.so
NAME_TEMPLATE_OF_DATA_FRAMES= ../links/run_10_04_datacollection/Sample-8-1-05_10_4_1_??????.h5 ! H5
!STARTING_ANGLES_OF_SPINDLE_ROTATION= 0 180 10
!TOTAL_SPINDLE_ROTATION_RANGES= 60 180 10
DETECTOR_DISTANCE= 222.80
DETECTOR=EIGER MINIMUM_VALID_PIXEL_VALUE=0 OVERLOAD=256615
SENSOR_THICKNESS=0.45
ORGX= 1584.59 ORGY= 1613.89
NX=3108 NY=3262
QX= 0.0750 QY= 0.0750
VALUE_RANGE_FOR_TRUSTED_DETECTOR_PIXELS= 6000 30000
DIRECTION_OF_DETECTOR_X-AXIS= -1.0 0.0 0.0
DIRECTION_OF_DETECTOR_Y-AXIS= 0.0 -1.0 0.0
ROTATION_AXIS= 0.0 1.0 0.0
INCIDENT_BEAM_DIRECTION= 0.0 0.0 1.0
FRACTION_OF_POLARIZATION= 0.99
POLARIZATION_PLANE_NORMAL= 0.0 1.0 0.0
SPACE_GROUP_NUMBER= 0
UNIT_CELL_CONSTANTS= 0 0 0 0 0 0
INCLUDE_RESOLUTION_RANGE= 50.0 0.0
!RESOLUTION_SHELLS= 15.0 8.0 4.0 2.8 2.4
!FRIEDEL'S_LAW= FALSE !default is TRUE
!STRICT_ABSORPTION_CORRECTION=TRUE
REFINE(IDXREF)=CELL BEAM ORIENTATION AXIS
REFINE(INTEGRATE)= POSITION BEAM ORIENTATION ! AXIS CELL . If 1.5A or higher it is ok to refine CELL (unless electron diffraction)
REFINE(CORRECT)= CELL BEAM ORIENTATION AXIS POSITION ! Default is: refine everything
NUMBER_OF_PROFILE_GRID_POINTS_ALONG_ALPHA/BETA=13 ! Default is 9 - Increasing may improve data
NUMBER_OF_PROFILE_GRID_POINTS_ALONG_GAMMA=13 ! accuracy, particularly if finely-sliced on phi,
!== Default value recommended
!DELPHI= %.3f
MAXIMUM_NUMBER_OF_PROCESSORS= 16
MAXIMUM_NUMBER_OF_JOBS= 1
SEPMIN= 4
CLUSTER_RADIUS= 2
""" # noqa W293
[docs]
def is_sbatch_available() -> bool:
"""Check if the sbatch command is available on the system."""
return shutil.which("sbatch") is not None
[docs]
@pytest.mark.skipif(
not os.path.exists("/tmp_14_days"),
reason="/tmp_14_days does not exist",
)
@pytest.mark.skipif(
not os.path.exists(
"/data/scisoft/pxsoft/data/WORKFLOW_TEST_DATA/id23eh2/20250121/RAW_DATA/Sample-8-1-05/run_10_MXPressF/run_10_04_datacollection"
),
reason="Data directory does not exist",
)
@pytest.mark.skipif(
not shutil.which(Grenades_fastproc_pipeline.CALC_CELL_CMD),
reason="Cell determination program not exist",
)
@pytest.mark.skipif(not is_sbatch_available(), reason="requires the sbatch command")
def test_get_spg_and_cell_params():
# tmp_path must be on /tmp_14_days, otherwise test will not work
# as the job is submitted to slurm
user_dir = pathlib.Path("/tmp_14_days") / getpass.getuser()
if not user_dir.exists():
user_dir.mkdir()
# Create temp test dir
tmp_path = pathlib.Path(
tempfile.mkdtemp(dir=user_dir, prefix="test_get_spg_and_cell_params_")
)
# Create XDS.INP file
xds_inp_path = tmp_path / "XDS.INP"
with open(xds_inp_path, "w") as f:
f.write(XDS_INP)
# Test metadata
metadata = {
"MX_dataCollectionId": 123456,
"MX_directory": "/data/scisoft/pxsoft/data/WORKFLOW_TEST_DATA/id23eh2/20250121/RAW_DATA/Sample-8-1-05/run_10_MXPressF/run_10_04_datacollection",
"forced_spacegroup": "P1",
"xds_inp_path": str(xds_inp_path),
"reprocess_path": str(tmp_path),
}
# Run test
grenades_fastproc_pipeline = Grenades_fastproc_pipeline(
inputs={"metadata": metadata, "grenades_fastproc": True}
)
working_dir = grenades_fastproc_pipeline.grenades_working_dir
working_dir.mkdir(parents=True)
grenades_fastproc_pipeline._create_xds_inp()
grenades_fastproc_pipeline._modify_xds_inp()
# Check results
assert grenades_fastproc_pipeline.space_group_number == 1
# Allow a delta of 0.2 for the cell parameters - depends on the XDS version
delta = 0.2
cell_params = grenades_fastproc_pipeline.cell_params
reference_cell_params = [67.4, 67.5, 67.5, 109.5, 109.4, 109.4]
for cell_param, reference_cell_param in zip(cell_params, reference_cell_params):
assert abs(cell_param - reference_cell_param) <= delta
# Cleanup
shutil.rmtree(tmp_path)