Source code for darfix.core.data_selection

import os
import urllib.parse
import urllib.request
from typing import Optional
from typing import Union

from esrf_pathlib import ESRFPath
from silx.io.url import DataUrl
from silx.io.utils import get_data

from darfix.core.dataset import Dataset
from darfix.io.progress import OptionalProgressCallback




def load_process_data(
    detector_url: Union[str, DataUrl],
    root_dir: Optional[str] = None,
    dark_detector_url: Optional[Union[str, DataUrl]] = None,
    title: str = "",
    metadata_url=None,
    progress_callback: OptionalProgressCallback = None,
):
    """
    Loads data from `detector_url`.

    :param detector_url: detector_url to be loaded.
    :param metadata_url: path to the scan metadata for HDF5 containing positioner information in order to load metadata for non-edf files
    """
    root_dir_specified = bool(root_dir)

    if not root_dir_specified:
        root_dir = _get_root_dir(detector_url)

    dark_images = None
    if dark_detector_url:
        dark_images = get_data(dark_detector_url)

    dataset = Dataset(
        _dir=root_dir,
        detector_url=detector_url,
        title=title,
        metadata_url=metadata_url,
        progress_callback=progress_callback,
        dark_images=dark_images,
    )
    assert dataset.data is not None and dataset.data.size > 0, "No data was loaded!"

    return dataset



def _get_root_dir(filename: str) -> str:
    url = urllib.parse.urlparse(filename, scheme="file")
    return os.path.dirname(urllib.request.url2pathname(url.path))


def get_default_output_directory(raw_data_file: str) -> str:
    esrf_raw_data_file = ESRFPath(raw_data_file)
    try:
        return str(esrf_raw_data_file.processed_dataset_path)
    except AttributeError:
        # Not an ESRF path : Default directory is cwd.
        return os.getcwd()