Quick qfas xanes

[stan-dot]

move of the Bragg syncrhonized with IDGAP

detectors: i0, it, fluorescence (xpress3)

[stan-dot]

draft of a sample implementation:

# Run the quick QFAS XANES scan
RE(quick_qfas_xanes_no_panda(angles_list, gaps_list, scan_count_time))


@inject
def quick_qfas_xanes(
    angles: List[float],
    gaps: List[float],
    exposure: float,
    panda: HDFPanda = inject(panda),
    detectors: List[StandardDetector] = [i0, IT, fluorescence],
    baseline: List[Device] = [monochromator, panda],
    metadata: Optional[Dict[str, Any]] = None,
) -> MsgGenerator:
    assert len(angles) == len(gaps), "Angles and gaps lists must be the same length"

    stream_name = "main"
    flyer = HardwareTriggeredFlyable(StaticSeqTableTriggerLogic(panda.seq[1]))
    devices = [flyer] + detectors + [panda] + baseline

    # Collect metadata
    plan_args = {
        "angles": angles,
        "gaps": gaps,
        "exposure": exposure,
        "panda": repr(panda),
        "detectors": [repr(device) for device in detectors],
        "baseline": [repr(device) for device in baseline],
    }

    _md = {
        "detectors": [device.name for device in detectors],
        "plan_args": plan_args,
        "hints": {},
    }
    _md.update(metadata or {})

    @baseline_decorator(baseline)
    @stage_decorator(devices)
    @run_decorator(md=_md)
    def inner_plan():
        yield from prepare_seq_table_flyer_and_det(
            flyer=flyer,
            detectors=detectors,
            angles=angles,
            gaps=gaps,
            exposure=exposure,
        )
        yield from fly_and_collect(
            stream_name=stream_name,
            detectors=detectors,
            flyer=flyer,
        )

    yield from inner_plan()


def prepare_seq_table_flyer_and_det(
    flyer: HardwareTriggeredFlyable,
    detectors: List[StandardDetector],
    angles: List[float],
    gaps: List[float],
    exposure: float,
) -> MsgGenerator:
    table_info = create_xanes_seq_table(angles, gaps, exposure)
    for det in detectors:
        yield from bps.prepare(det, table_info, wait=False, group="prep")
    yield from bps.prepare(flyer, table_info, wait=False, group="prep")
    yield from bps.wait(group="prep")


def create_xanes_seq_table(
    angles: List[float],
    gaps: List[float],
    exposure: float,
) -> StaticSeqTableTriggerLogic:
    rows = [
        {
            "bragg": angle,
            "idgap": gap,
            "exposure": exposure,
            "detectors": {"i0": True, "IT": True, "fluorescence": True},
        }
        for angle, gap in zip(angles, gaps)
    ]

    seq_table = StaticSeqTableTriggerLogic(rows)
    return seq_table


def fly_and_collect(
    stream_name: str,
    detectors: List[StandardDetector],
    flyer: HardwareTriggeredFlyable,
) -> MsgGenerator:
    yield from bps.fly(flyer)
    yield from bps.collect(detectors, stream_name)