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