Source code for ewoksfluo.tasks.input.pick_utils

from typing import List
from typing import Optional
from typing import Tuple

from ..nexus_utils import wait_scan_finished




def pick_scans(
    filenames: List[str],
    scan_ranges: List[Tuple[int, int]],
    exclude_scans: List[List[int]],
    retry_timeout: Optional[float] = None,
    retry_period: float = 0.5,
) -> List[str]:
    """
    Return list of scan URIs based on files names with associated scan ranges
    and individual scans to be excluded.
    """
    do_wait = retry_timeout is not None and retry_timeout >= 0

    bliss_scan_uris = []
    for filename, scan_range, excluded_scans in zip(
        filenames, scan_ranges, exclude_scans
    ):
        scan_min, scan_max = scan_range
        excluded_scans = excluded_scans if excluded_scans else []
        for scan_number in range(scan_min, scan_max + 1):
            if scan_number in excluded_scans:
                continue

            bliss_scan_uri = f"{filename}::/{scan_number}.1"

            if do_wait:
                wait_scan_finished(
                    bliss_scan_uri,
                    retry_timeout=retry_timeout,
                    retry_period=retry_period,
                )
            bliss_scan_uris.append(bliss_scan_uri)
    return bliss_scan_uris