Source code for blissoda.demo.tests.itest_bm02

from pathlib import Path

from blissoda.bm02.xrpd_processor import scan_processed_directory

from ...bliss_globals import setup_globals
from ...bm02.utils import export_filename_prefix
from ...demo.processors.bm02 import DemoBm02XrpdProcessor
from ...demo.tests.utils.xrpd import _scan_context
from ...processor import BlissScanType
from .. import testing




@testing.integration_fixture
def bm02_xrpd_processor(gc_collect):
    bm02_processor = DemoBm02XrpdProcessor()
    bm02_processor.enable(setup_globals.difflab6)
    yield bm02_processor
    bm02_processor.disable_empty_cell_subtraction(setup_globals.difflab6)
    bm02_processor.disable()



def _scan(expo=0.2, npoints=10, counters=tuple()) -> BlissScanType:
    with _scan_context() as scans:
        scan = setup_globals.loopscan(npoints, expo, setup_globals.difflab6, *counters)
        scans.append(scan)
        return scan




@testing.integration_test
def test_bm02_1d(bm02_xrpd_processor, expo=0.2, npoints=10, ascii=False):
    bm02_xrpd_processor.ascii_export_enabled = ascii
    bm02_xrpd_processor.integration_options["difflab6"].pop("nbpt_azim", None)
    _scan(expo, npoints)
    testing.wait_workflows(raise_on_error=True)





@testing.integration_test
def test_bm02_2d(bm02_xrpd_processor, expo=0.2, npoints=10, ascii=False):
    bm02_xrpd_processor.integration_options["difflab6"]["nbpt_azim"] = 100
    bm02_xrpd_processor.ascii_export_enabled = ascii
    _scan(expo, npoints)
    testing.wait_workflows(raise_on_error=True)





@testing.integration_test
def test_bm02_1d_with_cell_subtraction(
    bm02_xrpd_processor, expo=0.2, npoints=10, ascii=False
):
    bm02_xrpd_processor.ascii_export_enabled = ascii
    bm02_xrpd_processor.integration_options["difflab6"].pop("nbpt_azim", None)

    # Do a first scan to be subtracted
    scan = _scan(expo, npoints)
    testing.wait_workflows(raise_on_error=True)
    scan_metadata = scan.scan_saving

    bm02_xrpd_processor.enable_empty_cell_subtraction(setup_globals.difflab6)
    bm02_xrpd_processor.set_cell_pattern_url(
        sample=scan_metadata.sample_name,
        dataset=scan_metadata.dataset,
        scan_number=scan.scan_info["scan_nb"],
        detector="difflab6",
    )

    _scan(expo, npoints)
    testing.wait_workflows(raise_on_error=True)





@testing.integration_test
def test_bm02_1d_with_energy(bm02_xrpd_processor, expo=0.2, npoints=10, energy=13):
    bm02_xrpd_processor.energy = energy
    bm02_xrpd_processor.integration_options["difflab6"].pop("nbpt_azim", None)

    _scan(expo, npoints)
    testing.wait_workflows(raise_on_error=True)





@testing.integration_test
def test_bm02_1d_with_exported_counters(
    bm02_xrpd_processor: DemoBm02XrpdProcessor, expo=0.2, npoints=10
):
    bm02_xrpd_processor.enable_ascii_export()
    bm02_xrpd_processor.exported_counters = ["diode1"]

    try:
        scan = _scan(expo, npoints, counters=(setup_globals.diode1,))
        _assert_counter_in_file(
            bm02_xrpd_processor,
            scan=scan,
            lima_name="difflab6",
        )
    finally:
        bm02_xrpd_processor.disable_ascii_export()



@testing.demo_assert("BM02: check that the counter are in of the ASCII files")
def _assert_counter_in_file(
    bm02_xrpd_processor: DemoBm02XrpdProcessor,
    scan,
    lima_name: str,
):
    future = bm02_xrpd_processor._last_future
    assert future is not None
    future.result(timeout=10)
    output_filenames = [
        Path(scan_processed_directory(scan))
        / "export"
        / f"{export_filename_prefix(scan, lima_name)}_f{i:04d}.dat"
        for i in range(10)
    ]
    for i, filename in enumerate(output_filenames):
        with open(filename, "r") as file:
            for line in file:
                if "# diode1 =" in line:
                    break
            else:
                raise ValueError(f"diode1 not in {filename}")