1import functools
2import inspect
3import itertools
4import warnings
5
6from .common import BACKEND_ENTRYPOINTS, BackendEntrypoint
7
8try:
9    from importlib.metadata import entry_points
10except ImportError:
11    # if the fallback library is missing, we are doomed.
12    from importlib_metadata import entry_points  # type: ignore[no-redef]
13
14
15STANDARD_BACKENDS_ORDER = ["netcdf4", "h5netcdf", "scipy"]
16
17
18def remove_duplicates(entrypoints):
19    # sort and group entrypoints by name
20    entrypoints = sorted(entrypoints, key=lambda ep: ep.name)
21    entrypoints_grouped = itertools.groupby(entrypoints, key=lambda ep: ep.name)
22    # check if there are multiple entrypoints for the same name
23    unique_entrypoints = []
24    for name, matches in entrypoints_grouped:
25        matches = list(matches)
26        unique_entrypoints.append(matches[0])
27        matches_len = len(matches)
28        if matches_len > 1:
29            selected_module_name = matches[0].module_name
30            all_module_names = [e.module_name for e in matches]
31            warnings.warn(
32                f"Found {matches_len} entrypoints for the engine name {name}:"
33                f"\n {all_module_names}.\n It will be used: {selected_module_name}.",
34                RuntimeWarning,
35            )
36    return unique_entrypoints
37
38
39def detect_parameters(open_dataset):
40    signature = inspect.signature(open_dataset)
41    parameters = signature.parameters
42    parameters_list = []
43    for name, param in parameters.items():
44        if param.kind in (
45            inspect.Parameter.VAR_KEYWORD,
46            inspect.Parameter.VAR_POSITIONAL,
47        ):
48            raise TypeError(
49                f"All the parameters in {open_dataset!r} signature should be explicit. "
50                "*args and **kwargs is not supported"
51            )
52        if name != "self":
53            parameters_list.append(name)
54    return tuple(parameters_list)
55
56
57def backends_dict_from_pkg(entrypoints):
58    backend_entrypoints = {}
59    for entrypoint in entrypoints:
60        name = entrypoint.name
61        try:
62            backend = entrypoint.load()
63            backend_entrypoints[name] = backend
64        except Exception as ex:
65            warnings.warn(f"Engine {name!r} loading failed:\n{ex}", RuntimeWarning)
66    return backend_entrypoints
67
68
69def set_missing_parameters(backend_entrypoints):
70    for name, backend in backend_entrypoints.items():
71        if backend.open_dataset_parameters is None:
72            open_dataset = backend.open_dataset
73            backend.open_dataset_parameters = detect_parameters(open_dataset)
74
75
76def sort_backends(backend_entrypoints):
77    ordered_backends_entrypoints = {}
78    for be_name in STANDARD_BACKENDS_ORDER:
79        if backend_entrypoints.get(be_name, None) is not None:
80            ordered_backends_entrypoints[be_name] = backend_entrypoints.pop(be_name)
81    ordered_backends_entrypoints.update(
82        {name: backend_entrypoints[name] for name in sorted(backend_entrypoints)}
83    )
84    return ordered_backends_entrypoints
85
86
87def build_engines(entrypoints):
88    backend_entrypoints = {}
89    for backend_name, backend in BACKEND_ENTRYPOINTS.items():
90        if backend.available:
91            backend_entrypoints[backend_name] = backend
92    entrypoints = remove_duplicates(entrypoints)
93    external_backend_entrypoints = backends_dict_from_pkg(entrypoints)
94    backend_entrypoints.update(external_backend_entrypoints)
95    backend_entrypoints = sort_backends(backend_entrypoints)
96    set_missing_parameters(backend_entrypoints)
97    return {name: backend() for name, backend in backend_entrypoints.items()}
98
99
100@functools.lru_cache(maxsize=1)
101def list_engines():
102    entrypoints = entry_points().get("xarray.backends", ())
103    return build_engines(entrypoints)
104
105
106def guess_engine(store_spec):
107    engines = list_engines()
108
109    for engine, backend in engines.items():
110        try:
111            if backend.guess_can_open(store_spec):
112                return engine
113        except Exception:
114            warnings.warn(f"{engine!r} fails while guessing", RuntimeWarning)
115
116    compatible_engines = []
117    for engine, backend_cls in BACKEND_ENTRYPOINTS.items():
118        try:
119            backend = backend_cls()
120            if backend.guess_can_open(store_spec):
121                compatible_engines.append(engine)
122        except Exception:
123            warnings.warn(f"{engine!r} fails while guessing", RuntimeWarning)
124
125    installed_engines = [k for k in engines if k != "store"]
126    if not compatible_engines:
127        if installed_engines:
128            error_msg = (
129                "did not find a match in any of xarray's currently installed IO "
130                f"backends {installed_engines}. Consider explicitly selecting one of the "
131                "installed engines via the ``engine`` parameter, or installing "
132                "additional IO dependencies, see:\n"
133                "http://xarray.pydata.org/en/stable/getting-started-guide/installing.html\n"
134                "http://xarray.pydata.org/en/stable/user-guide/io.html"
135            )
136        else:
137            error_msg = (
138                "xarray is unable to open this file because it has no currently "
139                "installed IO backends. Xarray's read/write support requires "
140                "installing optional IO dependencies, see:\n"
141                "http://xarray.pydata.org/en/stable/getting-started-guide/installing.html\n"
142                "http://xarray.pydata.org/en/stable/user-guide/io"
143            )
144    else:
145        error_msg = (
146            "found the following matches with the input file in xarray's IO "
147            f"backends: {compatible_engines}. But their dependencies may not be installed, see:\n"
148            "http://xarray.pydata.org/en/stable/user-guide/io.html \n"
149            "http://xarray.pydata.org/en/stable/getting-started-guide/installing.html"
150        )
151
152    raise ValueError(error_msg)
153
154
155def get_backend(engine):
156    """Select open_dataset method based on current engine."""
157    if isinstance(engine, str):
158        engines = list_engines()
159        if engine not in engines:
160            raise ValueError(
161                f"unrecognized engine {engine} must be one of: {list(engines)}"
162            )
163        backend = engines[engine]
164    elif isinstance(engine, type) and issubclass(engine, BackendEntrypoint):
165        backend = engine()
166    else:
167        raise TypeError(
168            (
169                "engine must be a string or a subclass of "
170                f"xarray.backends.BackendEntrypoint: {engine}"
171            )
172        )
173
174    return backend
175