diff --git a/argopy/stores/implementations/local.py b/argopy/stores/implementations/local.py index 9bbd7b077..6cca7506a 100644 --- a/argopy/stores/implementations/local.py +++ b/argopy/stores/implementations/local.py @@ -102,7 +102,7 @@ def open_dataset( path: str The local path of the netcdf file to open - errors: + errors: str, "raise", "ignore", "silent", default = "raise" lazy: bool, default=False Define if we should try to open the netcdf dataset lazily or not @@ -137,6 +137,12 @@ def load_in_memory(path, errors="raise", xr_opts={}): raise e elif errors == "ignore": log.error("FileNotFoundError raised from: %s" % path) + except Exception as e: + if errors == "raise": + raise e + elif errors == "ignore": + log.error(f"{str(e)} raised from: {path}") + return None, None def load_lazily(path, errors="raise", xr_opts={}, akoverwrite: bool = False): diff --git a/argopy/stores/index/implementations/pyarrow/index.py b/argopy/stores/index/implementations/pyarrow/index.py index 9dfbfc245..2b25d266e 100644 --- a/argopy/stores/index/implementations/pyarrow/index.py +++ b/argopy/stores/index/implementations/pyarrow/index.py @@ -406,7 +406,7 @@ def xmax(xtble): ] else: if not hasattr(self, "index"): - self.load() + self.load(nrows=self._nrows_index) return [ xmin(self.index), xmax(self.index), @@ -424,6 +424,8 @@ def read_files(self, index: bool = False, multi: bool = False) -> List[str]: for f in self.search["file"] ] else: + if not hasattr(self, "index"): + self.load(nrows=self._nrows_index) flist = [ sep.join(["dac", f.as_py().replace("/", sep)]) for f in self.index["file"] diff --git a/argopy/utils/arco/__init__.py b/argopy/utils/arco/__init__.py new file mode 100644 index 000000000..f8c29f51a --- /dev/null +++ b/argopy/utils/arco/__init__.py @@ -0,0 +1,12 @@ +""" +This submodule is a naive approach to massively load Argo data as a large collection of points, or profiles interpolated on standard pressure levels. + +This is an experimental submodule based on previous work from the EA-RISE project. +This submodule is adapted from the original OSnet codebase developped by Sean Tokunaga and Guillaume Maze, but never published before. +https://github.com/euroargodev/argopy-osnet + +API may change anytime, not all files may be necessary in here and shall be removed/refactored without notice. +""" +from argopy.utils.arco.fetcher import MassFetcher + +__all__ = ('MassFetcher') \ No newline at end of file diff --git a/argopy/utils/arco/argo_multiprof.py b/argopy/utils/arco/argo_multiprof.py new file mode 100644 index 000000000..6f2f68733 --- /dev/null +++ b/argopy/utils/arco/argo_multiprof.py @@ -0,0 +1,243 @@ +import numpy as np +import datetime as dt +# import gsw + +from argopy.utils.arco.xarr_utils import get_use_adj_map, get_param +from argopy.utils.arco.nc_parsed import NCParsed +from argopy.utils.arco.flags import UnknownDataMode, MissingParameter, InappropriateAdjustedValues, NegativePressure, \ + UnauthorisedNan, QCCountsMismatch +from argopy.utils.arco.equality import element_wise_nan_equal +from argopy.utils.arco.errors import UnrecognisedDataSelectionMode + + +class ArgoMultiProf(NCParsed): + """A super class for an Argo xarray dataset created with a multi-profile file""" + def __init__(self, arr, data_selection_mode="adj_non_empty"): + + self.wmo = arr["PLATFORM_NUMBER"].values.astype(int) + self.institute = arr["DATA_CENTRE"].values.astype(str) + + super().__init__(arr) + self.data_selection_mode = data_selection_mode + self.temp_found = np.isin(["TEMP", "TEMP_ADJUSTED", "TEMP_QC", "TEMP_ADJUSTED_QC"], self.data_vars).all() + self.psal_found = np.isin(["PSAL", "PSAL_ADJUSTED", "PSAL_QC", "PSAL_ADJUSTED_QC"], self.data_vars).all() + + self.data_mode = np.array([0 if x == "R" else 1 if x == "A" else 2 if x == "D" else -1 for x in + arr["DATA_MODE"].values.astype(str)]) + + if self.data_selection_mode == "data_mode": + self.temp_selection_mask = (self.data_mode > 0) if self.temp_found else None + self.psal_selection_mask = (self.data_mode > 0) if self.psal_found else None + self.pres_selection_mask = (self.data_mode > 0) + + elif self.data_selection_mode == "adj_non_empty": + + self.pres_selection_mask = get_use_adj_map(adjusted_param_name="PRES_ADJUSTED", xarr=arr) + if self.temp_found: + self.temp_selection_mask = get_use_adj_map(adjusted_param_name="TEMP_ADJUSTED", xarr=arr) + else: + self.temp_selection_mask = None + if self.psal_found: + self.psal_selection_mask = get_use_adj_map(adjusted_param_name="PSAL_ADJUSTED", xarr=arr) + else: + self.psal_selection_mask = None + elif self.data_selection_mode == "real-time": + self.pres_selection_mask = np.zeros(arr["PRES"].values.shape[0]).astype(np.bool) + if self.temp_found: + self.temp_selection_mask = np.zeros(arr["TEMP"].values.shape[0]).astype(np.bool) + else: + self.temp_selection_mask = None + if self.psal_found: + self.psal_selection_mask = np.zeros(arr["PSAL"].values.shape[0]).astype(np.bool) + else: + self.psal_selection_mask = None + elif self.data_selection_mode == "delayed_mode": + self.pres_selection_mask = np.ones(arr["PRES_ADJUSTED"].values.shape[0]).astype(np.bool) + if self.temp_found: + self.temp_selection_mask = np.ones(arr["TEMP_ADJUSTED"].values.shape[0]).astype(np.bool) + else: + self.temp_selection_mask = None + if self.psal_found: + self.psal_selection_mask = np.ones(arr["PSAL_ADJUSTED"].values.shape[0]).astype(np.bool) + else: + self.psal_selection_mask = None + + else: + raise UnrecognisedDataSelectionMode(institute=self.institute, wmo=self.wmo) + + self.pres = get_param("PRES", "PRES_ADJUSTED", self.pres_selection_mask, arr).astype(np.float32) + self.pres_qc = get_param("PRES_QC", "PRES_ADJUSTED_QC", self.pres_selection_mask, arr, assign_nan=7).astype( + np.float32) + self.pres_error = get_param("PRES_ADJUSTED_ERROR", "PRES_ADJUSTED_ERROR", self.pres_selection_mask, arr).astype(np.float32) + + if self.temp_selection_mask is not None: + self.temp = get_param("TEMP", "TEMP_ADJUSTED", self.temp_selection_mask, arr).astype(np.float32) + self.temp_qc = get_param("TEMP_QC", "TEMP_ADJUSTED_QC", self.temp_selection_mask, arr, assign_nan=7).astype( + np.float32) + self.temp_error = get_param("TEMP_ADJUSTED_ERROR", "TEMP_ADJUSTED_ERROR", self.temp_selection_mask, + arr).astype(np.float32) + else: + self.temp = None + self.temp_qc = None + + if self.psal_selection_mask is not None: + self.psal = get_param("PSAL", "PSAL_ADJUSTED", self.psal_selection_mask, arr).astype(np.float32) + self.psal_qc = get_param("PSAL_QC", "PSAL_ADJUSTED_QC", self.psal_selection_mask, arr, assign_nan=7).astype( + np.float32) + self.psal_error = get_param("PSAL_ADJUSTED_ERROR", "PSAL_ADJUSTED_ERROR", self.psal_selection_mask, + arr).astype(np.float32) + else: + self.psal = None + self.psal_qc = None + + self.lon = arr["LONGITUDE"].values.astype(np.float32) + self.lat = arr["LATITUDE"].values.astype(np.float32) + self.position_qc = arr["POSITION_QC"].values.astype(np.float32) + self.position_qc[~np.isfinite(self.position_qc)] = 7 + + self.juld = arr["JULD"].values.astype(np.float32) + self.juld_qc = arr["JULD_QC"].values.astype(np.float32) + self.juld_qc[~np.isfinite(self.juld_qc)] = 7 + + # self.datetime = np.array([dt.timedelta(days=float(x)) for x in np.nan_to_num(self.juld)]) + \ + # dt.datetime(year=1950, month=1, day=1, hour=0, minute=0, second=0) + self.datetime = np.array([dt.datetime.utcfromtimestamp(x) for x in (arr['JULD'].values - np.datetime64(0, 's')) / np.timedelta64(1, 's')]) + self.month = np.array([x.month for x in self.datetime]) + self.year = np.array([x.year for x in self.datetime]) + self.cycle = arr["CYCLE_NUMBER"].values.astype(int) + self.direction = np.array([1 if x == "A" else -1 for x in arr["DIRECTION"].values.astype(str)]) + + self.psal_n = np.sum(np.isin(self.psal_qc, [0, 1, 2, 3, 4, 5, 6, 8])) + self.temp_n = np.sum(np.isin(self.temp_qc, [0, 1, 2, 3, 4, 5, 6, 8])) + self.pres_n = np.sum(np.isin(self.pres_qc, [0, 1, 2, 3, 4, 5, 6, 8])) + + # By convention, because various measurements don't match up about 30% of the time, + # Take the mean number of qcs to be 'the number of measurements'. + self.n = np.mean([self.psal_n, self.pres_n, self.temp_n]) + + self.verify(arr) # TODO Make this optional?? + + def verify(self, arr): + + # Check if parameters are missing. Some floats don't seem to contain either PSAL or TEMP. + # This is to filter those out. + # For estimating the number of missing points, we refer to pressure (always available). + psal_missing = False + if "PSAL_QC" not in self.data_vars: + psal_missing = True + self.flags.append(MissingParameter("PSAL_QC", np.sum(np.isfinite(self.pres)), len(self.pres))) + + if "PSAL_ADJUSTED_QC" not in self.data_vars: + psal_missing = True + self.flags.append(MissingParameter("PSAL_ADJUSTED_QC", np.sum(np.isfinite(self.pres)), len(self.pres))) + + if "PSAL" not in self.data_vars: + psal_missing = True + self.flags.append(MissingParameter("PSAL", np.sum(np.isfinite(self.pres)), len(self.pres))) + + if "PSAL_ADJUSTED" not in self.data_vars: + psal_missing = True + self.flags.append(MissingParameter("PSAL_ADJUSTED", np.sum(np.isfinite(self.pres)), len(self.pres))) + + temp_missing = False + if "TEMP_QC" not in self.data_vars: + temp_missing = True + self.flags.append(MissingParameter("TEMP_QC", np.sum(np.isfinite(self.pres)), len(self.pres))) + + if "TEMP_ADJUSTED_QC" not in self.data_vars: + temp_missing = True + self.flags.append(MissingParameter("TEMP_ADJUSTED_QC", np.sum(np.isfinite(self.pres)), len(self.pres))) + + if "TEMP" not in self.data_vars: + temp_missing = True + self.flags.append(MissingParameter("TEMP", np.sum(np.isfinite(self.pres)), len(self.pres))) + + if "TEMP_ADJUSTED" not in self.data_vars: + temp_missing = True + self.flags.append(MissingParameter("TEMP_ADJUSTED", np.sum(np.isfinite(self.pres)), len(self.pres))) + + if np.isin(-1, self.data_mode): + self.flags(UnknownDataMode(np.argwhere(self.data_mode == -1))) + + mask_diffs: np.ndarray = (self.pres_selection_mask != (self.data_mode > 0)) + if mask_diffs.any(): + """ + The sum of "all points in the selected profile whose values are different to those in the alt-profile" + """ + + indices = np.argwhere(mask_diffs == True) + alt_pres = get_param("PRES", "PRES_ADJUSTED", ~self.pres_selection_mask, arr).astype(np.float32) + affected_points_n = np.sum(~element_wise_nan_equal(alt_pres[indices], self.pres[indices])) + if affected_points_n: + self.flags.append(InappropriateAdjustedValues("PRES", indices, affected_points_n)) + + if not psal_missing: + mask_diffs: np.ndarray = (self.psal_selection_mask != (self.data_mode > 0)) + if mask_diffs.any(): + indices = np.argwhere(mask_diffs == True) + alt_psal = get_param("PSAL", "PSAL_ADJUSTED", ~self.psal_selection_mask, arr).astype(np.float32) + affected_points_n = np.sum(~element_wise_nan_equal(alt_psal[indices], self.psal[indices])) + if affected_points_n: + self.flags.append(InappropriateAdjustedValues("PSAL", indices, affected_points_n)) + + if not temp_missing: + mask_diffs: np.ndarray = (self.temp_selection_mask != (self.data_mode > 0)) + if mask_diffs.any(): + indices = np.argwhere(mask_diffs == True) + alt_temp = get_param("TEMP", "TEMP_ADJUSTED", ~self.temp_selection_mask, arr).astype(np.float32) + affected_points_n = np.sum(~element_wise_nan_equal(alt_temp[indices], self.temp[indices])) + if affected_points_n: + self.flags.append(InappropriateAdjustedValues("TEMP", indices, affected_points_n)) + + neg_pres = np.logical_and.reduce([np.isfinite(self.pres), (np.nan_to_num(self.pres) < 0), (self.pres_qc == 1)]) + if neg_pres.any(): + pts = np.argwhere(neg_pres) + affected_profiles_n = len(np.unique(pts[:, 0])) + self.flags.append(NegativePressure(pts, affected_profiles_n)) + + if not psal_missing: + _is_finite = np.isfinite(self.psal) + _qc_not_7 = (self.psal_qc != 7) + if (_is_finite != _qc_not_7).any(): + self.flags.append(UnauthorisedNan("PSAL", np.argwhere(_is_finite != _qc_not_7))) + + _is_finite = np.isfinite(self.pres) + _qc_not_7 = (self.pres_qc != 7) + if (_is_finite != _qc_not_7).any(): + self.flags.append(UnauthorisedNan("PRES", np.argwhere(_is_finite != _qc_not_7))) + + if not temp_missing: + _is_finite = np.isfinite(self.temp) + _qc_not_7 = (self.temp_qc != 7) + if (_is_finite != _qc_not_7).any(): + self.flags.append(UnauthorisedNan("TEMP", np.argwhere(_is_finite != _qc_not_7))) + + _is_finite = np.isfinite(self.juld) + _qc_not_7 = (self.juld_qc != 7) + if (_is_finite != _qc_not_7).any(): + self.flags.append(UnauthorisedNan("JULD", np.argwhere(_is_finite != _qc_not_7))) + + _is_finite = np.isfinite(self.lon) + _qc_not_7 = (self.position_qc != 7) + if (_is_finite != _qc_not_7).any(): + self.flags.append(UnauthorisedNan("LON", np.argwhere(_is_finite != _qc_not_7))) + + _is_finite = np.isfinite(self.lat) + _qc_not_7 = (self.position_qc != 7) + if (_is_finite != _qc_not_7).any(): + self.flags.append(UnauthorisedNan("LON", np.argwhere(_is_finite != _qc_not_7))) + + if self.temp_found and self.psal_found: + temp_psal_diffs:np.ndarray = self.psal_selection_mask != self.temp_selection_mask + if temp_psal_diffs.any(): + self.flags.append(QCCountsMismatch(len(temp_psal_diffs), np.sum([x.any() for x in temp_psal_diffs]))) + psal_pres_diffs:np.ndarray = self.psal_selection_mask != self.pres_selection_mask + if psal_pres_diffs.any(): + self.flags.append(QCCountsMismatch(len(psal_pres_diffs), np.sum([x.any() for x in psal_pres_diffs]))) + + # def get_gsw_st0(self): + # lons = np.repeat(self.lon, self.psal.shape[1]).reshape(self.psal.shape) + # lats = np.repeat(self.lat, self.psal.shape[1]).reshape(self.psal.shape) + # sa = gsw.SA_from_SP(self.psal, self.pres, lons, lats) + # ct = gsw.CT_from_t(sa, self.temp, self.pres) + # return gsw.sigma0(sa, ct) diff --git a/argopy/utils/arco/equality.py b/argopy/utils/arco/equality.py new file mode 100644 index 000000000..2444ebc1b --- /dev/null +++ b/argopy/utils/arco/equality.py @@ -0,0 +1,31 @@ +__author__ = 'sean.tokunaga@ifremer.fr' + +import numpy as np + +""" +A few helper functions for testing equality. +""" + +def element_wise_nan_equal(a, b): + """ + np.nan == np.nan returns False. I don't think we want this in our case. So I wrote an equality test where it + returns True for all equal elements including nans + :param a: numpy array + :param b: numpy array + :return: + """ + return (a == b) + np.logical_and(np.isnan(a), np.isnan(b)) + + +def nan_equal(a, b): + """ + This is equivalent to element_wise_nan_equal(a, b).all() + :param a: numpy array + :param b: numpy array + :return: + """ + try: + np.testing.assert_equal(a, b) + except AssertionError: + return False + return True diff --git a/argopy/utils/arco/errors.py b/argopy/utils/arco/errors.py new file mode 100644 index 000000000..126e8519e --- /dev/null +++ b/argopy/utils/arco/errors.py @@ -0,0 +1,35 @@ +__author__ = 'sean.tokunaga@ifremer.fr' + +""" +A bunch of custom errors used in argopy. +""" + +class NetCDF4FileNotFoundError(FileNotFoundError): + """ + Most common error. Basically just a file not found. + I made a custom one to make it easier to catch. + """ + def __init__(self, path, verbose=True): + if verbose: + print("Error: Couldn't find NetCDF4 file:") + print(path) + self.path = path + + +class UnrecognisedDataSelectionMode(ValueError): + """ + This is to be used when the user fails to specify a valid data selection mode. + the valid ones to date are ""delayed_mode", "real-time", "adj_non_empty" (default), "data_mode" + """ + def __init__(self, institute=None, wmo=None): + self.institute = institute + self.wmo = wmo + + +class UnrecognisedProfileDirection(ValueError): + """ + Not "A" or "D". Argopy should have recognized those. + """ + def __init__(self, institute=None, wmo=None): + self.institute = institute + self.wmo = wmo diff --git a/argopy/utils/arco/fetcher.py b/argopy/utils/arco/fetcher.py new file mode 100644 index 000000000..a197c55cf --- /dev/null +++ b/argopy/utils/arco/fetcher.py @@ -0,0 +1,945 @@ +import xarray as xr + +# import gsw +import pandas as pd +import numpy as np +import datetime as dt +from typing import Literal + +from argopy.options import OPTIONS, VALIDATE +from argopy.errors import OptionValueError +from argopy.stores.filesystems import distributed + +from argopy.stores.implementations.gdac import gdacfs +from argopy.utils.format import argo_split_path +from argopy.utils.arco.argo_multiprof import ArgoMultiProf + + +class MassFetcher(object): + """Fetch Argo data massively + + This class implements a direct approach to fetch Argo data massively, i.e. to fetch a very large selection of measurements, as fast as possible. In other word, this class bypasses all Argopy features to hard code all data processing steps for user modes, making data fetching effective for large selections. + + Warnings + -------- + This class is highly experimental and API can change without further notice + + Notes + ----- + This class should work as long as data fit in memory, so check out your RAM levels. + + This class use a :class:`distributed.client.Client` to scale Argo data fetching from a large collection of floats. + + Examples + -------- + ..code-block:: python + :caption: Creating a Massfetcher + + from argopy.utils.arco import MassFetcher + from argopy import ArgoIndex + + # It needs an instance of ArgoIndex search with a box: + idx = ArgoIndex().query.box([-180, 180, -90, 90, '20251001', '20251101']) + + # Create the MassFetcher: + MassFetcher(idx) + + # Possibly select user mode, otherwise use the global OPTIONS + MassFetcher(idx, mode='standard') + MassFetcher(idx, mode='research') + + ..code-block:: python + :caption: Fetch to load global Argo dataset for a given month + + from dask.distributed import Client + from argopy import ArgoIndex + from argopy.utils.arco import MassFetcher + + # Create the Dask cluster to work with: + client = Client(processes=True) + print(client) + + # Select October 2025 global data: + # This is about 8_184_913 points (12_287 prof, 3578 floats) + idx = ArgoIndex(index_file='core') + idx.query.box([-180, 180, -90, 90, '20251001', '20251101']) + + # Fetch and load data in memory: + # (~6 mins on laptop with 4 cores and 32Gb of RAM) + dsp = MassFetcher(idx).to_xarray() + + ..code-block:: python + :caption: Fetch to save global Argo dataset for a given month + + from dask.distributed import Client + from argopy import ArgoIndex + from argopy.utils.arco import MassFetcher + + # Create the Dask cluster to work with: + client = Client(processes=True) + print(client) + + # Select October 2025 global data: + # This is about 8_184_913 points (12_287 prof, 3578 floats) + idx = ArgoIndex(index_file='core') + idx.query.box([-180, 180, -90, 90, '20251001', '20251101']) + + # Fetch and save data to a zarr archive: + # (~6 mins on laptop with 4 cores and 32Gb of RAM) + zarr_archive = f"zarr/{idx.search_type['BOX'][-2]}_ARGO_STANDARD_POINTS.zarr" + MassFetcher(idx).to_zarr(zarr_archive) + + ..code-block:: python + :caption: Work with interpolated data + + from dask.distributed import Client + import numpy as np + from argopy import ArgoIndex + from argopy.utils.arco import MassFetcher + + # Create the Dask cluster to work with: + client = Client(processes=True) + print(client) + + # Select October 2025 global data: + # This is about 8_184_913 points (12_287 prof, 3578 floats) + idx = ArgoIndex(index_file='core') + idx.query.box([-180, 180, -90, 90, '20251001', '20251101']) + + # Define standard pressure levels: + sdl = np.arange(0, 2005., 5) + + # Fetch, interpolate and load data in memory: + dsp = MassFetcher(idx, sdl=sdl).to_xarray() + + # Fetch, interpolate and save data to zarr: + zarr_archive = f"zarr/{idx.search_type['BOX'][-2]}_ARGO_STANDARD_POINTS.zarr" + dsp = MassFetcher(idx, sdl=sdl).to_zarr(zarr_archive) + + ..code-block:: python + :caption: Load time series of global Argo dataset as points + + # Considering the above example where monthly data have been saved in several zarr archives, + # one can load back everything like this: + + import xarray as xr + + bigds = xr.open_mfdataset(["./zarr/20250901_ARGO_STANDARD_POINTS.zarr", + "./zarr/20251001_ARGO_STANDARD_POINTS.zarr", + "./zarr/20251101_ARGO_STANDARD_POINTS.zarr"], + combine='nested', concat_dim='N_POINTS') + bigds + + ..code-block:: python + :caption: Load time series of global Argo dataset as interpolated profiles + + # Considering the above example where monthly data have been saved in several zarr archives, + # one can load back everything like this: + + import xarray as xr + + bigds = xr.open_mfdataset(["./zarr/20250901_ARGO_STANDARD_SDLPROF.zarr", + "./zarr/20251001_ARGO_STANDARD_SDLPROF.zarr", + "./zarr/20251101_ARGO_STANDARD_SDLPROF.zarr"], + combine='nested', concat_dim='N_PROF') + bigds + + + """ + + def __init__(self, idx: "ArgoIndex", sdl=None, mode : Literal['standard', 'research'] | None = None): + """ + Parameters + ---------- + idx: :class:`ArgoIndex` + An instance of :class:`ArgoIndex` for which a 'box' query has been performed. + sdl: list, optional, default=None + Standard pressure levels to interpolate T/S profiles on. + """ + if "BOX" not in idx.search_type.keys(): + raise OptionValueError( + f"Argo data mass fetching is only available for an ArgoIndex that was search with a 'box' !" + ) + + self.mode = OPTIONS['mode'] if mode is None else mode + VALIDATE('mode', self.mode) + + self.domain = idx.domain + self.cname = idx.cname + self.box = idx.search_type["BOX"] + + # Possibly read Standard Pressure Levels: + self.sdl_axis = sdl + + # Handle other options related to file access: + self.protocol = idx.fs["src"].protocol + self.fs = gdacfs(cache=idx.cache, cachedir=idx.cachedir) + self.host = self.fs.fs.path + + # List netcdf files available for processing: + self.argo_files = idx.read_files(multi=True) + self.N_RECORDS = len(self.argo_files) + + + def __repr__(self): + summary = [f""] + summary.append(f"Host: {self.host}") + summary.append(f"Domain: {self.cname}") + summary.append(f"Number of floats: {self.N_RECORDS}") + sdl_str = ( + "-" + if str(self.sdl_axis) == "None" + else f"{len(self.sdl_axis)} levels from {min(self.sdl_axis)} to {max(self.sdl_axis)}" + ) + summary.append(f"Standard pressure levels: {sdl_str}") + return "\n".join(summary) + + def _to_sdl(self, sdl, var, var_prs, mask, lat=45.0, name="var"): + """Interpolate a 2D variable onto standard pressure levels, abusively called 'depth' levels""" + + def VI(zi, z, c): + zi, z = np.abs(zi), np.abs( + z + ) # abs ensure depths are sorted for the interpolation to work + if c.shape[0] > 0: + ci = np.interp(zi, z, c, left=c[0], right=9999.0) + if np.any(ci >= 9999.0): + return np.array(()) + else: + return ci + else: + return np.array(()) + + var_sdl = [] + ip = [] + for i in range(0, var.shape[0]): + c = var[i, mask[i, :] == True] + p = var_prs[i, mask[i, :] == True] + + # Interpolate on z: + # z = gsw.z_from_p(p, lat[i]) + # ci = VI(sdl, z, c) + + # Interpolate of p: + ci = VI(sdl, p, c) + + if ci.shape[0] > 0: + var_sdl.append(ci) + ip.append(i) + if len(var_sdl) > 0: + return xr.DataArray( + var_sdl, + dims=["N_PROF", "PRES_INTERPOLATED"], + coords={"PRES_INTERPOLATED": sdl, "N_PROF": ip}, + name=name, + ) + else: + return None + + def _add_dsattributes(self, ds, title="Argo float data", mode='standard'): + ds.attrs["title"] = title + ds.attrs["Conventions"] = "CF-1.6" + ds.attrs["CreationDate"] = pd.to_datetime("now").strftime("%Y/%m/%d") + if mode == 'standard': + ds.attrs["Comment"] = ( + "Measurements in this dataset are those with Argo QC flags " + "1, 5 or 8 on position and date; and with Argo QC flag 1 on pressure, temperature and " + "salinity. Adjusted variables were used wherever necessary " + "(depending on the data mode). See the Argo user manual for " + "more information: https://archimer.ifremer.fr/doc/00187/29825" + ) + elif mode == 'research': + ds.attrs["Comment"] = ( + "Measurements in this dataset are those with Argo QC flags " + "1, 5 or 8 on position and date; and with Argo QC flag 1 on pressure, temperature and " + "salinity. Only delayed-mode variables were used and samples with a pressure error smaller than 20db. " + "See the Argo user manual for more information: https://archimer.ifremer.fr/doc/00187/29825" + ) + + # This is following: http://cfconventions.org/Data/cf-standard-names/70/build/cf-standard-name-table.html + if "TEMP" in ds: + ds["TEMP"].attrs = { + "long_name": "Sea temperature in-situ ITS-90 scale", + "standard_name": "sea_water_temperature", + "units": "degree_Celsius", + "valid_min": -2.5, + "valid_max": 40.0, + "resolution": 0.001, + } + + if "PSAL" in ds: + ds["PSAL"].attrs = { + "long_name": "Practical salinity", + "standard_name": "sea_water_salinity", + "units": "psu", + "valid_min": 2.0, + "valid_max": 41.0, + "resolution": 0.001, + } + + if "LONGITUDE" in ds: + ds["LONGITUDE"].attrs = { + "long_name": "Longitude of the station, best estimate", + "standard_name": "longitude", + "units": "degree_east", + "valid_min": -180.0, + "valid_max": 180.0, + "resolution": 0.001, + "axis": "X", + } + + if "LATITUDE" in ds: + ds["LATITUDE"].attrs = { + "long_name": "Latitude of the station, best estimate", + "standard_name": "latitude", + "units": "degree_north", + "valid_min": -90.0, + "valid_max": 90.0, + "resolution": 0.001, + "axis": "Y", + } + + if "PRES" in ds: + ds["PRES"].attrs = { + "long_name": "Sea water pressure, equals 0 at sea-level", + "standard_name": "sea_water_pressure", + "units": "dbar", + "valid_min": 0.0, + "valid_max": 12000.0, + "resolution": 1.0, + "axis": "Z", + } + + # if 'PRES_INTERPOLATED' in ds: + # ds['PRES_INTERPOLATED'].attrs = { + # 'long_name': "Vertical distance below the surface", + # 'standard_name': "depth", + # 'units': "m", + # 'valid_min': 0., + # 'valid_max': 12000., + # 'resolution': 1., + # 'axis': 'Z', + # 'positive': 'down'} + + if "TIME" in ds: + ds["TIME"].encoding["units"] = "days since 1950-01-01" + ds["TIME"].attrs = { + "long_name": "Time", + "standard_name": "time", + "axis": "T", + } + + # Specific to this machinery: + if "ARCOID" in ds: + ds["ARCOID"].attrs = { + "long_name": "Profile unique ID", + "standard_name": "ID", + "comment": "Computed as: 1000 * FLOAT_WMO + CYCLE_NUMBER", + } + + return ds + + def _xload_multiprof_standard( + self, ds: xr.Dataset | None, domain: list + ) -> xr.Dataset | None: + """Load a core-Argo multi-profile file as a collection of points or sdl profiles in 'standard' user mode + + Parameters + ---------- + ds: :class:`xr.Dataset` + A raw loaded Xarray dataset to work with + domain: list[float | datetime] + The `box` domain to select. Contains: [lon_min, lon_max, lat_min, lat_max, date_min, date_max] + + Returns + ------- + :class:`xr.Dataset` + """ + if ds is None: + return None + + try: + argo = ArgoMultiProf(ds) + except Exception as e: + print("Error with", ds.attrs) + return None + + if argo.psal_qc is None: + return None + if argo.temp_qc is None: + return None + if (argo.juld < 0).any(): + return None + + # Profile selection + metas_finite = np.logical_and.reduce( + [np.isfinite(argo.lon), np.isfinite(argo.lat), np.isfinite(argo.juld)] + ) + # pos_good = ~(np.isin(argo.position_qc, [3, 4])) + # juld_good = ~(np.isin(argo.juld_qc, [3, 4])) + pos_good = np.isin(argo.position_qc, [1, 5, 8]) + juld_good = np.isin(argo.juld_qc, [1, 5, 8]) + in_domain = np.logical_and.reduce( + [ + argo.lon >= domain[0], + argo.lon <= domain[1], + argo.lat >= domain[2], + argo.lat <= domain[3], + argo.datetime >= dt.datetime.utcfromtimestamp(domain[4].tolist() / 1e9), + argo.datetime < dt.datetime.utcfromtimestamp(domain[5].tolist() / 1e9), + ] + ) + good_profiles = np.logical_and.reduce( + [metas_finite, juld_good, pos_good, in_domain] + ) + + if (~good_profiles).all(): + return None + + temps = argo.temp[ + good_profiles + ] # R/A/D measurements have been merged according to DATA_MODE + psals = argo.psal[ + good_profiles + ] # R/A/D measurements have been merged according to DATA_MODE + pres = argo.pres[ + good_profiles + ] # R/A/D measurements have been merged according to DATA_MODE + lons, lats, dats = ( + argo.lon[good_profiles], + argo.lat[good_profiles], + argo.datetime[good_profiles], + ) + cycs = argo.cycle[good_profiles] + + # Assign an id for each good profile + meta = argo_split_path(ds.encoding["source"]) + dac_wmo = meta["dac"], meta["wmo"] + # profile_id = np.array( + # [str(dac_wmo[1]) + "_" + x for x in np.arange(pres.shape[0]).astype(str)] + # ) + + assert temps.shape == psals.shape + assert temps.shape == pres.shape + + # per-point selection + finite_measurements = np.logical_and.reduce( + [np.isfinite(temps), np.isfinite(pres), np.isfinite(psals)] + ) + + # Exclude points with QC=3 or 4 + # temp_good = ~(np.isin(argo.temp_qc[good_profiles], [3, 4])) + # pres_good = ~(np.isin(argo.pres_qc[good_profiles], [3, 4])) + # psal_good = ~(np.isin(argo.psal_qc[good_profiles], [3, 4])) + + # Keep points with QC=1 + temp_good = np.isin(argo.temp_qc[good_profiles], [1]) + pres_good = np.isin(argo.pres_qc[good_profiles], [1]) + psal_good = np.isin(argo.psal_qc[good_profiles], [1]) + + # More sanity checks: + pres_in_range = np.nan_to_num(pres) >= 0 + temp_in_range = np.logical_and( + np.nan_to_num(temps) < 40.0, np.nan_to_num(temps) > -2.0 + ) + psal_in_range = np.logical_and( + np.nan_to_num(psals) < 41.0, np.nan_to_num(psals) > 2.0 + ) + + assert finite_measurements.shape == temps.shape + assert temp_good.shape == temps.shape + assert pres_good.shape == temps.shape + assert psal_good.shape == temps.shape + assert pres_in_range.shape == temps.shape + assert psal_in_range.shape == temps.shape + assert temp_in_range.shape == temps.shape + assert lons.shape == temps[:, 0].shape + assert lats.shape == temps[:, 0].shape + assert dats.shape == temps[:, 0].shape + assert cycs.shape == temps[:, 0].shape + + # good points? + good = np.logical_and.reduce( + [ + finite_measurements, + temp_good, + pres_good, + psal_good, + pres_in_range, + temp_in_range, + psal_in_range, + ] + ) + + assert good.shape == temps.shape + + if np.sum(good) > 0: + lons = lons[good[:, 0]] + lats = lats[good[:, 0]] + dats = dats[good[:, 0]] + cycs = cycs[good[:, 0]] + + sdl = self.sdl_axis + + if sdl is not None: + # Return data interpolated onto Standard Depth Levels: + temps = self._to_sdl(sdl, temps, pres, good, lats, name="TEMP") + psals = self._to_sdl(sdl, psals, pres, good, lats, name="PSAL") + if temps is not None and psals is not None: + assert np.all(temps["N_PROF"] == psals["N_PROF"]) == True + ip = temps["N_PROF"].values + try: + # Create dataset: + ds = xr.merge( + ( + xr.DataArray(lons[ip], name="LONGITUDE", dims="N_PROF"), + xr.DataArray(lats[ip], name="LATITUDE", dims="N_PROF"), + xr.DataArray(dats[ip], name="TIME", dims="N_PROF"), + xr.DataArray( + 1000 * argo.wmo[0] + cycs[ip], + name="id", + dims="N_PROF", + ), + temps, + psals, + ) + ) + except: + print( + "Error while merging SDL:\n", + temps["N_PROF"], + "\n", + min(ip), + max(ip), + "\n", + good[ip, 0].shape, + ) + return None + + # Preserve Argo attributes: + ds = self._add_dsattributes( + ds, + title="Argo float profiles interpolated onto Standard Pressure Levels", + mode=self.mode, + ) + ds.attrs["DAC"] = dac_wmo[0] + ds.attrs["PLATFORM_NUMBER"] = int(dac_wmo[1]) + ds.attrs["Method"] = "Vertical linear interpolation" + ds["N_PROF"].attrs = {"long_name": "Profile samples"} + return ds + else: + return None + + else: + # Return a collection of points: + pres_pts = pres[good] + temp_pts = temps[good] + psal_pts = psals[good] + + repeat_n = [np.sum(x) for x in good] + lons = np.concatenate( + [ + np.repeat(x, repeat_n[i]) + for i, x in enumerate(argo.lon[good_profiles]) + ] + ) + lats = np.concatenate( + [ + np.repeat(x, repeat_n[i]) + for i, x in enumerate(argo.lat[good_profiles]) + ] + ) + dts = np.concatenate( + [ + np.repeat(x, repeat_n[i]) + for i, x in enumerate(argo.datetime[good_profiles]) + ] + ) + # profile_id_pp = np.concatenate([np.repeat(x, repeat_n[i]) for i, x in enumerate(profile_id)]) + profile_numid = np.concatenate( + [ + np.repeat(x, repeat_n[i]) + for i, x in enumerate( + 1000 * argo.wmo[good_profiles] + argo.cycle[good_profiles] + ) + ] + ) + + ds = xr.merge( + ( + xr.DataArray(lons, name="LONGITUDE", dims="N_POINTS"), + xr.DataArray(lats, name="LATITUDE", dims="N_POINTS"), + xr.DataArray(pres_pts, name="PRES", dims="N_POINTS"), + xr.DataArray(dts, name="TIME", dims="N_POINTS"), + xr.DataArray(profile_numid, name="ARCOID", dims="N_POINTS"), + xr.DataArray(temp_pts, name="TEMP", dims="N_POINTS"), + xr.DataArray(psal_pts, name="PSAL", dims="N_POINTS"), + ) + ) + # xr.DataArray(gsw.z_from_p(pres_pts, lats, geo_strf_dyn_height=0), name='PRES_INTERPOLATED', dims='N_POINTS'), + + # Preserve Argo attributes: + ds = self._add_dsattributes( + ds, title="Argo float profiles ravelled data", mode=self.mode, + ) + ds.attrs["DAC"] = dac_wmo[0] + ds.attrs["PLATFORM_NUMBER"] = int(dac_wmo[1]) + ds["N_POINTS"].attrs = {"long_name": "Measurement samples"} + return ds + + else: + return None + + def _xload_multiprof_research( + self, ds: xr.Dataset | None, domain: list + ) -> xr.Dataset | None: + """Load a core-Argo multi-profile file as a collection of points or sdl profiles in 'research' user mode + + Parameters + ---------- + ds: :class:`xr.Dataset` + A raw loaded Xarray dataset to work with + domain: list[float | datetime] + The `box` domain to select. Contains: [lon_min, lon_max, lat_min, lat_max, date_min, date_max] + + Returns + ------- + :class:`xr.Dataset` + """ + if ds is None: + return None + + try: + argo = ArgoMultiProf(ds) + except Exception as e: + print("Error with", ds.attrs) + return None + + if argo.psal_qc is None: + return None + if argo.temp_qc is None: + return None + if (argo.juld < 0).any(): + return None + + # Profile selection + metas_finite = np.logical_and.reduce( + [np.isfinite(argo.lon), np.isfinite(argo.lat), np.isfinite(argo.juld)] + ) + pos_good = np.isin(argo.position_qc, [1, 5, 8]) + juld_good = np.isin(argo.juld_qc, [1, 5, 8]) + dm_good = np.isin(argo.data_mode, [2]) # Only delayed-mode data + in_domain = np.logical_and.reduce( + [ + argo.lon >= domain[0], + argo.lon <= domain[1], + argo.lat >= domain[2], + argo.lat <= domain[3], + argo.datetime >= dt.datetime.utcfromtimestamp(domain[4].tolist() / 1e9), + argo.datetime < dt.datetime.utcfromtimestamp(domain[5].tolist() / 1e9), + ] + ) + good_profiles = np.logical_and.reduce( + [metas_finite, juld_good, pos_good, dm_good, in_domain] + ) + + if (~good_profiles).all(): + return None + + temps = argo.temp[ + good_profiles + ] # R/A/D measurements have been merged according to DATA_MODE + psals = argo.psal[ + good_profiles + ] # R/A/D measurements have been merged according to DATA_MODE + pres = argo.pres[ + good_profiles + ] # R/A/D measurements have been merged according to DATA_MODE + + lons, lats, dats = ( + argo.lon[good_profiles], + argo.lat[good_profiles], + argo.datetime[good_profiles], + ) + cycs = argo.cycle[good_profiles] + + perrs = argo.pres_error[good_profiles] + serrs = argo.temp_error[good_profiles] + terrs = argo.psal_error[good_profiles] + + # Assign an id for each good profile + meta = argo_split_path(ds.encoding["source"]) + dac_wmo = meta["dac"], meta["wmo"] + # profile_id = np.array( + # [str(dac_wmo[1]) + "_" + x for x in np.arange(pres.shape[0]).astype(str)] + # ) + + assert temps.shape == psals.shape + assert temps.shape == pres.shape + assert temps.shape == perrs.shape + assert temps.shape == serrs.shape + assert temps.shape == terrs.shape + + # per-point selection + finite_measurements = np.logical_and.reduce( + [np.isfinite(temps), np.isfinite(pres), np.isfinite(psals)] + ) + + # Keep points with QC=1 + temp_good = np.isin(argo.temp_qc[good_profiles], [1]) + pres_good = np.isin(argo.pres_qc[good_profiles], [1]) + psal_good = np.isin(argo.psal_qc[good_profiles], [1]) + + # Keep points with pressure error smaller than 20db: + perr_in_range = argo.pres_error[good_profiles] < 20 + + # More sanity checks: + pres_in_range = np.nan_to_num(pres) >= 0 + temp_in_range = np.logical_and( + np.nan_to_num(temps) < 40.0, np.nan_to_num(temps) > -2.0 + ) + psal_in_range = np.logical_and( + np.nan_to_num(psals) < 41.0, np.nan_to_num(psals) > 2.0 + ) + + assert finite_measurements.shape == temps.shape + assert temp_good.shape == temps.shape + assert pres_good.shape == temps.shape + assert psal_good.shape == temps.shape + assert pres_in_range.shape == temps.shape + assert psal_in_range.shape == temps.shape + assert temp_in_range.shape == temps.shape + assert perr_in_range.shape == temps.shape + assert lons.shape == temps[:, 0].shape + assert lats.shape == temps[:, 0].shape + assert dats.shape == temps[:, 0].shape + assert cycs.shape == temps[:, 0].shape + + # good points? + good = np.logical_and.reduce( + [ + finite_measurements, + temp_good, + pres_good, + psal_good, + pres_in_range, + temp_in_range, + psal_in_range, + perr_in_range, + ] + ) + + assert good.shape == temps.shape + + if np.sum(good) > 0: + lons = lons[good[:, 0]] + lats = lats[good[:, 0]] + dats = dats[good[:, 0]] + cycs = cycs[good[:, 0]] + + sdl = self.sdl_axis + + if sdl is not None: + # Return data interpolated onto Standard Depth Levels: + temps = self._to_sdl(sdl, temps, pres, good, lats, name="TEMP") + psals = self._to_sdl(sdl, psals, pres, good, lats, name="PSAL") + if temps is not None and psals is not None: + assert np.all(temps["N_PROF"] == psals["N_PROF"]) == True + ip = temps["N_PROF"].values + try: + # Create dataset: + ds = xr.merge( + ( + xr.DataArray(lons[ip], name="LONGITUDE", dims="N_PROF"), + xr.DataArray(lats[ip], name="LATITUDE", dims="N_PROF"), + xr.DataArray(dats[ip], name="TIME", dims="N_PROF"), + xr.DataArray( + 1000 * argo.wmo[0] + cycs[ip], + name="id", + dims="N_PROF", + ), + temps, + psals, + ) + ) + except: + print( + "Error while merging SDL:\n", + temps["N_PROF"], + "\n", + min(ip), + max(ip), + "\n", + good[ip, 0].shape, + ) + return None + + # Preserve Argo attributes: + ds = self._add_dsattributes( + ds, + title="Argo float profiles interpolated onto Standard Pressure Levels", + mode=self.mode, + ) + ds.attrs["DAC"] = dac_wmo[0] + ds.attrs["PLATFORM_NUMBER"] = int(dac_wmo[1]) + ds.attrs["Method"] = "Vertical linear interpolation" + ds["N_PROF"].attrs = {"long_name": "Profile samples"} + return ds + else: + return None + + else: + # Return a collection of points: + pres_pts = pres[good] + temp_pts = temps[good] + psal_pts = psals[good] + + perr_pts = perrs[good] + terr_pts = terrs[good] + serr_pts = serrs[good] + + repeat_n = [np.sum(x) for x in good] + lons = np.concatenate( + [ + np.repeat(x, repeat_n[i]) + for i, x in enumerate(argo.lon[good_profiles]) + ] + ) + lats = np.concatenate( + [ + np.repeat(x, repeat_n[i]) + for i, x in enumerate(argo.lat[good_profiles]) + ] + ) + dts = np.concatenate( + [ + np.repeat(x, repeat_n[i]) + for i, x in enumerate(argo.datetime[good_profiles]) + ] + ) + # profile_id_pp = np.concatenate([np.repeat(x, repeat_n[i]) for i, x in enumerate(profile_id)]) + profile_numid = np.concatenate( + [ + np.repeat(x, repeat_n[i]) + for i, x in enumerate( + 1000 * argo.wmo[good_profiles] + argo.cycle[good_profiles] + ) + ] + ) + + ds = xr.merge( + ( + xr.DataArray(profile_numid, name="ARCOID", dims="N_POINTS"), + xr.DataArray(lons, name="LONGITUDE", dims="N_POINTS"), + xr.DataArray(lats, name="LATITUDE", dims="N_POINTS"), + xr.DataArray(dts, name="TIME", dims="N_POINTS"), + xr.DataArray(pres_pts, name="PRES", dims="N_POINTS"), + xr.DataArray(temp_pts, name="TEMP", dims="N_POINTS"), + xr.DataArray(psal_pts, name="PSAL", dims="N_POINTS"), + xr.DataArray(perr_pts, name="PRES_ERROR", dims="N_POINTS"), + xr.DataArray(terr_pts, name="TEMP_ERROR", dims="N_POINTS"), + xr.DataArray(serr_pts, name="PSAL_ERROR", dims="N_POINTS"), + ) + ) + # xr.DataArray(gsw.z_from_p(pres_pts, lats, geo_strf_dyn_height=0), name='PRES_INTERPOLATED', dims='N_POINTS'), + + # Preserve Argo attributes: + ds = self._add_dsattributes( + ds, title="Argo float profiles ravelled data", + mode=self.mode, + ) + ds.attrs["DAC"] = dac_wmo[0] + ds.attrs["PLATFORM_NUMBER"] = int(dac_wmo[1]) + ds["N_POINTS"].attrs = {"long_name": "Measurement samples"} + return ds + + else: + return None + + def to_xarray( + self, + client: distributed.client.Client | None = None, + nfloats: int | None = None, + ) -> xr.Dataset | None: + """Massively Fetch data using a :class:`distributed.client.Client` + + Parameters + ---------- + client: :class:`distributed.client.Client`, optional, default=None + A Dask client to use. If set to None, we fall back on a possibly existing client otherwise raise an `OptionValueError` if no client is found. + nfloats: int, optional, default=None + A number of floats to randomly select a subsample of the dataset for test purposes. + + Returns + ------- + xr.Dataset + + Notes + ----- + Selected Argo measurements include: + - QC=[1,5,8] on position and date + - QC=1 on pres/temp/psal + - Real-time, adjusted or delayed, according to data_mode + + """ + if client is None: + try: + client = distributed.Client.current() + except ValueError: + raise OptionValueError(f"No Distributed client running nor provided !") + + ncfiles = self.argo_files + if nfloats is not None: + ncfiles = [ + self.argo_files[f] + for f in np.random.choice(range(0, len(self.argo_files) - 1), nfloats) + ] + concat_dim = "N_PROF" if self.sdl_axis is not None else "N_POINTS" + + preprocess = self._xload_multiprof_standard if self.mode == 'standard' else self._xload_multiprof_research + + ds = self.fs.open_mfdataset( + ncfiles, + method=client, + concat=True, + concat_dim=concat_dim, + preprocess=preprocess, + preprocess_opts={"domain": self.domain}, + open_dataset_opts={"errors": "ignore"}, + errors="ignore", + ) + ds.attrs.pop("DAC") + ds.attrs.pop("PLATFORM_NUMBER") + ds.attrs["domain"] = self.cname + ds.attrs["nb_floats"] = len(np.unique((ds["ARCOID"] / 1000).astype(int))) + ds.attrs["nb_profiles"] = len(np.unique((ds["ARCOID"]))) + ds = ds.sortby("TIME") + ds[concat_dim] = np.arange(0, len(ds[concat_dim])) + return ds + + def to_zarr( + self, + store : str | None = None, + chunks : dict[str, int] | None = None, + client: distributed.client.Client | None = None, + nfloats: int | None = None, + ): + + ds = self.to_xarray(client, nfloats) + + from zarr.codecs import BloscCodec + + compressor = BloscCodec(cname="zstd", clevel=3, shuffle="shuffle") + encoding = {} + for v in ds: + encoding.update({v: {"compressors": [compressor]}}) + + if chunks is None: + if self.sdl_axis is None: + ds = ds.chunk({"N_POINTS": len(ds["N_POINTS"])}) + else: + ds = ds.chunk({"N_PROF": len(ds["N_PROF"]), "PRES_INTERPOLATED": len(ds["PRES_INTERPOLATED"])}) + else: + ds = ds.chunk(chunks) + + return ds.to_zarr(store, encoding=encoding) diff --git a/argopy/utils/arco/flags.py b/argopy/utils/arco/flags.py new file mode 100644 index 000000000..f951871a8 --- /dev/null +++ b/argopy/utils/arco/flags.py @@ -0,0 +1,151 @@ +__author__ = 'sean.tokunaga@ifremer.fr' + +import numpy as np + +""" +Flags used to identify bad datasets. This is sort of an attempt to automatically detect an log anomalies in the +data structure. They should only really get used in the verify() function. + +By convention, do not store anything that can't be read by pure python +(otherwise, it's annoying for printing, json dumping etc.) +=> Stick to ints, floats, lists, str...etc. NO PANDAS CRAP and no NDARRAYS! +""" + + +# Errors caused by something being wrong in the dataset +class BadData(object): + def __init__(self, description): + self.description = description + self.flag_id = None + +class UnknownDataMode(BadData): + def __init__(self, cycle_index): + BadData.__init__(self, "unrecognized data mode") + self.cycle_index = cycle_index.flatten().tolist() + self.affected_cycle_n = len(self.cycle_index) + self.flag_id = 0 + + +class InappropriateAdjustedValues(BadData): + def __init__(self, param, cycle_index, affected_measurement_n): + BadData.__init__(self, "adjusted values inconsistent with data mode") + self.param = param + self.cycle_index = cycle_index.flatten().tolist() + self.affected_cycle_n = len(self.cycle_index) + self.affected_measurement_n = int( + affected_measurement_n) # estimated from # points according to data_mode selection + + self.flag_id = 1 + + +class NegativePressure(BadData): + def __init__(self, points, affected_cycle_n): + BadData.__init__(self, "found negative pressures") + self.points = points.tolist() + self.affected_measurement_n = int(len(self.points)) + self.affected_cycle_n = int(affected_cycle_n) + self.flag_id = 2 + + +class MissingParameter(BadData): + def __init__(self, param, affected_measurement_n, affected_cycle_n): + BadData.__init__(self, "missing parameter") + self.param = param + self.affected_measurement_n = int(affected_measurement_n) # Estimate by counting number of finite pres points. + self.affected_cycle_n = int(affected_cycle_n) + self.flag_id = 3 + + +class BadQCCount(BadData): + def __init__(self, param, qc_count, measurement_count): + BadData.__init__(self, "qc count differs from measurement count") + self.qc_name = param + self.qc_count = int(qc_count) + self.measurement_count = int(measurement_count) + + self.flag_id = 4 + + +class QCCountsMismatch(BadData): + """ + Trigger if we don't get the same number of psal/pres/temp qcs. + CAUTION: This triggers a lot, and MAY be normal. It just means we don't have the same number of measurements, + for the 3 parameters. + """ + + def __init__(self, affeted_measurements_n, affected_cycles_n): + BadData.__init__(self, "qc counts don't match across params") + self.affected_measurement_n = int(affeted_measurements_n) # Estimate by counting number of finite pres points. + self.affected_cycle_n = int(affected_cycles_n) + + self.flag_id = 5 + + +class MeasurementCountsMismatch(BadData): + def __init__(self, temp_n, psal_n, pres_n): + BadData.__init__(self, "number of measurements don't match across params") + self.temp_n = int(temp_n) + self.pres_n = int(pres_n) + self.psal_n = int(psal_n) + + self.flag_id = 6 + + +class UnauthorisedNan(BadData): + def __init__(self, param, points): + BadData.__init__(self, "qc is nan for a measured value (or vice versa)") + self.qc_name = param + self.affected_measurement_n = int(len(points)) # Estimate by counting number of finite pres points. + self.affected_cycle_n = int(len(np.unique(points[:, 0]))) + + self.flag_id = 7 + + +############################################################################################# +# FLAGS DURING VALUE SELECTION IN ARRAYS: + + +class LevelNotFound(BadData): + def __init__(self, isas_15_alarm): + BadData.__init__(self, "Level specified in alarm wasn't found in argo data.") + self.isas_15_alarm = isas_15_alarm + self.affected_measurement_n = 1 # 1 alarm point not found + self.flag_id = 8 + + +class PressureMismatch(BadData): + def __init__(self, isas_15_alarm): + BadData.__init__(self, "Pressure specified in alarm differs by more than 0.2 from argo data.") + self.isas_15_alarm = isas_15_alarm + self.affected_measurement_n = 1 # 1 alarm point not found + self.flag_id = 9 + + +class NonUniqueProfileSpecification(BadData): + def __init__(self, isas_15_alarm): + BadData.__init__(self, "Multiple profiles found for specified Cycle_Number + Direction.") + self.isas_15_alarm = isas_15_alarm + self.affected_measurement_n = 1 # 1 alarm point not found + self.flag_id = 10 + + +class ProfileNotFound(BadData): + def __init__(self, isas_15_alarm): + BadData.__init__(self, "Cycle_Number + Direction doesn't point to any profile in data.") + self.isas_15_alarm = isas_15_alarm + self.affected_measurement_n = 1 # 1 alarm point not found + self.flag_id = 11 + + +class DuplicateAlarmsFound(BadData): + def __init__(self, isas_15_alarm): + BadData.__init__(self, "Found multiple alarms for a single measurement.") + self.isas_15_alarm = isas_15_alarm + self.affected_measurement_n = 1 # 1 alarm point not found + self.flag_id = 12 + +class InappropriateFloater(BadData): + def __init__(self, affected_alarms_n): + BadData.__init__(self, "Parameter missing (Temp or Psal).") + self.affected_measurement_n = affected_alarms_n + self.flag_id = 13 \ No newline at end of file diff --git a/argopy/utils/arco/nc_parsed.py b/argopy/utils/arco/nc_parsed.py new file mode 100644 index 000000000..b21dd97fe --- /dev/null +++ b/argopy/utils/arco/nc_parsed.py @@ -0,0 +1,15 @@ +class NCParsed(object): + """ + Abstract object for parsed .nc objects. + + The main purpose for having this base class is to be able to store flags that get raised, + as well as a standardized place for storing the variable names in a xarray file. + """ + def __init__(self, arr, meta=None): + if meta and type(meta) is dict: + self.__dict__.update(meta) + self.data_vars = list(arr.data_vars) + self.flags = [] + + def verify(self, *argv): + pass diff --git a/argopy/utils/arco/xarr_loaders.py b/argopy/utils/arco/xarr_loaders.py new file mode 100644 index 000000000..447800524 --- /dev/null +++ b/argopy/utils/arco/xarr_loaders.py @@ -0,0 +1,67 @@ +import xarray as xr +import os + +from argopy.utils.arco.errors import NetCDF4FileNotFoundError + + +class Loader(object): + """ + A generic loader class based on xarray. + If it can't find a file, it raises a specific error for easy catching. + """ + def __init__(self): + self._dac = {'KM': 'kma', + 'IF': 'coriolis', + 'AO': 'aoml', + 'CS': 'csiro', + 'KO': 'kordi', + 'JA': 'jma', + 'HZ': 'csio', + 'IN': 'incois', + 'NM': 'nmdis', + 'ME': 'meds', + 'BO': 'bodc'} + + @staticmethod + def _load_nc(file_path, verbose): + """ + Loads a .nc file using xarray, with a check for file 404s. + :param file_path: + :return: + """ + if os.path.isfile(file_path): + return xr.open_dataset(file_path, decode_times=False) + else: + raise NetCDF4FileNotFoundError(path=file_path, verbose=verbose) + + +class ArgoMultiProfLoader(Loader): + """ + Set the snapshot root path when you create the instance. + Then, it knows how to navigate the folder structure of a snapshot. + """ + def __init__(self, argo_root_path): + Loader.__init__(self) + self.argo_root_path = argo_root_path + + def load_from_inst_code(self, institute_code, wmo, verbose=True): + """ + Wrapper load function for argo. + :param institute_code: the code used to identify institutes (e.g. "IF") + :param wmo: the wmo floater ID (int) + :param verbose: prints error message + :return: the contents as an xrarray + """ + doifile = os.path.join(self.argo_root_path, self._dac[institute_code], str(wmo), ("%i_prof.nc" % wmo)) + return self._load_nc(doifile, verbose=verbose) + + def load_from_inst(self, institute, wmo, verbose=True): + """ + Wrapper load function for argo. + :param institute: the name of the institute (e.g. "coriolis") + :param wmo: the wmo floater ID (int) + :param verbose: prints error message + :return: the contents as an xrarray + """ + doifile = os.path.join(self.argo_root_path, institute, str(wmo), ("%i_prof.nc" % wmo)) + return self._load_nc(doifile, verbose) \ No newline at end of file diff --git a/argopy/utils/arco/xarr_utils.py b/argopy/utils/arco/xarr_utils.py new file mode 100644 index 000000000..4e5de5e79 --- /dev/null +++ b/argopy/utils/arco/xarr_utils.py @@ -0,0 +1,48 @@ +__author__ = 'sean.tokunaga@ifremer.fr' + +import numpy as np + + +def get_param(param_name, param_adjusted_name, adj_mask, xarr, assign_nan=None): + """ + from an xarray Dataset, extracts values as an np.ndarray for a parameter. + :param param_name: Name of parameter to extract + :param param_adjusted_name: Name of adjusted parameter to extract + :param adj_mask: 1-D boolean array of length profile_n for deciding which param to choose from (adj or not) + :param xarr: the xarray + :param assign_nan: (default=None) if specified, nans, infs and -infs are replaced with this value + :return: an np.ndarray of shape(param_name) (usually profile_n, level_n) with a mixture of param and param adjusted, + according to adj_mask + """ + vals = xarr[param_name].values # as np.ndarray + adj_vals = xarr[param_adjusted_name].values # as np.ndarray + + # detemrine which variable has the longer profile length. + # (the fact that they're not always equal is a profoundly boring mystery) + max_shape = np.max(np.array([vals.shape, adj_vals.shape]).transpose(), axis=-1) + + # Make an array full of nans with the right shape. + merged_vals = np.empty(max_shape) + merged_vals.fill(np.nan) + merged_vals[adj_mask] = adj_vals[adj_mask] # Insert values from adjusted column + merged_vals[~adj_mask] = vals[~adj_mask] # Insert values from non-adjusted column + # Replace nans with some value of choice. + if assign_nan is not None: + merged_vals[~np.isfinite(merged_vals)] = assign_nan + return merged_vals + + +def get_use_adj_map(adjusted_param_name, xarr): + """ + Checks to see if the profile contains finite values. + :param adjusted_param_name: Parameter to check + :param xarr: xarray dataset + :return: 1-D array of booleans: True if we should use PARAM_ADJUSTED else False. + length of array == number of profiles. + """ + return np.apply_along_axis(lambda x: np.isfinite(x).any(), + axis=1, + arr=xarr[adjusted_param_name].values.astype(np.float32)) + #return np.apply_along_axis(lambda x: np.logical_and(~np.isnan(x), (x != 9)).any(), + # axis=1, + # arr=xarr[adjusted_param_name].values.astype(np.float32)) \ No newline at end of file diff --git a/argopy/xarray.py b/argopy/xarray.py index 11b7f80d2..3cefb2a09 100644 --- a/argopy/xarray.py +++ b/argopy/xarray.py @@ -117,6 +117,7 @@ def __init__(self, xarray_obj): self.encoding = xarray_obj.encoding self.attrs = xarray_obj.attrs + self._shape = {"N_POINTS": None, "N_PROF": None, "N_LEVELS": None} if "N_PROF" in self._dims: self._type = "profile" elif "N_POINTS" in self._dims: @@ -163,68 +164,99 @@ def __repr__(self): @property def N_PROF(self): """Number of profiles""" - if self._type == "point": - if self.N_POINTS > 0: - dummy_argo_uid = xr.DataArray( - self.uid( - self._obj["PLATFORM_NUMBER"].values, - self._obj["CYCLE_NUMBER"].values, - self._obj["DIRECTION"].values, - ), - dims="N_POINTS", - coords={"N_POINTS": self._obj["N_POINTS"]}, - name="dummy_argo_uid", - ) - N_PROF = len(np.unique(dummy_argo_uid)) + if self._shape["N_PROF"] is None: + if self._type == "point": + if self.N_POINTS > 0: + if "ARCOID" in self._obj: + self._shape["N_PROF"] = len( + np.unique((self._obj["ARCOID"] / 1000).astype(int)) + ) + else: + dummy_argo_uid = xr.DataArray( + self.uid( + self._obj["PLATFORM_NUMBER"].values, + self._obj["CYCLE_NUMBER"].values, + self._obj["DIRECTION"].values, + ), + dims="N_POINTS", + coords={"N_POINTS": self._obj["N_POINTS"]}, + name="dummy_argo_uid", + ) + self._shape["N_PROF"] = len(np.unique(dummy_argo_uid)) + else: + self._shape["N_PROF"] = 0.0 else: - N_PROF = 0. - else: - N_PROF = len(np.unique(self._obj["N_PROF"])) - return N_PROF + self._shape["N_PROF"] = len(np.unique(self._obj["N_PROF"])) + return self._shape["N_PROF"] @property def N_LEVELS(self): """Number of vertical levels""" - if self._type == "point": - if self.N_POINTS > 0: - dummy_argo_uid = xr.DataArray( - self.uid( - self._obj["PLATFORM_NUMBER"].values, - self._obj["CYCLE_NUMBER"].values, - self._obj["DIRECTION"].values, - ), - dims="N_POINTS", - coords={"N_POINTS": self._obj["N_POINTS"]}, - name="dummy_argo_uid", - ) - N_LEVELS = int( - xr.DataArray( - np.ones_like(self._obj["N_POINTS"].values), - dims="N_POINTS", - coords={"N_POINTS": self._obj["N_POINTS"]}, - ) - .groupby(dummy_argo_uid) - .sum() - .max() - .values - ) + if self._shape["N_LEVELS"] is None: + if self._type == "point": + if self.N_POINTS > 0: + if "ARCOID" in self._obj: + this = xr.DataArray( + np.ones_like(self._obj["N_POINTS"].values), + dims="N_POINTS", + coords={"N_POINTS": self._obj["N_POINTS"]}, + name="counter", + ).to_dataset() + this["ARCOID"] = self._obj["ARCOID"] + N_LEVELS = ( + this.groupby( + ARCOID=xr.groupers.UniqueGrouper( + labels=np.unique(this["ARCOID"]) + ) + ) + .sum() + .max()["counter"] + .values + ) + self._shape["N_LEVELS"] = int(N_LEVELS[np.newaxis][0]) + + else: + dummy_argo_uid = xr.DataArray( + self.uid( + self._obj["PLATFORM_NUMBER"].values, + self._obj["CYCLE_NUMBER"].values, + self._obj["DIRECTION"].values, + ), + dims="N_POINTS", + coords={"N_POINTS": self._obj["N_POINTS"]}, + name="dummy_argo_uid", + ) + self._shape["N_LEVELS"] = int( + xr.DataArray( + np.ones_like(self._obj["N_POINTS"].values), + dims="N_POINTS", + coords={"N_POINTS": self._obj["N_POINTS"]}, + ) + .groupby(dummy_argo_uid) + .sum() + .max() + .values + ) + else: + self._shape["N_LEVELS"] = 0.0 else: - N_LEVELS = 0. - else: - N_LEVELS = len(np.unique(self._obj["N_LEVELS"])) - return N_LEVELS + self._shape["N_LEVELS"] = len(np.unique(self._obj["N_LEVELS"])) + + return self._shape["N_LEVELS"] @property def N_POINTS(self): """Number of measurement points""" - if self._type == "profile": - if self.N_PROF > 0: - N_POINTS = self.N_PROF * self.N_LEVELS + if self._shape["N_POINTS"] is None: + if self._type == "profile": + if self.N_PROF > 0: + self._shape["N_POINTS"] = self.N_PROF * self.N_LEVELS + else: + self._shape["N_POINTS"] = 0.0 else: - N_POINTS = 0. - else: - N_POINTS = len(np.unique(self._obj["N_POINTS"])) - return N_POINTS + self._shape["N_POINTS"] = len(np.unique(self._obj["N_POINTS"])) + + return self._shape["N_POINTS"] def add_history(self, txt): if "Processing_history" in self._obj.attrs: @@ -277,16 +309,19 @@ def _TNAME(self): def _dummy_argo_uid(self): if self._type == "point": if self.N_POINTS > 0: - return xr.DataArray( - self.uid( - self._obj["PLATFORM_NUMBER"].values, - self._obj["CYCLE_NUMBER"].values, - self._obj["DIRECTION"].values, - ), - dims="N_POINTS", - coords={"N_POINTS": self._obj["N_POINTS"]}, - name="dummy_argo_uid", - ) + if "ARCOID" in self._obj: + return self._obj["ARCOID"] + else: + return xr.DataArray( + self.uid( + self._obj["PLATFORM_NUMBER"].values, + self._obj["CYCLE_NUMBER"].values, + self._obj["DIRECTION"].values, + ), + dims="N_POINTS", + coords={"N_POINTS": self._obj["N_POINTS"]}, + name="dummy_argo_uid", + ) else: raise NoData("Cannot create UID for an empty dataset") else: @@ -517,7 +552,9 @@ def fillvalue(da): i_uid, prof = grp for iv, vname in enumerate(this.data_vars): try: - count[i_prof, iv] = len(np.unique(prof[vname])) # This is very long because it must read all the data ! + count[i_prof, iv] = len( + np.unique(prof[vname]) + ) # This is very long because it must read all the data ! except Exception: log.error( "point2profile: An error happened when dealing with the '%s' data variable" @@ -525,9 +562,9 @@ def fillvalue(da): ) raise - # Variables with a unique value for each profiles: + # Variables with a unique value for each profile: list_1d = list(np.array(this.data_vars)[count.sum(axis=0) == count.shape[0]]) - # Variables with more than 1 value for each profiles: + # Variables with more than 1 value for each profile: list_2d = list(np.array(this.data_vars)[count.sum(axis=0) != count.shape[0]]) # Create new empty dataset: @@ -629,7 +666,9 @@ def profile2point(self) -> xr.Dataset: "Method only available for a collection of profiles (N_PROF dimension)" ) ds = self._obj - ds = ds.argo.datamode.split() # Otherwise this method will fail with BGC netcdf files + ds = ( + ds.argo.datamode.split() + ) # Otherwise this method will fail with BGC netcdf files # Remove all variables for which a dimension is length=0 (eg: N_HISTORY) # todo: We should be able to find a way to keep them somewhere in the data structure @@ -916,7 +955,9 @@ def filter_researchmode(self) -> xr.Dataset: core_params.remove("PSAL") # Apply transforms and filters: - this = this.argo.filter_qc(QC_list=1, QC_fields=["POSITION_QC", f"{self._TNAME}_QC"]) + this = this.argo.filter_qc( + QC_list=1, QC_fields=["POSITION_QC", f"{self._TNAME}_QC"] + ) this = this.argo.datamode.merge(params=core_params) this = this.argo.datamode.filter(params=core_params, dm="D") @@ -1053,7 +1094,7 @@ def interp_std_levels( for co in coords: ds_out.coords[co] = this_dsp[co] - ds_out = ds_out.drop_vars(["N_LEVELS", "Z_LEVELS"], errors='ignore') + ds_out = ds_out.drop_vars(["N_LEVELS", "Z_LEVELS"], errors="ignore") ds_out = ds_out[np.sort(ds_out.data_vars)] ds_out = ds_out.argo.cast_types() ds_out.attrs = self.attrs # Preserve original attributes @@ -1482,9 +1523,9 @@ def mid(x): if "SIG0" in vlist: SIG0 = xr.DataArray(sig0, coords=this["TEMP"].coords, name="SIG0") - SIG0.attrs[ - "long_name" - ] = "Potential density anomaly with reference pressure of 0 dbar" + SIG0.attrs["long_name"] = ( + "Potential density anomaly with reference pressure of 0 dbar" + ) SIG0.attrs["standard_name"] = "sea_water_sigma_theta" SIG0.attrs["unit"] = "kg/m^3" that.append(SIG0) @@ -1523,7 +1564,7 @@ def mid(x): that.append(CS) # Create a dataset with all new variables: - that = xr.merge(that, compat='no_conflicts') + that = xr.merge(that, compat="no_conflicts") # Add to the dataset essential Argo variables (allows to keep using the argo accessor): that = that.assign( { @@ -1811,7 +1852,10 @@ def preprocess_one_float( # Compute fractional year: # https://github.com/euroargodev/dm_floats/blob/c580b15202facaa0848ebe109103abe508d0dd5b/src/ow_source/create_float_source.m#L334 DATES = np.array( - [toYearFraction(d) for d in pd.to_datetime(this_one[self._TNAME].values)] + [ + toYearFraction(d) + for d in pd.to_datetime(this_one[self._TNAME].values) + ] )[np.newaxis, :] # Read measurements: @@ -2009,11 +2053,12 @@ def to_zarr(self, *args, **kwargs) -> Union[ZarrStore, Delayed]: # Add zarr compression to encoding: if "encoding" not in kwargs: from numcodecs import Blosc + compressor = Blosc(cname="zstd", clevel=3, shuffle=2) encoding = {} for v in self._obj: encoding.update({v: {"compressor": compressor}}) - kwargs.update({'encoding': encoding}) + kwargs.update({"encoding": encoding}) # Convert to a zarr file using compression: return self._obj.to_zarr(*args, **kwargs) @@ -2105,8 +2150,10 @@ def max_salinity_depth(pres, psal): for param in plist: if param not in self._obj: raise ValueError(f"Parameter {param} not in dataset") - if 'N_LEVELS' not in self._obj[param].dims: - raise ValueError(f"Parameter {param} must have the 'N_LEVELS' dimension") + if "N_LEVELS" not in self._obj[param].dims: + raise ValueError( + f"Parameter {param} must have the 'N_LEVELS' dimension" + ) # There should be one input core dimension 'N_LEVELS' for each argument of the reduce function input_core_dims = [["N_LEVELS"] for _ in plist] @@ -2119,11 +2166,9 @@ def max_salinity_depth(pres, psal): ufunc_kwargs = dict( kwargs=kwargs, # Keywords arguments to be passed to the reduce function input_core_dims=input_core_dims, - # dimensions allowed to change size. Must be set! # must also appear in ``input_core_dims`` for at least one argument exclude_dims=set(("N_LEVELS",)), - vectorize=True, # loop over non-core dims dask="parallelized", ) @@ -2134,13 +2179,23 @@ def max_salinity_depth(pres, psal): ) return reduced - def map_vars_to_dict(self, var_key: str, var_val:str, duplicate:bool=False) -> dict[Any, Any]: - return map_vars_to_dict(self._obj, var_key=var_key, var_val=var_val, duplicate=duplicate) + def map_vars_to_dict( + self, var_key: str, var_val: str, duplicate: bool = False + ) -> dict[Any, Any]: + return map_vars_to_dict( + self._obj, var_key=var_key, var_val=var_val, duplicate=duplicate + ) def open_Argo_dataset(filename_or_obj): time_coder = xr.coders.CFDatetimeCoder(use_cftime=False) - ds = xr.open_dataset(filename_or_obj, decode_cf=1, decode_times=time_coder, mask_and_scale=1, decode_timedelta=True) + ds = xr.open_dataset( + filename_or_obj, + decode_cf=1, + decode_times=time_coder, + mask_and_scale=1, + decode_timedelta=True, + ) ds = cast_Argo_variable_type(ds) return ds diff --git a/ci/requirements/py3.11-all-free.yml b/ci/requirements/py3.11-all-free.yml index 468d795d6..d0b4320b1 100644 --- a/ci/requirements/py3.11-all-free.yml +++ b/ci/requirements/py3.11-all-free.yml @@ -32,6 +32,7 @@ dependencies: # EXT.PERF: - dask - distributed + - flox - joblib - numba - pyarrow diff --git a/ci/requirements/py3.11-all-pinned.yml b/ci/requirements/py3.11-all-pinned.yml index ef465ce01..30a697c1f 100644 --- a/ci/requirements/py3.11-all-pinned.yml +++ b/ci/requirements/py3.11-all-pinned.yml @@ -32,6 +32,7 @@ dependencies: # EXT.PERF: - dask = 2025.5.1 - distributed = 2025.5.1 + - flox = 0.10.8 - joblib = 1.5.2 - numba = 0.62.1 - pyarrow = 20.0.0 diff --git a/ci/requirements/py3.11-core-free.yml b/ci/requirements/py3.11-core-free.yml index c9b1e2828..749aac03d 100644 --- a/ci/requirements/py3.11-core-free.yml +++ b/ci/requirements/py3.11-core-free.yml @@ -32,6 +32,7 @@ dependencies: # EXT.PERF: # - dask # - distributed +# - flox # - joblib # - numba # - pyarrow diff --git a/ci/requirements/py3.11-core-pinned.yml b/ci/requirements/py3.11-core-pinned.yml index 7efbdf793..815f0b23d 100644 --- a/ci/requirements/py3.11-core-pinned.yml +++ b/ci/requirements/py3.11-core-pinned.yml @@ -32,6 +32,7 @@ dependencies: # EXT.PERF: # - dask = 2025.5.1 # - distributed = 2025.5.1 +# - flox = 0.10.8 # - joblib = 1.5.2 # - numba = 0.62.1 # - pyarrow = 20.0.0 diff --git a/ci/requirements/py3.12-all-free.yml b/ci/requirements/py3.12-all-free.yml index 0859b8066..67cc26e84 100644 --- a/ci/requirements/py3.12-all-free.yml +++ b/ci/requirements/py3.12-all-free.yml @@ -32,6 +32,7 @@ dependencies: # EXT.PERF: - dask - distributed + - flox - joblib - numba - pyarrow diff --git a/ci/requirements/py3.12-all-pinned.yml b/ci/requirements/py3.12-all-pinned.yml index 545f98389..c207f400a 100644 --- a/ci/requirements/py3.12-all-pinned.yml +++ b/ci/requirements/py3.12-all-pinned.yml @@ -32,6 +32,7 @@ dependencies: # EXT.PERF: - dask = 2025.5.1 - distributed = 2025.5.1 + - flox = 0.10.8 - joblib = 1.5.2 - numba = 0.62.1 - pyarrow = 20.0.0 diff --git a/ci/requirements/py3.12-core-free.yml b/ci/requirements/py3.12-core-free.yml index 3c38cf35c..0155cfffa 100644 --- a/ci/requirements/py3.12-core-free.yml +++ b/ci/requirements/py3.12-core-free.yml @@ -32,6 +32,7 @@ dependencies: # EXT.PERF: # - dask # - distributed +# - flox # - joblib # - numba # - pyarrow diff --git a/ci/requirements/py3.12-core-pinned.yml b/ci/requirements/py3.12-core-pinned.yml index 8bb357715..014ae9150 100644 --- a/ci/requirements/py3.12-core-pinned.yml +++ b/ci/requirements/py3.12-core-pinned.yml @@ -32,6 +32,7 @@ dependencies: # EXT.PERF: # - dask = - # - distributed = - +# - flox = - # - joblib = - # - numba = - # - pyarrow = -